mirror of
https://github.com/R0m1k3/Loki.git
synced 2026-10-21 22:18:13 +02:00
v0.8.2 : audit du coeur, correctifs de securite et de robustesse
- Le mode agent redevient une vraie fermeture : une surcharge de requete ne peut plus RALLUMER l'agent (donc bash/write/edit/MCP) quand la machine l'a coupe. - Une lecture ratee de la cle de pilotage ne se confond plus avec « aucune cle » : requireWebAuth ferme (503) au lieu d'ouvrir l'API. - healthCheck : timeout de 3 s. Sans lui, un moteur qui accepte la connexion sans repondre laissait /api/chat/send bloque indefiniment. - Acces distant : la session de tunnel ecoute enfin un contexte. Un redemarrage ne fait plus cohabiter deux sessions ; le backoff se remet a zero apres une session qui a tenu. - Flux SSE : selection des evenements neufs par dichotomie au lieu de relire tout le journal a chaque token et pour chaque appareil. - Cache de lecture de la configuration, invalide par ecriture locale, par mtime et par age : fin des ~100 ouvertures de base par tour agentique. - Erreurs moteur classees par type (errors.Is / net.Error) et non par sous-chaine anglaise ; les 3 compactions partagent un seul chemin ; champs morts de chatReq retires ; ReadHeaderTimeout sur les deux serveurs HTTP.
This commit is contained in:
1 parent
bc270ac15c
commit
e6a948c314
17 files changed
+499
-168
No files matched your search
@@ -89,8 +89,10 @@ func cmdSetAPIKey(args []string) error {
|
||||
return serviceAction("restart")
|
||||
}
|
||||
|
||||
// ReadConfig renvoie la configuration active de llama-server.
|
||||
func ReadConfig() map[string]string { return allKV(bkConfig) }
|
||||
// ReadConfig renvoie la configuration active de llama-server. Passe par le cache
|
||||
// de lecture (store.go) : elle est relue à chaque itération de la boucle
|
||||
// d'inférence et après chaque appel d'outil.
|
||||
func ReadConfig() map[string]string { return cachedKV(bkConfig) }
|
||||
|
||||
// SetConfigKey définit une clé de configuration. Une valeur vide la supprime.
|
||||
func SetConfigKey(key, value string) error { return putStr(bkConfig, key, value) }
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -198,6 +199,46 @@ func (c *Conversation) compactLogLocked() {
|
||||
c.Log = out
|
||||
}
|
||||
|
||||
// compactAndPublish exécute UNE compaction et en publie tout le cycle de vie :
|
||||
// bannière de progression, journal, installation du résultat, jauge de contexte.
|
||||
// Renvoie l'historique (compacté ou inchangé) et s'il a changé.
|
||||
//
|
||||
// Les trois compactions (début de tour, fin de tour, bouton manuel) déroulaient
|
||||
// la même douzaine de lignes recopiées, y compris le calcul du surcoût fixe :
|
||||
// une correction dans l'une ne suivait pas dans les deux autres.
|
||||
//
|
||||
// Le surcoût, justement : le contexte RÉEL mesuré moins l'estimation des
|
||||
// messages donne ce que l'estimation ne voit pas (prompt système injecté,
|
||||
// schémas d'outils, gabarit de chat). On le rajoute à l'estimation d'après
|
||||
// compaction, sinon la jauge s'effondre puis resaute au tour suivant.
|
||||
func (c *Conversation) compactAndPublish(ctx context.Context, epoch int, phase string, msgs []Message, ctxUsed int, caps Caps) ([]Message, bool) {
|
||||
c.appendDelta(epoch, map[string]any{"compacting": true})
|
||||
compacted, changed := compactMessages(ctx, msgs, caps)
|
||||
c.appendDelta(epoch, map[string]any{"compacting": false})
|
||||
logCompact(phase, ctxUsed, msgs, compacted, changed)
|
||||
if !changed {
|
||||
// Seuil franchi mais compaction sans effet (torse vide, ou réduction sous
|
||||
// le minimum exigé) : on le DIT. Retirer la bannière sans un mot donnait,
|
||||
// vu de l'UI, « le compactage automatique ne fait rien ».
|
||||
c.appendDelta(epoch, map[string]any{"compact_noop": true})
|
||||
return msgs, false
|
||||
}
|
||||
overhead := ctxUsed - estimateTokens(msgs)
|
||||
if overhead < 0 {
|
||||
overhead = 0
|
||||
}
|
||||
est := estimateTokens(compacted) + overhead
|
||||
c.mu.Lock()
|
||||
if c.epoch == epoch {
|
||||
c.Messages = compacted
|
||||
c.CtxUsed = est // le vrai compte reviendra avec les stats du prochain tour
|
||||
}
|
||||
c.mu.Unlock()
|
||||
c.appendDelta(epoch, map[string]any{"compacted": true})
|
||||
c.appendDelta(epoch, map[string]any{"ctx_used": est}) // fait chuter la jauge tout de suite
|
||||
return compacted, true
|
||||
}
|
||||
|
||||
// convState renvoie un instantané léger (pour /api/chat/state).
|
||||
func (c *Conversation) state() map[string]any {
|
||||
c.mu.Lock()
|
||||
@@ -272,46 +313,13 @@ func (c *Conversation) generate(ctx context.Context, caps Caps, temperature floa
|
||||
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.
|
||||
// Si le contexte dépasse le seuil, le résumé (un appel modèle non streamé) va
|
||||
// bloquer plusieurs secondes AVANT que la vraie réponse commence. On affiche
|
||||
// une bannière « compactage en cours » pendant ce temps (sinon l'UI se fige
|
||||
// sans aucune info), puis on la retire — que le compactage ait changé quelque
|
||||
// chose ou non.
|
||||
willCompact := compactWouldTrigger(msgs, ctxUsed)
|
||||
if willCompact {
|
||||
c.appendDelta(epoch, map[string]any{"compacting": true})
|
||||
}
|
||||
compacted, changed := MaybeCompact(ctx, msgs, caps, ctxUsed)
|
||||
if willCompact {
|
||||
logCompact("début-tour", ctxUsed, msgs, compacted, changed)
|
||||
}
|
||||
if changed {
|
||||
// Surcoût fixe non compactable (prompt système injecté + schémas d'outils +
|
||||
// gabarit) = contexte réel mesuré − estimation des messages. On le rajoute à
|
||||
// l'estimation post-compaction pour que la jauge affiche une valeur réaliste
|
||||
// (sinon elle chute trop bas puis resaute au tour suivant).
|
||||
overhead := ctxUsed - estimateTokens(msgs)
|
||||
if overhead < 0 {
|
||||
overhead = 0
|
||||
// d'affichage garde le fil complet. Le résumé est un appel modèle non streamé :
|
||||
// il bloque plusieurs secondes AVANT que la vraie réponse commence, d'où la
|
||||
// bannière de progression émise par compactAndPublish.
|
||||
if compactWouldTrigger(msgs, ctxUsed) {
|
||||
if out, changed := c.compactAndPublish(ctx, epoch, "début-tour", msgs, ctxUsed, caps); changed {
|
||||
msgs = out
|
||||
}
|
||||
est := estimateTokens(compacted) + overhead
|
||||
msgs = compacted
|
||||
c.mu.Lock()
|
||||
if c.epoch == epoch {
|
||||
c.Messages = compacted
|
||||
c.CtxUsed = est // le vrai compte reviendra avec les stats du tour
|
||||
}
|
||||
c.mu.Unlock()
|
||||
c.appendDelta(epoch, map[string]any{"compacting": false})
|
||||
c.appendDelta(epoch, map[string]any{"compacted": true})
|
||||
c.appendDelta(epoch, map[string]any{"ctx_used": est}) // fait chuter la jauge tout de suite
|
||||
} else if willCompact {
|
||||
// Seuil franchi mais compaction sans effet (torse vide, ou réduction sous les
|
||||
// 20% exigés) : on le DIT. Avant, on retirait juste la bannière sans un mot —
|
||||
// vu de l'UI, « le compactage automatique ne fait rien » sans aucune trace.
|
||||
c.appendDelta(epoch, map[string]any{"compacting": false})
|
||||
c.appendDelta(epoch, map[string]any{"compact_noop": true})
|
||||
}
|
||||
|
||||
// Prompt système personnalisé (UI → /api/sysprompt, fichier côté serveur).
|
||||
@@ -414,27 +422,9 @@ func (c *Conversation) generate(ctx context.Context, caps Caps, temperature floa
|
||||
if stale || ctx.Err() != nil || !compactWouldTrigger(msgs, ctxUsed) {
|
||||
return
|
||||
}
|
||||
c.appendDelta(epoch, map[string]any{"compacting": true})
|
||||
compacted, changed = compactMessages(context.Background(), msgs, caps)
|
||||
c.appendDelta(epoch, map[string]any{"compacting": false})
|
||||
logCompact("fin-tour", ctxUsed, msgs, compacted, changed)
|
||||
if !changed {
|
||||
c.appendDelta(epoch, map[string]any{"compact_noop": true})
|
||||
return
|
||||
}
|
||||
overhead := ctxUsed - estimateTokens(msgs)
|
||||
if overhead < 0 {
|
||||
overhead = 0
|
||||
}
|
||||
est := estimateTokens(compacted) + overhead
|
||||
c.mu.Lock()
|
||||
if c.epoch == epoch {
|
||||
c.Messages = compacted
|
||||
c.CtxUsed = est
|
||||
}
|
||||
c.mu.Unlock()
|
||||
c.appendDelta(epoch, map[string]any{"compacted": true})
|
||||
c.appendDelta(epoch, map[string]any{"ctx_used": est})
|
||||
// context.Background() et non ctx : le tour est terminé, son contexte peut
|
||||
// être annulé alors que cette compaction-là doit aller au bout.
|
||||
c.compactAndPublish(context.Background(), epoch, "fin-tour", msgs, ctxUsed, caps)
|
||||
}
|
||||
|
||||
// CompactNow force une compaction du contexte MAINTENANT, sans attendre le seuil
|
||||
@@ -466,28 +456,9 @@ func (c *Conversation) CompactNow() error {
|
||||
c.cancel = nil
|
||||
c.mu.Unlock()
|
||||
}()
|
||||
c.appendDelta(epoch, map[string]any{"compacting": true})
|
||||
compacted, changed := compactMessages(ctx, msgs, Caps{})
|
||||
c.appendDelta(epoch, map[string]any{"compacting": false})
|
||||
logCompact("manuel", lastReal, msgs, compacted, changed)
|
||||
if !changed {
|
||||
c.appendDelta(epoch, map[string]any{"compact_noop": true})
|
||||
return
|
||||
if _, changed := c.compactAndPublish(ctx, epoch, "manuel", msgs, lastReal, Caps{}); changed {
|
||||
c.persist()
|
||||
}
|
||||
overhead := lastReal - estimateTokens(msgs)
|
||||
if overhead < 0 {
|
||||
overhead = 0
|
||||
}
|
||||
est := estimateTokens(compacted) + overhead // + surcoût fixe (système+outils) pour une jauge réaliste
|
||||
c.mu.Lock()
|
||||
if c.epoch == epoch {
|
||||
c.Messages = compacted
|
||||
c.CtxUsed = est // feedback immédiat ; le vrai compte revient au prochain tour
|
||||
}
|
||||
c.mu.Unlock()
|
||||
c.appendDelta(epoch, map[string]any{"compacted": true})
|
||||
c.appendDelta(epoch, map[string]any{"ctx_used": est})
|
||||
c.persist()
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
@@ -727,11 +698,16 @@ func (c *Conversation) Subscribe(ctx context.Context, from int, emit func(map[st
|
||||
// Copie des événements en attente SOUS verrou, émission HORS verrou : on
|
||||
// n'itère jamais sur c.Log pendant que la génération peut y écrire ou que
|
||||
// la troncature (maxLogEvents) peut le déplacer.
|
||||
//
|
||||
// c.Log est trié par Seq croissant : on trouve le premier événement neuf
|
||||
// par dichotomie plutôt qu'en relisant tout. Le balayage complet coûtait
|
||||
// la longueur du journal (jusqu'à 20 000) À CHAQUE TOKEN et pour CHAQUE
|
||||
// appareil connecté, verrou tenu — donc au détriment de la génération
|
||||
// elle-même. C'est quadratique sur un long fil.
|
||||
i := sort.Search(len(c.Log), func(i int) bool { return c.Log[i].Seq > last })
|
||||
var pending []LogEvent
|
||||
for _, ev := range c.Log {
|
||||
if ev.Seq > last {
|
||||
pending = append(pending, ev)
|
||||
}
|
||||
if i < len(c.Log) {
|
||||
pending = append(pending, c.Log[i:]...)
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
c.cond.Wait()
|
||||
|
||||
@@ -178,3 +178,50 @@ func TestReplayMarksReplaceWhenClientSawPartOfBlock(t *testing.T) {
|
||||
t.Fatalf("replace ne doit PAS être posé quand le client n'a rien vu du bloc")
|
||||
}
|
||||
}
|
||||
|
||||
// Le direct sélectionne les événements neufs par dichotomie (c.Log est trié par
|
||||
// Seq). Deux cas limites doivent rester justes : un journal TRONQUÉ, dont le
|
||||
// premier Seq est très supérieur au `from` du client, et un client déjà à jour.
|
||||
func TestLiveSelectionSurJournalTronque(t *testing.T) {
|
||||
c := newTestConv()
|
||||
// Journal amputé de son début, comme après la troncature à maxLogEvents.
|
||||
c.Log = []LogEvent{
|
||||
{Seq: 500, Delta: map[string]any{"user": "a"}},
|
||||
{Seq: 501, Delta: map[string]any{"user": "b"}},
|
||||
{Seq: 502, Delta: map[string]any{"user": "c"}},
|
||||
}
|
||||
c.Seq = 502
|
||||
|
||||
for _, tc := range []struct {
|
||||
from int
|
||||
first int // premier seq attendu
|
||||
}{
|
||||
{from: 0, first: 500}, // client neuf : il reçoit tout ce qui reste
|
||||
{from: 500, first: 501}, // au milieu du journal
|
||||
{from: 501, first: 502},
|
||||
} {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
got := make(chan int, 8)
|
||||
go c.Subscribe(ctx, tc.from, func(m map[string]any) bool {
|
||||
if s, ok := m["seq"].(int); ok {
|
||||
got <- s
|
||||
}
|
||||
return true
|
||||
})
|
||||
waitSeq(t, got, tc.first)
|
||||
cancel()
|
||||
}
|
||||
|
||||
// Client déjà à jour : rien à rejouer, mais le direct doit suivre.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
got := make(chan int, 8)
|
||||
go c.Subscribe(ctx, 502, func(m map[string]any) bool {
|
||||
if s, ok := m["seq"].(int); ok {
|
||||
got <- s
|
||||
}
|
||||
return true
|
||||
})
|
||||
c.appendDelta(c.epoch, map[string]any{"content": "suite"})
|
||||
waitSeq(t, got, 503)
|
||||
}
|
||||
@@ -8,10 +8,13 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
)
|
||||
|
||||
@@ -472,24 +475,55 @@ const thinkClose = "</think>"
|
||||
// « actively refused » (Windows) — arrive quand le moteur redémarre ou charge
|
||||
// encore le modèle ; l'erreur brute (« dial tcp 127.0.0.1:8080… ») ne dit rien à
|
||||
// l'utilisateur. On garde le silence sur une annulation volontaire (/stop).
|
||||
// On classe d'abord sur les erreurs TYPÉES (errors.Is / net.Error) : la
|
||||
// reconnaissance par sous-chaîne dépend de la langue et du format des messages
|
||||
// de l'OS, et « connexion refusée » d'un Windows français ne ressemble à aucun
|
||||
// des motifs anglais. Les sous-chaînes restent en second rideau, pour les
|
||||
// erreurs enveloppées par une bibliothèque qui perd le type d'origine.
|
||||
func friendlyLLMError(err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return err // /stop : pas d'alarme
|
||||
}
|
||||
switch {
|
||||
case errors.Is(err, syscall.ECONNREFUSED):
|
||||
return errEngineDown()
|
||||
case errors.Is(err, context.DeadlineExceeded), isNetTimeout(err):
|
||||
return fmt.Errorf("⚠️ Le moteur (llama-server) met trop de temps à répondre (port %d) — il est peut-être surchargé ou en plein chargement. Réessaie dans un instant.", LLMPort())
|
||||
case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF), errors.Is(err, syscall.ECONNRESET):
|
||||
return errEngineReset()
|
||||
}
|
||||
low := strings.ToLower(err.Error())
|
||||
switch {
|
||||
case strings.Contains(low, "context canceled"):
|
||||
return err // /stop : pas d'alarme
|
||||
return err
|
||||
case strings.Contains(low, "connection refused"), strings.Contains(low, "actively refused"), strings.Contains(low, "connectex"), strings.Contains(low, "no connection could be made"):
|
||||
return fmt.Errorf("⚠️ Le moteur (llama-server) ne répond pas sur le port %d. Il est probablement en train de démarrer ou de charger le modèle — réessaie dans quelques secondes.", LLMPort())
|
||||
return errEngineDown()
|
||||
case strings.Contains(low, "timeout"), strings.Contains(low, "deadline exceeded"):
|
||||
return fmt.Errorf("⚠️ Le moteur (llama-server) met trop de temps à répondre (port %d) — il est peut-être surchargé ou en plein chargement. Réessaie dans un instant.", LLMPort())
|
||||
case strings.Contains(low, "eof"), strings.Contains(low, "connection reset"):
|
||||
return fmt.Errorf("⚠️ Connexion au moteur (llama-server, port %d) interrompue — il a peut-être redémarré. Réessaie.", LLMPort())
|
||||
return errEngineReset()
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func errEngineDown() error {
|
||||
return fmt.Errorf("⚠️ Le moteur (llama-server) ne répond pas sur le port %d. Il est probablement en train de démarrer ou de charger le modèle — réessaie dans quelques secondes.", LLMPort())
|
||||
}
|
||||
|
||||
func errEngineReset() error {
|
||||
return fmt.Errorf("⚠️ Connexion au moteur (llama-server, port %d) interrompue — il a peut-être redémarré. Réessaie.", LLMPort())
|
||||
}
|
||||
|
||||
// isNetTimeout : une erreur réseau qui se déclare elle-même comme un délai
|
||||
// dépassé (net.Error.Timeout), quel que soit son libellé.
|
||||
func isNetTimeout(err error) bool {
|
||||
var ne net.Error
|
||||
return errors.As(err, &ne) && ne.Timeout()
|
||||
}
|
||||
|
||||
func runChat(ctx context.Context, messages []Message, temperature float64, caps Caps, cb ChatCallback) ([]Message, error) {
|
||||
var extra []Message
|
||||
tools := EnabledTools(caps)
|
||||
@@ -575,8 +609,11 @@ func runChat(ctx context.Context, messages []Message, temperature float64, caps
|
||||
if compactEnabled() && !compactedRetry {
|
||||
if c, changed := compactMessages(ctx, messages, caps); changed {
|
||||
compactedRetry = true
|
||||
messages = c
|
||||
// ⚠️ Journaliser AVANT d'installer le résultat : l'ancien ordre
|
||||
// passait `messages` déjà remplacé comme état « avant », donc la
|
||||
// ligne comparait le résultat à lui-même et n'apprenait rien.
|
||||
logCompact("réactif", 0, messages, c, changed)
|
||||
messages = c
|
||||
// Même publication qu'en cours de tour : sans elle, la compaction de
|
||||
// secours ne survit pas à la fin du tour et le prompt re-déborde au
|
||||
// message suivant.
|
||||
@@ -1023,9 +1060,17 @@ func runChat(ctx context.Context, messages []Message, temperature float64, caps
|
||||
}
|
||||
}
|
||||
|
||||
// healthClient : /health doit répondre tout de suite ou pas du tout. Sans
|
||||
// timeout (http.Get et son client par défaut n'en ont aucun), un moteur qui
|
||||
// accepte la connexion sans jamais répondre — cas classique d'un très gros
|
||||
// modèle en cours de chargement, ou d'un process figé — bloquait healthCheck
|
||||
// indéfiniment. Et comme StartTurn commence par là, /api/chat/send restait
|
||||
// pendu : l'utilisateur voyait un bouton d'envoi qui ne rendait jamais la main.
|
||||
var healthClient = &http.Client{Timeout: 3 * time.Second}
|
||||
|
||||
// healthCheck pings llama.cpp's /health endpoint.
|
||||
func healthCheck() bool {
|
||||
resp, err := http.Get(fmt.Sprintf("http://localhost:%d/health", LLMPort()))
|
||||
resp, err := healthClient.Get(fmt.Sprintf("http://localhost:%d/health", LLMPort()))
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -214,7 +214,8 @@ func linkPrintIdentity() error {
|
||||
var appLink struct {
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
stop chan struct{}
|
||||
stop context.CancelFunc
|
||||
done chan struct{} // fermé quand la boucle a vraiment rendu la main
|
||||
}
|
||||
|
||||
// appOwnsLink : vrai dans le process de l'app de bureau, qui pilote le tunnel
|
||||
@@ -238,21 +239,25 @@ func startAppLink(mux *http.ServeMux) {
|
||||
if token == "" {
|
||||
return
|
||||
}
|
||||
stop := make(chan struct{})
|
||||
appLink.running, appLink.stop = true, stop
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
appLink.running, appLink.stop, appLink.done = true, cancel, done
|
||||
handler := newLinkHandler(mux)
|
||||
oaiTLS := oaiTLSConfig()
|
||||
go func() {
|
||||
defer close(done)
|
||||
backoff := time.Second
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
for ctx.Err() == nil {
|
||||
started := time.Now()
|
||||
_ = runLinkSession(ctx, token, handler, oaiTLS)
|
||||
// Une session qui a TENU repart de zéro. Sans cette remise, le délai
|
||||
// grimpait de coupure en coupure et restait collé à 30 s pour toujours,
|
||||
// y compris pour reconnecter un lien qui venait de vivre des heures.
|
||||
if time.Since(started) > time.Minute {
|
||||
backoff = time.Second
|
||||
}
|
||||
_ = runLinkSession(token, handler, oaiTLS)
|
||||
select {
|
||||
case <-stop:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
@@ -266,15 +271,25 @@ func startAppLink(mux *http.ServeMux) {
|
||||
}()
|
||||
}
|
||||
|
||||
// stopAppLink arrête la boucle interne (la session en cours se termine d'elle-même).
|
||||
// stopAppLink arrête la boucle interne ET attend qu'elle ait rendu la main.
|
||||
// L'attente est ce qui garantit qu'un redémarrage ne fait pas cohabiter deux
|
||||
// sessions : sans elle, le nouveau tunnel s'ouvrait pendant que l'ancien vivait
|
||||
// encore, et le relais voyait deux agents pour la même machine.
|
||||
func stopAppLink() {
|
||||
appLink.mu.Lock()
|
||||
defer appLink.mu.Unlock()
|
||||
if !appLink.running {
|
||||
appLink.mu.Unlock()
|
||||
return
|
||||
}
|
||||
close(appLink.stop)
|
||||
appLink.running = false
|
||||
cancel, done := appLink.stop, appLink.done
|
||||
appLink.running, appLink.stop, appLink.done = false, nil, nil
|
||||
appLink.mu.Unlock()
|
||||
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(5 * time.Second): // filet : on ne bloque pas l'UI indéfiniment
|
||||
}
|
||||
}
|
||||
|
||||
func appLinkRunning() bool {
|
||||
@@ -519,8 +534,14 @@ func withLocalAuth(next http.Handler) http.Handler {
|
||||
// runLinkSession opens one WebSocket→yamux session and serves it until it dies.
|
||||
// It blocks for the lifetime of the connection and returns the error that ended
|
||||
// it (so the caller can reconnect).
|
||||
func runLinkSession(token string, handler http.Handler, oaiTLS *tls.Config) error {
|
||||
ctx := context.Background()
|
||||
// runLinkSession tient UNE session de tunnel, jusqu'à ce qu'elle meure ou que
|
||||
// ctx soit annulé. Le contexte est indispensable : sans lui, un « arrêter le
|
||||
// lien » ne prenait effet qu'à la mort naturelle du WebSocket, qui n'arrive
|
||||
// jamais tant que le relais répond. Un redémarrage (stop puis start enchaînés)
|
||||
// lançait donc une seconde session pendant que la première vivait encore : deux
|
||||
// agents connectés au relais pour la même machine, une seule conversation
|
||||
// derrière.
|
||||
func runLinkSession(ctx context.Context, token string, handler http.Handler, oaiTLS *tls.Config) error {
|
||||
dialCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -569,7 +590,10 @@ func runLinkSession(token string, handler http.Handler, oaiTLS *tls.Config) erro
|
||||
// va au HTTP (comportement historique inchangé).
|
||||
httpLn := newChanListener(sess.Addr())
|
||||
defer httpLn.Close()
|
||||
srv := &http.Server{Handler: handler}
|
||||
// Même garde-fou que le serveur local : un stream du tunnel qui n'envoie
|
||||
// jamais d'en-têtes ne doit pas retenir une goroutine indéfiniment. Pas de
|
||||
// WriteTimeout : le chat E2E est un flux SSE de longue durée.
|
||||
srv := &http.Server{Handler: handler, ReadHeaderTimeout: 30 * time.Second}
|
||||
go srv.Serve(httpLn)
|
||||
|
||||
var oaiLn *chanListener
|
||||
@@ -579,9 +603,19 @@ func runLinkSession(token string, handler http.Handler, oaiTLS *tls.Config) erro
|
||||
go runOAIFront(oaiLn, oaiTLS)
|
||||
}
|
||||
|
||||
// Annulation : fermer la session yamux débloque l'Accept ci-dessous, qui est
|
||||
// le seul point où cette fonction attend. La goroutine meurt avec la session.
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
_ = sess.Close()
|
||||
}()
|
||||
|
||||
for {
|
||||
stream, err := sess.Accept()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err() // arrêt demandé, pas une panne
|
||||
}
|
||||
return fmt.Errorf("accept: %w", err)
|
||||
}
|
||||
go demuxTunnelStream(stream, httpLn, oaiLn)
|
||||
|
||||
@@ -10,7 +10,7 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
const Version = "0.8.1"
|
||||
const Version = "0.8.2"
|
||||
|
||||
// Main est le vrai main() du binaire (cmd/ajean ne fait que l'appeler).
|
||||
func Main() {
|
||||
|
||||
+88
-5
@@ -86,6 +86,7 @@ func view(bucket string, fn func(b *bolt.Bucket) error) error {
|
||||
|
||||
// update exécute une écriture, en créant le bucket au besoin.
|
||||
func update(bucket string, fn func(b *bolt.Bucket) error) error {
|
||||
cacheBust(bucket)
|
||||
return withDB(func(d *bolt.DB) error {
|
||||
return d.Update(func(tx *bolt.Tx) error {
|
||||
b, err := tx.CreateBucketIfNotExists([]byte(bucket))
|
||||
@@ -97,17 +98,28 @@ func update(bucket string, fn func(b *bolt.Bucket) error) error {
|
||||
})
|
||||
}
|
||||
|
||||
// getBytes lit une valeur. Une base inaccessible se comporte comme une base
|
||||
// vide : les appelants sont des lecteurs de réglages, aucun n'a de recours utile
|
||||
// face à une erreur d'E/S, et tous ont déjà un défaut.
|
||||
func getBytes(bucket, key string) []byte {
|
||||
// getBytesErr lit une valeur en REMONTANT l'erreur d'accès. À réserver aux
|
||||
// lecteurs pour qui « je n'ai pas pu lire » et « il n'y a rien » ne veulent pas
|
||||
// dire la même chose — au premier chef les secrets : sans clé enregistrée, l'API
|
||||
// de pilotage est ouverte, donc une lecture ratée traitée comme « pas de clé »
|
||||
// ouvrirait l'API au lieu de la fermer. Voir readWebKeyErr.
|
||||
func getBytesErr(bucket, key string) ([]byte, error) {
|
||||
var out []byte
|
||||
_ = view(bucket, func(b *bolt.Bucket) error {
|
||||
err := view(bucket, func(b *bolt.Bucket) error {
|
||||
if v := b.Get([]byte(key)); v != nil {
|
||||
out = append([]byte(nil), v...) // la valeur ne survit pas à la transaction
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
|
||||
// getBytes lit une valeur. Une base inaccessible se comporte comme une base
|
||||
// vide : les appelants sont des lecteurs de réglages, aucun n'a de recours utile
|
||||
// face à une erreur d'E/S, et tous ont déjà un défaut. Les lecteurs pour qui
|
||||
// l'erreur CHANGE la décision prennent getBytesErr.
|
||||
func getBytes(bucket, key string) []byte {
|
||||
out, _ := getBytesErr(bucket, key)
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -159,6 +171,76 @@ func putJSON(bucket, key string, v any) error {
|
||||
return putBytes(bucket, key, b)
|
||||
}
|
||||
|
||||
// --- Cache de lecture ---------------------------------------------------------
|
||||
//
|
||||
// La base est rouverte à CHAQUE opération (choix délibéré, voir plus haut), ce
|
||||
// qui est parfait pour un réglage lu de temps en temps mais coûteux dans le
|
||||
// chemin chaud : la boucle d'inférence relit le port, la clé, le mode
|
||||
// raisonnement et le seuil de compactage à chaque itération, et le compactage
|
||||
// se re-teste après chaque appel d'outil. Un tour agentique un peu fourni
|
||||
// rouvrait la base une centaine de fois.
|
||||
//
|
||||
// cachedKV garde donc le contenu d'un bucket en mémoire, invalidé par :
|
||||
// - une écriture de CE process (cacheBust, appelé par update/replaceKV) ;
|
||||
// - un changement de taille ou de date du fichier, qui trahit l'écriture d'un
|
||||
// AUTRE process (la CLI pendant que le service tourne) ;
|
||||
// - l'âge, plafonné à une seconde, filet pour le cas limite où deux écritures
|
||||
// rapprochées laisseraient date et taille inchangées.
|
||||
//
|
||||
// Il ne sert PAS aux secrets : eux se lisent directement (voir getBytesErr).
|
||||
const cacheMaxAge = time.Second
|
||||
|
||||
type kvCache struct {
|
||||
kv map[string]string
|
||||
when time.Time
|
||||
mtime time.Time
|
||||
size int64
|
||||
}
|
||||
|
||||
var (
|
||||
cacheMu sync.Mutex
|
||||
caches = map[string]kvCache{}
|
||||
)
|
||||
|
||||
// cacheBust vide le cache d'un bucket après une écriture locale.
|
||||
func cacheBust(bucket string) {
|
||||
cacheMu.Lock()
|
||||
delete(caches, bucket)
|
||||
cacheMu.Unlock()
|
||||
}
|
||||
|
||||
// dbStamp renvoie la date et la taille du fichier de base — de quoi repérer
|
||||
// l'écriture d'un autre process pour le prix d'un stat.
|
||||
func dbStamp() (time.Time, int64) {
|
||||
fi, err := os.Stat(dbPath())
|
||||
if err != nil {
|
||||
return time.Time{}, -1
|
||||
}
|
||||
return fi.ModTime(), fi.Size()
|
||||
}
|
||||
|
||||
// cachedKV renvoie tout le contenu d'un bucket, depuis le cache quand il est
|
||||
// encore valable. La carte renvoyée appartient à l'appelant (copie).
|
||||
func cachedKV(bucket string) map[string]string {
|
||||
mtime, size := dbStamp()
|
||||
cacheMu.Lock()
|
||||
c, ok := caches[bucket]
|
||||
fresh := ok && c.size == size && c.mtime.Equal(mtime) && time.Since(c.when) < cacheMaxAge
|
||||
cacheMu.Unlock()
|
||||
if !fresh {
|
||||
kv := allKV(bucket)
|
||||
c = kvCache{kv: kv, when: time.Now(), mtime: mtime, size: size}
|
||||
cacheMu.Lock()
|
||||
caches[bucket] = c
|
||||
cacheMu.Unlock()
|
||||
}
|
||||
out := make(map[string]string, len(c.kv))
|
||||
for k, v := range c.kv {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// allKV renvoie tout le contenu d'un bucket. Utilisé par la configuration, dont
|
||||
// les clés ne sont pas connues à l'avance (EXTRA_ARGS et consorts).
|
||||
func allKV(bucket string) map[string]string {
|
||||
@@ -176,6 +258,7 @@ func allKV(bucket string) map[string]string {
|
||||
// C'est ce qu'exige l'application d'un preset : à aucun instant la config ne
|
||||
// doit être un mélange de l'ancienne et de la nouvelle.
|
||||
func replaceKV(bucket string, m map[string]string) error {
|
||||
cacheBust(bucket)
|
||||
return withDB(func(d *bolt.DB) error {
|
||||
return d.Update(func(tx *bolt.Tx) error {
|
||||
if err := tx.DeleteBucket([]byte(bucket)); err != nil && !errors.Is(err, bolterrors.ErrBucketNotFound) {
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
package ajean
|
||||
|
||||
import "testing"
|
||||
|
||||
// Le cache de lecture ne doit jamais servir une valeur périmée après une
|
||||
// écriture de ce process : un « ajean edit » suivi d'un « ajean restart » doit
|
||||
// démarrer sur la NOUVELLE configuration.
|
||||
func TestCacheConfigInvalideParEcriture(t *testing.T) {
|
||||
testHome(t)
|
||||
if err := WriteConfig(map[string]string{"CTX": "4096"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := ReadConfig()["CTX"]; got != "4096" { // remplit le cache
|
||||
t.Fatalf("CTX = %q, attendu 4096", got)
|
||||
}
|
||||
if err := SetConfigKey("CTX", "8192"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := ReadConfig()["CTX"]; got != "8192" {
|
||||
t.Fatalf("valeur périmée servie par le cache : CTX = %q, attendu 8192", got)
|
||||
}
|
||||
if err := WriteConfig(map[string]string{"MODEL": "m.gguf"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := ReadConfig()["CTX"]; ok {
|
||||
t.Error("WriteConfig remplace tout : CTX ne doit plus exister")
|
||||
}
|
||||
}
|
||||
|
||||
// La carte renvoyée appartient à l'appelant : la modifier ne doit pas polluer
|
||||
// le cache, sans quoi un appelant distrait corromprait la configuration vue par
|
||||
// tous les autres.
|
||||
func TestCacheConfigRenvoieUneCopie(t *testing.T) {
|
||||
testHome(t)
|
||||
if err := WriteConfig(map[string]string{"CTX": "4096"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg := ReadConfig()
|
||||
cfg["CTX"] = "999"
|
||||
delete(cfg, "CTX")
|
||||
if got := ReadConfig()["CTX"]; got != "4096" {
|
||||
t.Fatalf("le cache a été altéré par l'appelant : CTX = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Une lecture ratée ne doit pas se confondre avec « aucune clé » : sans clé,
|
||||
// l'API de pilotage est OUVERTE, donc l'erreur doit remonter pour que
|
||||
// requireWebAuth ferme au lieu d'ouvrir.
|
||||
func TestWebKeyDistingueVideEtErreur(t *testing.T) {
|
||||
testHome(t)
|
||||
k, err := readWebKeyErr()
|
||||
if err != nil || k != "" {
|
||||
t.Fatalf("base neuve : attendu (\"\", nil), reçu (%q, %v)", k, err)
|
||||
}
|
||||
if err := putStr(bkState, "web_key", "secret"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if k, err := readWebKeyErr(); err != nil || k != "secret" {
|
||||
t.Fatalf("attendu (\"secret\", nil), reçu (%q, %v)", k, err)
|
||||
}
|
||||
}
|
||||
+15
-18
@@ -899,29 +899,26 @@ func handleBenchLast(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// chatReq est le corps d'une requête de chat (commun au chat clair et au chat E2E).
|
||||
//
|
||||
// `messages` et `ctx_used`, que les clients envoyaient du temps où l'historique
|
||||
// vivait dans le navigateur, ont été RETIRÉS : depuis que la conversation est
|
||||
// possédée par le serveur (chat_conversation.go), plus personne ne les lisait.
|
||||
// Les garder laissait croire qu'un client pouvait encore imposer son historique
|
||||
// ou son décompte de contexte. Un client qui les envoie encore n'est pas gêné :
|
||||
// le décodeur JSON ignore les champs qu'il ne connaît pas.
|
||||
type chatReq struct {
|
||||
Messages []Message `json:"messages"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
// Optional per-request override of the agent mode (used by ajean.link
|
||||
// agents, which carry their own toggle). nil = inherit the machine's
|
||||
// global config. Tools/Skills sont conservés pour la rétro-compat des
|
||||
// anciens clients relais : l'un OU l'autre à true active le mode agent.
|
||||
Temperature float64 `json:"temperature"`
|
||||
// Surcharges par requête, portées par les agents ajean.link qui ont leurs
|
||||
// propres interrupteurs. nil = on prend la configuration de la machine.
|
||||
// ⚠️ Elles ne peuvent que RESTREINDRE, jamais rallumer (voir capsFromBody).
|
||||
// Tools/Skills : rétro-compat des anciens clients relais.
|
||||
Agent *bool `json:"agent"`
|
||||
Tools *bool `json:"tools"`
|
||||
Skills *bool `json:"skills"`
|
||||
// Override par requête de l'accès internet (outils web). nil = config machine.
|
||||
// Surcharge par requête de l'accès internet (outils web).
|
||||
Internet *bool `json:"internet"`
|
||||
// Taille réelle du contexte au tour précédent (usage.prompt_tokens + tokens
|
||||
// générés), rapportée par le client qui l'affiche déjà. Sert à décider du
|
||||
// compactage sur le VRAI décompte plutôt qu'une estimation. 0 = inconnu.
|
||||
CtxUsed int `json:"ctx_used"`
|
||||
// Nouveau modèle « conversation serveur » : Message = texte du tour à lancer
|
||||
// (via /api/chat/send) ; From = dernier Seq déjà vu par le client (le flux
|
||||
// d'abonnement rejoue Log[From:] puis suit le direct).
|
||||
// Message = texte du tour à lancer (/api/chat/send) ; From = dernier Seq déjà
|
||||
// vu par le client (le flux d'abonnement rejoue Log[From:] puis suit le direct).
|
||||
Message string `json:"message"`
|
||||
From int `json:"from"`
|
||||
}
|
||||
|
||||
// capsFromBody dérive les capacités du tour à partir des overrides éventuels du
|
||||
// corps de requête (agents ajean.link portant leurs propres toggles), sinon la
|
||||
// config machine.
|
||||
@@ -19,15 +19,33 @@ import (
|
||||
// et inversement. Elle est relue à chaque requête (pas de cache) pour qu'un
|
||||
// changement de clé prenne effet sans redémarrer le serveur web.
|
||||
|
||||
// readWebKey renvoie la clé de pilotage, ou "" si aucune n'est définie.
|
||||
func readWebKey() string { return getStr(bkState, "web_key") }
|
||||
// readWebKeyErr renvoie la clé de pilotage en distinguant « aucune clé » d'une
|
||||
// LECTURE RATÉE. La nuance est tout sauf cosmétique : sans clé, l'API est
|
||||
// ouverte. Confondre les deux, c'est ouvrir l'API parce que la base était
|
||||
// momentanément verrouillée par une commande CLI — une panne d'E/S qui désarme
|
||||
// l'authentification. requireWebAuth refuse donc plutôt que d'ouvrir.
|
||||
func readWebKeyErr() (string, error) {
|
||||
b, err := getBytesErr(bkState, "web_key")
|
||||
return string(b), err
|
||||
}
|
||||
|
||||
// readWebKey renvoie la clé de pilotage, ou "" si aucune n'est définie (ou
|
||||
// illisible). Réservé à l'affichage ; toute décision d'accès passe par
|
||||
// readWebKeyErr.
|
||||
func readWebKey() string { k, _ := readWebKeyErr(); return k }
|
||||
|
||||
// requireWebAuth wraps an HTTP handler, rejecting requests that don't present
|
||||
// the configured Bearer token. When no key is configured the handler is left
|
||||
// open (pratique en local) — cmdWeb avertit alors bruyamment au démarrage.
|
||||
func requireWebAuth(next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
key := readWebKey()
|
||||
key, err := readWebKeyErr()
|
||||
if err != nil {
|
||||
// On ne sait pas si une clé protège cette API : on ferme.
|
||||
sendJSON(w, http.StatusServiceUnavailable,
|
||||
map[string]any{"error": "configuration illisible — réessaie dans un instant"})
|
||||
return
|
||||
}
|
||||
if key == "" {
|
||||
next(w, r)
|
||||
return
|
||||
|
||||
@@ -11,15 +11,30 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// capsFromBody dérive les capacités d'un tour à partir de la configuration
|
||||
// machine et des surcharges portées par la requête.
|
||||
//
|
||||
// ⚠️ Une surcharge ne peut que RESTREINDRE. Elle pouvait auparavant rallumer le
|
||||
// mode agent : un simple {"agent":true} redonnait bash, write, edit et les
|
||||
// outils MCP alors que l'interrupteur de la machine était sur OFF. Comme l'API
|
||||
// n'est pas protégée par défaut et écoute sur 0.0.0.0, l'interrupteur ne
|
||||
// garantissait donc rien. Il redevient une vraie fermeture : ce qui est éteint
|
||||
// sur la machine ne peut pas être rallumé par un client.
|
||||
func capsFromBody(body chatReq) Caps {
|
||||
caps := globalCaps()
|
||||
if body.Agent != nil {
|
||||
caps.Agent = *body.Agent
|
||||
} else if body.Tools != nil || body.Skills != nil {
|
||||
caps.Agent = (body.Tools != nil && *body.Tools) || (body.Skills != nil && *body.Skills)
|
||||
switch {
|
||||
case body.Agent != nil:
|
||||
caps.Agent = caps.Agent && *body.Agent
|
||||
case body.Tools != nil || body.Skills != nil:
|
||||
want := (body.Tools != nil && *body.Tools) || (body.Skills != nil && *body.Skills)
|
||||
caps.Agent = caps.Agent && want
|
||||
}
|
||||
if body.Internet != nil {
|
||||
caps.Internet = *body.Internet && crawlReachable()
|
||||
caps.Internet = caps.Internet && *body.Internet
|
||||
}
|
||||
// Les outils dépendent du mode agent : agent coupé, tout est coupé.
|
||||
if !caps.Agent {
|
||||
caps.Internet = false
|
||||
}
|
||||
return caps
|
||||
}
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
package ajean
|
||||
|
||||
import "testing"
|
||||
|
||||
func ptrBool(b bool) *bool { return &b }
|
||||
|
||||
// Une surcharge portée par la requête ne doit JAMAIS rallumer ce que la machine
|
||||
// a éteint : sans ça, {"agent":true} redonnait bash/write/edit (et les outils
|
||||
// MCP) alors que l'interrupteur agent était sur OFF, sur une API qui n'est pas
|
||||
// protégée par défaut.
|
||||
func TestCapsFromBodyNePeutPasRallumerLAgent(t *testing.T) {
|
||||
testHome(t)
|
||||
if err := setAgentEnabled(false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, body := range []chatReq{
|
||||
{Agent: ptrBool(true)},
|
||||
{Tools: ptrBool(true)},
|
||||
{Skills: ptrBool(true)},
|
||||
{Internet: ptrBool(true)},
|
||||
} {
|
||||
if caps := capsFromBody(body); caps.Agent || caps.Internet {
|
||||
t.Errorf("agent coupé sur la machine, mais %+v a donné %+v", body, caps)
|
||||
}
|
||||
}
|
||||
// Et les outils ne doivent pas non plus être servis au modèle.
|
||||
if tools := EnabledTools(capsFromBody(chatReq{Agent: ptrBool(true)})); len(tools) > 0 {
|
||||
t.Errorf("agent coupé : aucun outil ne devrait être proposé, %d le sont", len(tools))
|
||||
}
|
||||
}
|
||||
|
||||
// L'inverse reste vrai : une surcharge peut RESTREINDRE un agent actif.
|
||||
func TestCapsFromBodyPeutRestreindre(t *testing.T) {
|
||||
testHome(t)
|
||||
if err := setAgentEnabled(true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if caps := capsFromBody(chatReq{Agent: ptrBool(false)}); caps.Agent {
|
||||
t.Error("agent:false doit couper l'agent pour ce tour")
|
||||
}
|
||||
if caps := capsFromBody(chatReq{}); !caps.Agent {
|
||||
t.Error("sans surcharge, on doit hériter de la configuration machine")
|
||||
}
|
||||
}
|
||||
@@ -69,7 +69,12 @@ func cmdWeb(args []string) error {
|
||||
fmt.Printf("%s front OpenAI public prêt (activation en direct via l'UI ; état: %v)\n", green("[oai]"), oaiPublicEnabled())
|
||||
startAppLink(mux)
|
||||
}
|
||||
return http.Serve(ln, mux)
|
||||
// ReadHeaderTimeout : sans lui, une connexion qui n'envoie jamais sa requête
|
||||
// immobilise une goroutine pour toujours — et ce port écoute sur 0.0.0.0.
|
||||
// Surtout PAS de WriteTimeout ici : il couperait les flux SSE du chat, qui
|
||||
// restent ouverts aussi longtemps que l'utilisateur regarde la page.
|
||||
srv := &http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second}
|
||||
return srv.Serve(ln)
|
||||
}
|
||||
|
||||
// newWebMux construit le routeur HTTP de l'UI web. Extrait de cmdWeb pour être
|
||||
|
||||
Reference in new issue
Block a user