diff --git a/internal/loki/chat_compact.go b/internal/loki/chat_compact.go index 9b55a86..29a8893 100644 --- a/internal/loki/chat_compact.go +++ b/internal/loki/chat_compact.go @@ -242,6 +242,42 @@ func compactNeeded(msgs []Message, used, peakGen int) bool { return compactEnabled() && genHeadroomShort(used, peakGen, ctxWindow()) } +// compactRetryDue : une compaction en cours de tour peut-elle être tentée ? +// Après un essai qui n'a rien réduit (résumé raté, réduction sous 20 %), +// chaque étape suivante relançait un résumé complet voué au même échec : plus +// d'une heure de tentatives à vide, une par étape, sur une longue tâche (AJEAN +// 68659e73). compactRefusedSkip ne couvre que la discussion principale avec +// COMPACT_CONTINUATION ; ce garde-fou vaut pour tout tour, tâches et +// sous-agents compris. Jamais à 90 % de la fenêtre ou plus : là, essayer +// encore vaut mieux que déborder. +func compactRetryDue(retryAt, used int, msgs []Message) bool { + if retryAt <= 0 { + return true + } + est := compactUsedOrEstimate(used, msgs) + return est >= retryAt || est*10 >= ctxWindow()*9 +} + +// compactNextRetry : seuil du prochain essai après une compaction en cours de +// tour. 0 (aucun) si elle a réduit ; sinon le contexte d'aujourd'hui plus 5 % +// de la fenêtre. +func compactNextRetry(changed bool, used int, msgs []Message) int { + if changed { + return 0 + } + est := compactUsedOrEstimate(used, msgs) + next := est + ctxWindow()/20 + fmt.Fprintf(os.Stderr, "[compact] rien réduit en cours de tour : prochain essai à %d jetons (contexte à %d)\n", next, est) + return next +} + +func compactUsedOrEstimate(used int, msgs []Message) int { + if used > 0 { + return used + } + return estimateTokens(msgs) +} + // genHeadroomShort : la fenêtre ne laisse plus la place d'une génération de // 1,2 × la plus longue récente (plancher de 8 k jetons, ramené à 1/8 de la // fenêtre pour les petites), plus une petite marge. Fonction pure, testable. @@ -437,18 +473,24 @@ func compactMessagesCore(ctx context.Context, msgs []Message, caps Caps, opt com var index []recallEntry if caps.Agent && compactEnabled() { names := torsoToolNames(torso) + var blocks []recallArchiveInput + var at []int // position dans le torse de chaque bloc for i, m := range torso { if !recallEligible(m) { continue } content := msgText(m) - label := recallLabel(m.Role, names[m.ToolCallID], content) - id, aerr := archiveRecallBlock(label, m.Role, content) - if aerr != nil { - continue + blocks = append(blocks, recallArchiveInput{Label: recallLabel(m.Role, names[m.ToolCallID], content), Role: m.Role, Content: content}) + at = append(at, i) + } + // Une seule transaction pour tous les blocs (AJEAN #130). + if entries, aerr := archiveRecallBlocks(blocks); aerr == nil { + for k, e := range entries { + archived[at[k]] = &recallEntry{id: e.id, label: e.label} + index = append(index, e) } - archived[i] = &recallEntry{id: id, label: label} - index = append(index, recallEntry{id: id, label: label}) + } else if len(blocks) > 0 { + fmt.Fprintf(os.Stderr, "[compact] archives recall non écrites (%v) : repli sans rappel\n", aerr) } } @@ -528,6 +570,14 @@ func compactMessagesCore(ctx context.Context, msgs []Message, caps Caps, opt com } else { summary, err = summarizeTranscriptFor(ctx, renderTranscript(forSummary), caps.Code) } + // La cause d'un résumé raté n'était écrite nulle part : on ne voyait que des + // compactions « changé=false » à répétition, sans savoir pourquoi (AJEAN + // 68659e73). + if err != nil { + fmt.Fprintf(os.Stderr, "[compact] résumé raté : %v\n", err) + } else if summaryLooksEmpty(summary) { + fmt.Fprintf(os.Stderr, "[compact] résumé vide ou dégénéré (%d car.)\n", len([]rune(strings.TrimSpace(summary)))) + } var mid []Message if err != nil || summaryLooksEmpty(summary) { // Résumé raté (erreur, vide, ou juste une référence recall) → on garde le diff --git a/internal/loki/chat_compact_retry_test.go b/internal/loki/chat_compact_retry_test.go new file mode 100644 index 0000000..83858ae --- /dev/null +++ b/internal/loki/chat_compact_retry_test.go @@ -0,0 +1,32 @@ +package loki + +import "testing" + +// Après une compaction en cours de tour qui n'a rien réduit, l'étape suivante +// ne relance pas un résumé tant que le contexte n'a pas grossi de 5 % de la +// fenêtre — sauf à 90 % de la fenêtre ou plus. +func TestCompactRetryBackoff(t *testing.T) { + testHome(t) + if err := putStr(bkConfig, "CTX", "100000"); err != nil { + t.Fatal(err) + } + if !compactRetryDue(0, 70000, nil) { + t.Fatal("sans échec précédent, l'essai doit être permis") + } + next := compactNextRetry(false, 70000, nil) + if next != 75000 { + t.Fatalf("prochain essai à %d, attendu 75000", next) + } + if compactRetryDue(next, 72000, nil) { + t.Fatal("relancé alors que le contexte n'a presque pas grossi") + } + if !compactRetryDue(next, 75000, nil) { + t.Fatal("pas relancé alors que le contexte a grossi de 5 % de la fenêtre") + } + if !compactRetryDue(200000, 90000, nil) { + t.Fatal("à 90 % de la fenêtre, l'essai doit toujours être permis") + } + if compactNextRetry(true, 70000, nil) != 0 { + t.Fatal("une compaction réussie doit lever le garde-fou") + } +} diff --git a/internal/loki/chat_recall.go b/internal/loki/chat_recall.go index 959093d..911eed4 100644 --- a/internal/loki/chat_recall.go +++ b/internal/loki/chat_recall.go @@ -78,58 +78,88 @@ func recallIDToSeq(id string) (uint64, bool) { return n, true } -// archiveRecallBlock enregistre un bloc et renvoie son id ("r"). L'id est -// alloué par NextSequence dans la transaction d'écriture : monotone, sans -// collision, même si deux compactages se croisent. En cas d'échec d'écriture -// (base indisponible, ex. tests sans LOKI_HOME) on renvoie une erreur et -// l'appelant se rabat sur le compactage classique (troncature sans id). -func archiveRecallBlock(label, role, content string) (string, error) { +// recallArchiveInput : un bloc à archiver (voir archiveRecallBlocks). +type recallArchiveInput struct { + Label string + Role string + Content string +} + +// archiveRecallBlocks enregistre tous les blocs d'une compaction dans UNE +// transaction et renvoie leurs entrées, dans l'ordre des blocs fournis. Les ids +// sont alloués par NextSequence dans la transaction : monotones, sans +// collision, même si deux compactages se croisent. Une transaction par bloc +// payait une ouverture de la base et une synchronisation disque chacune, sous +// le verrou global de la base, en plein tour (AJEAN #130). Tout ou rien : en +// cas d'échec (base indisponible, ex. tests sans LOKI_HOME, mémoire +// verrouillée) aucun id n'est rendu, et l'appelant se rabat sur le compactage +// classique (troncature sans id). +func archiveRecallBlocks(blocks []recallArchiveInput) ([]recallEntry, error) { + if len(blocks) == 0 { + return nil, nil + } // Résolu AVANT d'ouvrir la base : chiffrer dans la transaction relirait la // configuration, donc la base, qui n'est pas rentrante (voir memEncoderNow). encode, err := memEncoderNow() if err != nil { - return "", err + return nil, err } - var id string + now := time.Now().Unix() + var entries []recallEntry err = withDB(func(d *bolt.DB) error { return d.Update(func(tx *bolt.Tx) error { + entries = entries[:0] b, err := tx.CreateBucketIfNotExists([]byte(bkRecall)) if err != nil { return err } - seq, err := b.NextSequence() - if err != nil { - return err + for _, in := range blocks { + seq, err := b.NextSequence() + if err != nil { + return err + } + blk := recallBlock{ + ID: "r" + strconv.FormatUint(seq, 10), + Seq: seq, + Label: in.Label, + Role: in.Role, + Content: in.Content, + Created: now, + } + raw, err := json.Marshal(blk) + if err != nil { + return err + } + // Un bloc archivé, c'est du verbatim de conversation : chiffré + // comme le reste quand la mémoire l'est (l'encodeur rend le clair + // inchangé si le chiffrement est inactif ; verrouillé, + // memEncoderNow a déjà refusé plus haut). + enc, err := encode(raw) + if err != nil { + return err + } + if err := b.Put(recallKey(seq), enc); err != nil { + return err + } + entries = append(entries, recallEntry{id: blk.ID, label: blk.Label}) } - blk := recallBlock{ - ID: "r" + strconv.FormatUint(seq, 10), - Seq: seq, - Label: label, - Role: role, - Content: content, - Created: time.Now().Unix(), - } - id = blk.ID - raw, err := json.Marshal(blk) - if err != nil { - return err - } - // Un bloc archivé, c'est du verbatim de conversation : chiffré comme - // le reste quand la mémoire l'est (l'encodeur rend le clair inchangé - // si le chiffrement est inactif ; verrouillé, memEncoderNow a déjà - // refusé plus haut). - enc, err := encode(raw) - if err != nil { - return err - } - return b.Put(recallKey(seq), enc) + return nil }) }) if err != nil { - return "", err + return nil, err } cacheBust(bkRecall) - return id, nil + return entries, nil +} + +// archiveRecallBlock enregistre un seul bloc et renvoie son id ("r"). +func archiveRecallBlock(label, role, content string) (string, error) { + e, err := archiveRecallBlocks([]recallArchiveInput{{Label: label, Role: role, Content: content}}) + if err != nil { + return "", err + } + return e[0].id, nil } // recallGet ramène un bloc par son id. false si l'id est mal formé ou absent. diff --git a/internal/loki/chat_recall_test.go b/internal/loki/chat_recall_test.go index 1071ce6..fbc2554 100644 --- a/internal/loki/chat_recall_test.go +++ b/internal/loki/chat_recall_test.go @@ -221,3 +221,47 @@ func TestCompactSummaryUserMsg(t *testing.T) { t.Fatalf("l'index des blocs rappelables est absent :\n%s", avec) } } + +// Les blocs d'une compaction s'écrivent en une seule transaction : ids dans +// l'ordre des blocs, croissants, chacun relisible mot pour mot. +func TestArchiveRecallBlocksBatch(t *testing.T) { + testHome(t) + if e, err := archiveRecallBlocks(nil); err != nil || e != nil { + t.Fatalf("lot vide : %v, %v", e, err) + } + in := []recallArchiveInput{ + {Label: "un", Role: "tool", Content: bigBlock("PREMIER")}, + {Label: "deux", Role: "user", Content: bigBlock("SECOND")}, + {Label: "trois", Role: "assistant", Content: bigBlock("TROISIEME")}, + } + entries, err := archiveRecallBlocks(in) + if err != nil { + t.Fatal(err) + } + if len(entries) != len(in) { + t.Fatalf("%d entrées pour %d blocs", len(entries), len(in)) + } + var last uint64 + for i, e := range entries { + seq, ok := recallIDToSeq(e.id) + if !ok || seq <= last { + t.Fatalf("id %q non croissant (précédent %d)", e.id, last) + } + last = seq + if e.label != in[i].Label { + t.Fatalf("entrée %d : libellé %q, attendu %q", i, e.label, in[i].Label) + } + blk, ok := recallGet(e.id) + if !ok || blk.Content != in[i].Content || blk.Role != in[i].Role { + t.Fatalf("bloc %s mal relu : %+v", e.id, blk) + } + } + // Un archivage isolé suit la même séquence. + id, err := archiveRecallBlock("quatre", "tool", bigBlock("QUATRE")) + if err != nil { + t.Fatal(err) + } + if seq, _ := recallIDToSeq(id); seq != last+1 { + t.Fatalf("id isolé %q, attendu r%d", id, last+1) + } +} diff --git a/internal/loki/chat_stuck_test.go b/internal/loki/chat_stuck_test.go index 47f6ef1..a80f1be 100644 --- a/internal/loki/chat_stuck_test.go +++ b/internal/loki/chat_stuck_test.go @@ -41,11 +41,15 @@ func TestRunShellNeBloquePasSurUnProcessDetache(t *testing.T) { } testHome(t) start := time.Now() - out := runShell(context.Background(), "sleep 30 & echo lance", 2) - if d := time.Since(start); d > 10*time.Second { + // Délai de 10 s, bien au-delà de WaitDelay (2 s) : c'est le détachement + // qui rend la main, pas le délai de la commande. + out := runShell(context.Background(), "sleep 30 & echo lance", 10) + if d := time.Since(start); d > 8*time.Second { t.Fatalf("runShell est resté accroché aux tubes du process détaché (%s)", d) } - if !strings.Contains(out, "lance") && !strings.Contains(out, "timeout") { + // La commande a fini (exit 0) : sa sortie est rendue, pas effacée par + // « WaitDelay expired » (AJEAN 7403a8a0). + if !strings.Contains(out, "exit: 0") || !strings.Contains(out, "lance") || !strings.Contains(out, "arrière-plan") { t.Errorf("sortie inattendue : %q", out) } } diff --git a/internal/loki/chat_tools.go b/internal/loki/chat_tools.go index 91c1411..0750759 100644 --- a/internal/loki/chat_tools.go +++ b/internal/loki/chat_tools.go @@ -297,6 +297,14 @@ func runShell(parent context.Context, command string, timeoutSec int) string { // issue connue était de redémarrer loki-ui. cmd.WaitDelay = 2 * time.Second err := cmd.Run() + // ErrWaitDelay : la commande s'est terminée normalement, mais un process + // laissé en arrière-plan tenait encore les tubes. Ce qu'elle a écrit est + // déjà capturé : le rendre, au lieu de « [erreur: exec: WaitDelay expired + // before I/O complete] » qui l'effaçait (repris d'AJEAN 7403a8a0). + detached := errors.Is(err, exec.ErrWaitDelay) + if detached { + err = nil + } // Sortie nettoyée (couleurs ANSI retirées), réduite à la fin si la commande // demandait un tail. clean := func(s string) string { return tailOutput(stripANSI(s)) } @@ -337,6 +345,9 @@ func runShell(parent context.Context, command string, timeoutSec int) string { if errOut != "" { parts = append(parts, "stderr:\n"+errOut) } + if detached { + parts = append(parts, "(un process lancé en arrière-plan tourne toujours ; sa sortie n'est plus suivie — pour un serveur, utilise bash_bg)") + } return strings.Join(parts, "\n\n") } diff --git a/internal/loki/code_retry.go b/internal/loki/code_retry.go index 6507bc6..7945e04 100644 --- a/internal/loki/code_retry.go +++ b/internal/loki/code_retry.go @@ -16,6 +16,7 @@ import ( "regexp" "strings" "sync" + "time" "unicode/utf8" ) @@ -40,39 +41,94 @@ var defaultRetryPatterns = []string{ // Compilés à la première utilisation, pas à l'init du package : la surcharge // vit dans la base bbolt, qu'on ne veut pas ouvrir avant que le process ait // choisi son rôle (CLI, serveur…). +// +// Relus quand la surcharge change (OpenFox c52c5bf7) : un sync.Once les +// figeait jusqu'au redémarrage du process. retryPatterns tourne pendant le flux, +// à chaque morceau : la base n'est rouverte que si une écriture de ce process a +// touché le bucket d'état (retryStateBust, via cacheBust) ou, au plus une fois +// par seconde, si date ou taille du fichier trahissent l'écriture d'un autre +// process (`loki config set` pendant que le service tourne). var ( - retryOnce sync.Once - retryRes []*regexp.Regexp + retryMu sync.Mutex + retryRes []*regexp.Regexp // retryCustom : la surcharge `retry_patterns` est active. Ses motifs sont // quelconques (\A, (?s).*? sur des Ko…) : le balayage fenêtré pendant le // flux (textualToolCallFrom) ne vaut que pour les motifs embarqués. - retryCustom bool + retryCustom bool + retryLoaded bool + retryStale bool // écriture locale du bucket d'état depuis la lecture + retryRaw string + retryPath string // la base lue (LOKI_HOME change d'un test à l'autre) + retryChecked time.Time + retryMtime time.Time + retrySize int64 ) -func retryPatterns() []*regexp.Regexp { - retryOnce.Do(func() { retryRes, retryCustom = compileRetryPatterns() }) - return retryRes +// retryStateBust : appelé par cacheBust pour toute écriture du bucket d'état. +func retryStateBust() { + retryMu.Lock() + retryStale = true + retryMu.Unlock() } -func compileRetryPatterns() ([]*regexp.Regexp, bool) { - pats := defaultRetryPatterns - custom := false - // Surcharge optionnelle : clé d'état `retry_patterns` = tableau JSON de - // regex. Pas d'UI dédiée — c'est un réglage d'expert, posé via - // `loki config` ou l'API ; l'embarqué couvre les cas connus. - if raw := getStr(bkState, "retry_patterns"); raw != "" { - var over []string - if json.Unmarshal([]byte(raw), &over) == nil && len(over) > 0 { - pats, custom = over, true +func retryPatterns() []*regexp.Regexp { + res, _ := retryState() + return res +} + +// retryState renvoie les motifs compilés et si ce sont ceux de la surcharge. +func retryState() ([]*regexp.Regexp, bool) { + retryMu.Lock() + defer retryMu.Unlock() + reload := !retryLoaded || retryStale || retryPath != dbPath() + if !reload && time.Since(retryChecked) >= time.Second { + retryChecked = time.Now() + if mtime, size := dbStamp(); size != retrySize || !mtime.Equal(retryMtime) { + reload = true } } + if reload { + raw := getStr(bkState, "retry_patterns") + retryMtime, retrySize = dbStamp() + retryChecked, retryStale, retryPath = time.Now(), false, dbPath() + if !retryLoaded || raw != retryRaw { + retryRes, retryCustom = compileRetryPatterns(raw) + retryRaw, retryLoaded = raw, true + } + } + return retryRes, retryCustom +} + +// compileRetryPatterns compile la surcharge raw (tableau JSON de regex), ou +// les motifs embarqués quand elle est absente, illisible, ou sans un seul motif +// utilisable. Un motif vide est écarté : "(?mi)" correspond à TOUT, et chaque +// tour était relancé une fois (OpenFox 8e9b4869). +func compileRetryPatterns(raw string) ([]*regexp.Regexp, bool) { + // Surcharge optionnelle : clé d'état `retry_patterns`. Pas d'UI dédiée — + // c'est un réglage d'expert, posé via `loki config` ou l'API ; l'embarqué + // couvre les cas connus. + if raw != "" { + var over []string + if json.Unmarshal([]byte(raw), &over) == nil { + if out := compileRetryList(over); len(out) > 0 { + return out, true + } + } + } + return compileRetryList(defaultRetryPatterns), false +} + +func compileRetryList(pats []string) []*regexp.Regexp { var out []*regexp.Regexp for _, p := range pats { + if strings.TrimSpace(p) == "" { + continue + } if re, err := regexp.Compile("(?mi)" + p); err == nil { out = append(out, re) } } - return out, custom + return out } // textualToolCall dit si le texte final du tour contient un appel d'outil @@ -100,9 +156,9 @@ func textualToolCallFrom(content string, scanned int) bool { if content == "" { return false } - res := retryPatterns() + res, custom := retryState() start := 0 - if !retryCustom { + if !custom { start = retryScanStart(content, scanned) } w := content[start:] diff --git a/internal/loki/code_retry_patterns_test.go b/internal/loki/code_retry_patterns_test.go new file mode 100644 index 0000000..cbdbf66 --- /dev/null +++ b/internal/loki/code_retry_patterns_test.go @@ -0,0 +1,47 @@ +package loki + +import "testing" + +// Un motif vide dans la surcharge correspondait à tout : chaque tour était +// relancé une fois. Il est écarté ; une surcharge sans motif utilisable retombe +// sur les motifs embarqués. +func TestCompileRetryPatternsSkipsEmpty(t *testing.T) { + res, custom := compileRetryPatterns(`["", " ", ""]`) + if !custom || len(res) != 1 { + t.Fatalf("custom=%v, %d motifs ; attendu la surcharge avec un seul motif", custom, len(res)) + } + if res[0].MatchString("une réponse tout à fait normale") { + t.Fatal("un texte ordinaire déclenche une relance") + } + res, custom = compileRetryPatterns(`["", "([invalide"]`) + if custom || len(res) != len(defaultRetryPatterns) { + t.Fatalf("custom=%v, %d motifs ; attendu les motifs embarqués", custom, len(res)) + } + if _, custom = compileRetryPatterns(`pas du json`); custom { + t.Fatal("surcharge illisible prise en compte") + } +} + +// Un changement de retry_patterns agit sans redémarrer le process. +func TestRetryPatternsReloadOnChange(t *testing.T) { + testHome(t) + retryStateBust() + if _, custom := retryState(); custom { + t.Fatal("surcharge active sans clé") + } + if err := putStr(bkState, "retry_patterns", `["OUPS-OUTIL"]`); err != nil { + t.Fatal(err) + } + if !textualToolCall("bon, OUPS-OUTIL, voilà") { + t.Fatal("nouveau motif ignoré après l'écriture") + } + if err := putStr(bkState, "retry_patterns", ""); err != nil { + t.Fatal(err) + } + if _, custom := retryState(); custom { + t.Fatal("surcharge retirée mais toujours active") + } + if textualToolCall("bon, OUPS-OUTIL, voilà") { + t.Fatal("ancien motif encore appliqué") + } +} diff --git a/internal/loki/facade_chat_test.go b/internal/loki/facade_chat_test.go index 184110c..259ef91 100644 --- a/internal/loki/facade_chat_test.go +++ b/internal/loki/facade_chat_test.go @@ -1,6 +1,7 @@ package loki import ( + "context" "encoding/json" "errors" "io" @@ -857,10 +858,10 @@ func TestFacadeMCP(t *testing.T) { if len(tools) != 1 || tools[0].Function.Name != "mcp__a__t" || mcpPromptLine(tools) == "" { t.Fatalf("outils : %+v", tools) } - if got := mcpCall("mcp__a__t", map[string]any{"x": 1}); got != "fait" { + if got := mcpCall(context.Background(), "mcp__a__t", map[string]any{"x": 1}); got != "fait" { t.Fatalf("appel : %q", got) } - if got := mcpCall("mcp__a__t", nil); got != "fait" || !strings.HasSuffix(calls[len(calls)-1], `{"args":{},"name":"mcp__a__t"}`) { + if got := mcpCall(context.Background(), "mcp__a__t", nil); got != "fait" || !strings.HasSuffix(calls[len(calls)-1], `{"args":{},"name":"mcp__a__t"}`) { t.Fatalf("appel sans arguments : %q (%v)", got, calls) } MCPPrewarm() @@ -872,7 +873,7 @@ func TestFacadeMCP(t *testing.T) { } // Façade injoignable : une erreur lisible, pas de panique. t.Setenv(facadeAddrEnv, "127.0.0.1:1") - if got := mcpCall("mcp__a__t", nil); !strings.HasPrefix(got, "[erreur MCP] façade injoignable") { + if got := mcpCall(context.Background(), "mcp__a__t", nil); !strings.HasPrefix(got, "[erreur MCP] façade injoignable") { t.Fatalf("façade injoignable : %q", got) } if tools := mcpTools(); tools != nil { diff --git a/internal/loki/llm_client.go b/internal/loki/llm_client.go index 4ee38b6..f6b765e 100644 --- a/internal/loki/llm_client.go +++ b/internal/loki/llm_client.go @@ -1373,6 +1373,9 @@ func runChatLoop(ctx context.Context, kt *keepImages, messages []Message, tools // Contexte estimé pour la requête suivante (en-tour), comparé au compte // réel du moteur quand il arrive — journalisé s'il s'en écarte. lastEst := 0 + // Compaction en cours de tour qui n'a rien pu réduire : la retenter + // seulement une fois le contexte nettement plus gros (compactRetryDue). + compactRetryAt := 0 // Garde-fou « appel d'outil écrit en texte » (code_retry.go) : relances // consécutives bornées, compteur remis à zéro après chaque outil exécuté. patternRetries := 0 @@ -2468,7 +2471,8 @@ func runChatLoop(ctx context.Context, kt *keepImages, messages []Message, tools publishHistory() } } - if compactNeeded(messages, used, peakGen) && !compactRefusedSkip(mainConv, used, messages) { + if compactNeeded(messages, used, peakGen) && compactRetryDue(compactRetryAt, used, messages) && + !compactRefusedSkip(mainConv, used, messages) { opt := compactOptsMid(chatCfg, ep, mainConv, messages, used, tools, reasoningEffort, reasoningKwargs, pol, disableTools || toolChoiceNone) yes, no := true, false @@ -2476,6 +2480,7 @@ func runChatLoop(ctx context.Context, kt *keepImages, messages []Message, tools c, changed := compactMessagesNoted(ctx, messages, caps, opt) cb(StreamEvent{Compacting: &no}) logCompact("en-tour", used, messages, c, changed) + compactRetryAt = compactNextRetry(changed, used, messages) if changed { lastEst = 0 // estimation d'avant compaction : plus rien à comparer messages = c @@ -2810,7 +2815,7 @@ func execTool(ctx context.Context, name string, args map[string]any, label strin result, visionImg = toolCUScreenshot() default: if isMCPTool(name) { - result = mcpCall(name, args) + result = mcpCall(ctx, name, args) } else { result = "[erreur] outil inconnu: " + name } diff --git a/internal/loki/lsp.go b/internal/loki/lsp.go index a90b0d2..19099bc 100644 --- a/internal/loki/lsp.go +++ b/internal/loki/lsp.go @@ -25,6 +25,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" ) @@ -73,6 +74,14 @@ const ( lspMaxDiags = 10 // diagnostics montrés au modèle ) +// lspDiagQuiet : calme attendu après une publication avant de la croire +// définitive. gopls, rust-analyzer et pyright publient souvent en plusieurs +// fois pour une même version (liste vide d'abord, puis leur analyse, puis +// celle d'un outil externe) : rendre la PREMIÈRE donnait « propre » à une +// édition qui cassait la compilation. OpenFox attend 400 ms ; variable pour +// les tests. +var lspDiagQuiet = 350 * time.Millisecond + type lspServer struct { lang string root string @@ -82,11 +91,17 @@ type lspServer struct { nextID int version map[string]int // uri → version de document // diags : dernier publishDiagnostics reçu par uri, et signal d'arrivée. - diagMu sync.Mutex - diags map[string][]lspDiagnostic + diagMu sync.Mutex + diags map[string][]lspDiagnostic + // diagVer : version de document annoncée par la dernière publication de + // l'uri, -1 quand le serveur n'en donne pas (champ facultatif du protocole). + diagVer map[string]int diagCh chan string // uri fraîchement publié lastUsed time.Time - dead bool + // dead : atomique, pas sous mu. writeMsg le pose sur une écriture ratée, et + // writeMsg tourne DANS check, qui tient déjà mu : le prendre là bloquait le + // serveur (et le tour) à vie. lspGet le lit aussi sans mu. + dead atomic.Bool } type lspDiagnostic struct { @@ -214,7 +229,7 @@ func lspGet(lang *lspLanguage, root string) (*lspServer, error) { key := lang.ID + "\x00" + root lspMu.Lock() defer lspMu.Unlock() - if s := lspServers[key]; s != nil && !s.dead { + if s := lspServers[key]; s != nil && !s.dead.Load() { return s, nil } s, err := lspStart(lang, root) @@ -242,7 +257,7 @@ func lspStart(lang *lspLanguage, root string) (*lspServer, error) { } s := &lspServer{ lang: lang.ID, root: root, cmd: cmd, stdin: stdin, - version: map[string]int{}, diags: map[string][]lspDiagnostic{}, + version: map[string]int{}, diags: map[string][]lspDiagnostic{}, diagVer: map[string]int{}, diagCh: make(chan string, 16), lastUsed: time.Now(), } go s.readLoop(stdout) @@ -268,7 +283,7 @@ func lspStart(lang *lspLanguage, root string) (*lspServer, error) { func (s *lspServer) check(absPath string) ([]lspDiagnostic, bool) { s.mu.Lock() defer s.mu.Unlock() - if s.dead { + if s.dead.Load() { return nil, false } s.lastUsed = time.Now() @@ -277,16 +292,29 @@ func (s *lspServer) check(absPath string) ([]lspDiagnostic, bool) { return nil, false } uri := fileURI(absPath) + // Signaux en attente : tous PÉRIMÉS. Un serveur qui republie après le + // retour du check précédent laisse son uri dans le tampon ; lu ici tout de + // suite après didChange, il faisait rendre « aucun diagnostic » avant même + // que le serveur ait vu la nouvelle version. + for drained := false; !drained; { + select { + case <-s.diagCh: + default: + drained = true + } + } // Vide les publications en attente pour cet uri : on veut la PROCHAINE. s.diagMu.Lock() delete(s.diags, uri) + delete(s.diagVer, uri) s.diagMu.Unlock() v, open := s.version[uri] if !open { - s.version[uri] = 1 + v = 1 + s.version[uri] = v s.notify("textDocument/didOpen", map[string]any{ "textDocument": map[string]any{ - "uri": uri, "languageId": s.lang, "version": 1, "text": string(content), + "uri": uri, "languageId": s.lang, "version": v, "text": string(content), }, }) } else { @@ -297,24 +325,56 @@ func (s *lspServer) check(absPath string) ([]lspDiagnostic, bool) { "contentChanges": []map[string]any{{"text": string(content)}}, }) } - // Attend la publication pour CET uri, bornée. - deadline := time.After(lspDiagWait) + if s.dead.Load() { + return nil, false // écriture ratée : le serveur ne verra jamais cette version + } + // current : les diagnostics publiés pour CETTE version (ou sans version), + // pas ceux d'une version antérieure arrivés en retard. + current := func() ([]lspDiagnostic, bool) { + s.diagMu.Lock() + defer s.diagMu.Unlock() + d, ok := s.diags[uri] + if !ok { + return nil, false + } + if pv, has := s.diagVer[uri]; has && pv >= 0 && pv < v { + return nil, false + } + return d, true + } + // Attend la publication pour CET uri, puis un calme de lspDiagQuiet sans + // nouvelle publication ; le tout borné par lspDiagWait. + deadline := time.NewTimer(lspDiagWait) + defer deadline.Stop() + var quiet <-chan time.Time + var quietT *time.Timer + defer func() { + if quietT != nil { + quietT.Stop() + } + }() for { select { case got := <-s.diagCh: if got != uri { continue } - s.diagMu.Lock() - d := s.diags[uri] - s.diagMu.Unlock() + if _, ok := current(); !ok { + continue // publication d'une version antérieure + } + if quietT == nil { + quietT = time.NewTimer(lspDiagQuiet) + } else { + quietT.Reset(lspDiagQuiet) + } + quiet = quietT.C + case <-quiet: + d, _ := current() return d, true - case <-deadline: + case <-deadline.C: // Pas de publication à temps : certains serveurs ne publient rien // quand il n'y a AUCUN diagnostic. On rend « rien », pas un échec. - s.diagMu.Lock() - d := s.diags[uri] - s.diagMu.Unlock() + d, _ := current() return d, true } } @@ -353,6 +413,7 @@ func (s *lspServer) readLoop(r io.Reader) { Method string `json:"method"` Params struct { URI string `json:"uri"` + Version *int `json:"version"` Diagnostics []lspDiagnostic `json:"diagnostics"` } `json:"params"` } @@ -362,6 +423,11 @@ func (s *lspServer) readLoop(r io.Reader) { if msg.Method == "textDocument/publishDiagnostics" { s.diagMu.Lock() s.diags[msg.Params.URI] = msg.Params.Diagnostics + ver := -1 + if msg.Params.Version != nil { + ver = *msg.Params.Version + } + s.diagVer[msg.Params.URI] = ver s.diagMu.Unlock() select { case s.diagCh <- msg.Params.URI: @@ -377,7 +443,7 @@ func (s *lspServer) idleWatch() { for range t.C { s.mu.Lock() idle := time.Since(s.lastUsed) > lspIdleTimeout - dead := s.dead + dead := s.dead.Load() s.mu.Unlock() if dead { return @@ -401,11 +467,7 @@ func (s *lspServer) shutdown() { lspMu.Unlock() } -func (s *lspServer) markDead() { - s.mu.Lock() - s.dead = true - s.mu.Unlock() -} +func (s *lspServer) markDead() { s.dead.Store(true) } func (s *lspServer) send(method string, params any, isRequest bool) { s.nextID++ diff --git a/internal/loki/lsp_check_test.go b/internal/loki/lsp_check_test.go new file mode 100644 index 0000000..5fef16e --- /dev/null +++ b/internal/loki/lsp_check_test.go @@ -0,0 +1,200 @@ +package loki + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// fakeLSP : un serveur LSP réduit à ce que check utilise. Pour chaque +// didOpen/didChange reçu, react(version, publish) décide des publications. +type fakeLSP struct { + s *lspServer + out *io.PipeWriter +} + +func lspFrame(w io.Writer, v any) { + b, _ := json.Marshal(v) + fmt.Fprintf(w, "Content-Length: %d\r\n\r\n%s", len(b), b) +} + +func newFakeLSP(t *testing.T, react func(ver int, publish func(ver *int, diags []lspDiagnostic))) *fakeLSP { + t.Helper() + inR, inW := io.Pipe() + outR, outW := io.Pipe() + s := &lspServer{ + lang: "go", root: t.TempDir(), stdin: inW, + version: map[string]int{}, diags: map[string][]lspDiagnostic{}, diagVer: map[string]int{}, + diagCh: make(chan string, 16), lastUsed: time.Now(), + } + go s.readLoop(outR) + go func() { + br := bufio.NewReader(inR) + for { + n := 0 + for { + line, err := br.ReadString('\n') + if err != nil { + return + } + line = strings.TrimSpace(line) + if line == "" { + break + } + if v, ok := strings.CutPrefix(line, "Content-Length:"); ok { + fmt.Sscanf(strings.TrimSpace(v), "%d", &n) + } + } + buf := make([]byte, n) + if _, err := io.ReadFull(br, buf); err != nil { + return + } + var msg struct { + Method string `json:"method"` + Params struct { + TextDocument struct { + URI string `json:"uri"` + Version int `json:"version"` + } `json:"textDocument"` + } `json:"params"` + } + if json.Unmarshal(buf, &msg) != nil { + continue + } + if msg.Method != "textDocument/didOpen" && msg.Method != "textDocument/didChange" { + continue + } + uri := msg.Params.TextDocument.URI + go react(msg.Params.TextDocument.Version, func(ver *int, diags []lspDiagnostic) { + params := map[string]any{"uri": uri, "diagnostics": diags} + if ver != nil { + params["version"] = *ver + } + lspFrame(outW, map[string]any{"jsonrpc": "2.0", "method": "textDocument/publishDiagnostics", "params": params}) + }) + } + }() + t.Cleanup(func() { inW.Close(); outW.Close() }) + return &fakeLSP{s: s, out: outW} +} + +func lspErr(msg string) []lspDiagnostic { + d := lspDiagnostic{Severity: 1, Message: msg} + return []lspDiagnostic{d} +} + +func lspTempFile(t *testing.T) string { + t.Helper() + p := filepath.Join(t.TempDir(), "main.go") + if err := os.WriteFile(p, []byte("package main\n"), 0o644); err != nil { + t.Fatal(err) + } + return p +} + +func withLSPQuiet(t *testing.T, d time.Duration) { + old := lspDiagQuiet + lspDiagQuiet = d + t.Cleanup(func() { lspDiagQuiet = old }) +} + +// Un signal laissé dans le tampon par une republication tardive ne doit pas +// faire rendre « aucun diagnostic » à l'édition suivante. +func TestLSPCheckIgnoresStaleSignal(t *testing.T) { + withLSPQuiet(t, 80*time.Millisecond) + p := lspTempFile(t) + f := newFakeLSP(t, func(ver int, publish func(*int, []lspDiagnostic)) { + time.Sleep(60 * time.Millisecond) + publish(&ver, lspErr("undefined: x")) + }) + uri := fileURI(p) + f.s.version[uri] = 1 // document déjà ouvert + f.s.diagCh <- uri // republication tardive du check précédent + f.s.diags[uri] = nil // …dont la liste était vide + d, ok := f.s.check(p) + if !ok || len(d) != 1 || d[0].Message != "undefined: x" { + t.Fatalf("diagnostics = %+v, %v ; attendu l'erreur de la nouvelle version", d, ok) + } +} + +// Une liste vide publiée d'abord (« clear ») puis l'analyse : c'est l'analyse +// qui compte. +func TestLSPCheckWaitsPastInitialClear(t *testing.T) { + withLSPQuiet(t, 150*time.Millisecond) + p := lspTempFile(t) + f := newFakeLSP(t, func(ver int, publish func(*int, []lspDiagnostic)) { + publish(&ver, nil) + time.Sleep(60 * time.Millisecond) + publish(&ver, lspErr("missing return")) + }) + d, ok := f.s.check(p) + if !ok || len(d) != 1 || d[0].Message != "missing return" { + t.Fatalf("diagnostics = %+v, %v ; attendu l'erreur publiée après la liste vide", d, ok) + } +} + +// Une publication d'une version antérieure, arrivée en retard, est ignorée. +func TestLSPCheckIgnoresOlderVersion(t *testing.T) { + withLSPQuiet(t, 80*time.Millisecond) + p := lspTempFile(t) + f := newFakeLSP(t, func(ver int, publish func(*int, []lspDiagnostic)) { + old := ver - 1 + publish(&old, lspErr("ancienne erreur")) + time.Sleep(40 * time.Millisecond) + publish(&ver, nil) + }) + f.s.version[fileURI(p)] = 3 + d, ok := f.s.check(p) + if !ok || len(d) != 0 { + t.Fatalf("diagnostics = %+v, %v ; attendu aucun (seule la version 4 compte)", d, ok) + } +} + +// Sans version dans la publication (champ facultatif), elle est prise. +func TestLSPCheckAcceptsUnversionedPublish(t *testing.T) { + withLSPQuiet(t, 50*time.Millisecond) + p := lspTempFile(t) + f := newFakeLSP(t, func(ver int, publish func(*int, []lspDiagnostic)) { + publish(nil, lspErr("syntax error")) + }) + start := time.Now() + d, ok := f.s.check(p) + if !ok || len(d) != 1 { + t.Fatalf("diagnostics = %+v, %v", d, ok) + } + if time.Since(start) > lspDiagWait { + t.Fatalf("check a attendu la borne au lieu du calme") + } +} + +// Une écriture ratée vers le serveur ne bloque plus check (markDead prenait +// mu, déjà tenu par check). +func TestLSPCheckWriteFailureDoesNotDeadlock(t *testing.T) { + p := lspTempFile(t) + r, w := io.Pipe() + r.Close() // toute écriture échoue + s := &lspServer{ + lang: "go", stdin: w, + version: map[string]int{}, diags: map[string][]lspDiagnostic{}, diagVer: map[string]int{}, + diagCh: make(chan string, 16), + } + done := make(chan struct{}) + go func() { + s.check(p) + close(done) + }() + select { + case <-done: + case <-time.After(lspDiagWait + 2*time.Second): + t.Fatal("check bloqué après une écriture ratée") + } + if !s.dead.Load() { + t.Fatal("serveur pas marqué mort après une écriture ratée") + } +} diff --git a/internal/loki/mcp_call_ctx_test.go b/internal/loki/mcp_call_ctx_test.go new file mode 100644 index 0000000..239e72e --- /dev/null +++ b/internal/loki/mcp_call_ctx_test.go @@ -0,0 +1,96 @@ +package loki + +import ( + "context" + "strings" + "testing" + "time" + + mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// fakeMCPSlow branche un serveur MCP en mémoire dont l'outil « slow » ne rend +// la main qu'à l'annulation de sa requête (ou au bout de 30 s). +func fakeMCPSlow(t *testing.T) string { + t.Helper() + srv := mcpsdk.NewServer(&mcpsdk.Implementation{Name: "lent", Version: "1"}, nil) + srv.AddTool(&mcpsdk.Tool{Name: "slow", InputSchema: map[string]any{"type": "object"}}, + func(ctx context.Context, _ *mcpsdk.CallToolRequest) (*mcpsdk.CallToolResult, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(30 * time.Second): + return &mcpsdk.CallToolResult{Content: []mcpsdk.Content{&mcpsdk.TextContent{Text: "fini"}}}, nil + } + }) + st, ct := mcpsdk.NewInMemoryTransports() + ctx := context.Background() + ss, err := srv.Connect(ctx, st, nil) + if err != nil { + t.Fatal(err) + } + cs, err := mcpsdk.NewClient(&mcpsdk.Implementation{Name: "loki-test", Version: "1"}, nil).Connect(ctx, ct, nil) + if err != nil { + t.Fatal(err) + } + ready := make(chan struct{}) + close(ready) + name := mcpExposedName("lent", "slow") + mcpMgr.mu.Lock() + if mcpMgr.sessions == nil { + mcpMgr.sessions = map[string]*mcpSession{} + } + if mcpMgr.registry == nil { + mcpMgr.registry = map[string]mcpToolRef{} + } + mcpMgr.sessions["lent"] = &mcpSession{name: "lent", sess: cs, ready: ready} + mcpMgr.registry[name] = mcpToolRef{server: "lent", tool: "slow"} + mcpMgr.mu.Unlock() + t.Cleanup(func() { + mcpMgr.mu.Lock() + delete(mcpMgr.sessions, "lent") + delete(mcpMgr.registry, name) + mcpMgr.mu.Unlock() + cs.Close() + ss.Close() + }) + return name +} + +// Le bouton stop annule l'appel MCP en cours au lieu d'attendre ses 120 s. +func TestMCPCallHonorsTurnCancel(t *testing.T) { + testHome(t) + name := fakeMCPSlow(t) + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(150*time.Millisecond, cancel) + start := time.Now() + out := mcpCall(ctx, name, map[string]any{}) + if d := time.Since(start); d > 5*time.Second { + t.Fatalf("l'appel a ignoré l'annulation du tour (%s)", d) + } + if !strings.Contains(out, "interrompu") { + t.Fatalf("sortie = %q ; attendu un appel interrompu", out) + } +} + +func TestMCPCallTimeoutFor(t *testing.T) { + cases := []struct { + cfg MCPServerConfig + args map[string]any + want time.Duration + }{ + {MCPServerConfig{}, nil, mcpCallTimeout}, + {MCPServerConfig{CallTimeoutSec: 600}, nil, 10 * time.Minute}, + {MCPServerConfig{}, map[string]any{"timeout": 300.0}, 330 * time.Second}, // secondes + {MCPServerConfig{}, map[string]any{"timeout": 200000.0}, 230 * time.Second}, // millisecondes + {MCPServerConfig{}, map[string]any{"timeout": 5000.0}, mcpCallTimeout}, // 5 s : le défaut reste + {MCPServerConfig{}, map[string]any{"timeout": "400"}, 430 * time.Second}, // chaîne + {MCPServerConfig{}, map[string]any{"timeout": 86400.0 * 1000}, mcpCallTimeoutMax}, // plafond + {MCPServerConfig{CallTimeoutSec: 30}, nil, 30 * time.Second}, // plus court que le défaut + } + for i, c := range cases { + if got := mcpCallTimeoutFor(c.cfg, c.args); got != c.want { + t.Errorf("cas %d : %s, attendu %s", i, got, c.want) + } + } +} diff --git a/internal/loki/mcp_client.go b/internal/loki/mcp_client.go index ee2bc72..8118a9d 100644 --- a/internal/loki/mcp_client.go +++ b/internal/loki/mcp_client.go @@ -11,6 +11,7 @@ import ( "path/filepath" "regexp" "sort" + "strconv" "strings" "sync" "time" @@ -374,7 +375,11 @@ func isMCPTool(name string) bool { // mcpCall exécute un outil MCP à partir de son nom exposé et renvoie une chaîne // résultat (texte aplati), cappée. Reconnecte une fois si la session est morte. -func mcpCall(name string, args map[string]any) string { +// +// parent est le contexte DU TOUR, comme pour runShell : l'appel naissait d'un +// context.Background(), donc le bouton stop n'arrêtait rien et le tour restait +// bloqué jusqu'au bout des 120 s (OpenFox 2ef584b8). +func mcpCall(parent context.Context, name string, args map[string]any) string { if facadeMCP() { // le pool est celui de la façade (facade_mcp.go) return facadeMCPCall(name, args) } @@ -384,35 +389,92 @@ func mcpCall(name string, args map[string]any) string { if !ok { return "[erreur] outil MCP inconnu ou serveur non connecté: " + name } + servers, _ := LoadMCPConfig() + cfg, exists := servers[ref.server] + timeout := mcpCallTimeoutFor(cfg, args) - result, err := mcpCallOnce(ref, args) + result, err := mcpCallOnce(parent, ref, args, timeout) if err != nil { + if parent.Err() != nil { + return "[appel MCP interrompu]" + } + if errors.Is(err, context.DeadlineExceeded) { + // Le serveur est vivant mais l'outil est lent : le relancer + // doublerait l'attente (et l'effet de l'outil). + return fmt.Sprintf("[erreur MCP] pas de réponse après %s (délai de l'appel ; un argument timeout plus grand l'allonge)", timeout.Round(time.Second)) + } // Session peut-être morte (process stdio tombé) : on l'invalide et on // retente une fois avec une reconnexion fraîche. - servers, _ := LoadMCPConfig() - cfg, exists := servers[ref.server] if exists && cfg.Enabled { mcpInvalidate(ref.server) mcpMgr.ensure(ref.server, cfg) - if result2, err2 := mcpCallOnce(ref, args); err2 == nil { + if result2, err2 := mcpCallOnce(parent, ref, args, timeout); err2 == nil { return result2 } else { err = err2 } + if parent.Err() != nil { + return "[appel MCP interrompu]" + } } return "[erreur MCP] " + err.Error() } return result } -func mcpCallOnce(ref mcpToolRef, args map[string]any) (string, error) { +// mcpCallTimeoutMax plafonne le délai d'un appel, quoi que demandent l'outil +// ou la config : au-delà, c'est un job d'arrière-plan, pas un appel d'outil. +const mcpCallTimeoutMax = 30 * time.Minute + +// mcpCallTimeoutFor : délai d'un appel. Défaut mcpCallTimeout, remplacé par +// CallTimeoutSec du serveur ; un argument « timeout » de l'outil lui-même +// (beaucoup en ont un : navigateur, build, requête) l'allonge de 30 s pour que +// l'outil ait le temps de rendre SA propre erreur de délai. +func mcpCallTimeoutFor(cfg MCPServerConfig, args map[string]any) time.Duration { + d := mcpCallTimeout + if cfg.CallTimeoutSec > 0 { + d = time.Duration(cfg.CallTimeoutSec) * time.Second + } + if t := mcpArgTimeout(args); t > 0 && t+30*time.Second > d { + d = t + 30*time.Second + } + return min(d, mcpCallTimeoutMax) +} + +// mcpArgTimeout lit l'argument « timeout » d'un appel. Les serveurs MCP ne +// s'accordent pas sur l'unité : au-delà de 3600 on le lit en millisecondes +// (5000 = 5 s), en deçà en secondes (600 = 10 min). +func mcpArgTimeout(args map[string]any) time.Duration { + var v float64 + switch x := args["timeout"].(type) { + case float64: + v = x + case int: + v = float64(x) + case int64: + v = float64(x) + case json.Number: + v, _ = x.Float64() + case string: + v, _ = strconv.ParseFloat(strings.TrimSpace(x), 64) + } + if v <= 0 { + return 0 + } + if v > 3600 { + return time.Duration(v) * time.Millisecond + } + return time.Duration(v * float64(time.Second)) +} + +func mcpCallOnce(parent context.Context, ref mcpToolRef, args map[string]any, timeout time.Duration) (string, error) { mcpMgr.mu.Lock() s := mcpMgr.sessions[ref.server] mcpMgr.mu.Unlock() if s == nil || s.sess == nil { return "", fmt.Errorf("serveur %s non connecté", ref.server) } - ctx, cancel := context.WithTimeout(context.Background(), mcpCallTimeout) + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() res, err := s.sess.CallTool(ctx, &mcpsdk.CallToolParams{Name: ref.tool, Arguments: args}) if err != nil { diff --git a/internal/loki/mcp_client_test.go b/internal/loki/mcp_client_test.go index a6008c5..fb85040 100644 --- a/internal/loki/mcp_client_test.go +++ b/internal/loki/mcp_client_test.go @@ -1,6 +1,7 @@ package loki import ( + "context" "os/exec" "strings" "sync" @@ -83,7 +84,7 @@ func TestMCPEndToEnd(t *testing.T) { t.Fatalf("outil %s introuvable dans %d outils", echoName, len(tools)) } - out := mcpCall(echoName, map[string]any{"message": "bonjour-loki"}) + out := mcpCall(context.Background(), echoName, map[string]any{"message": "bonjour-loki"}) if !strings.Contains(out, "bonjour-loki") { t.Fatalf("echo n'a pas renvoyé le message ; got: %q", out) } diff --git a/internal/loki/mcp_config.go b/internal/loki/mcp_config.go index 9ab71fc..601c9b7 100644 --- a/internal/loki/mcp_config.go +++ b/internal/loki/mcp_config.go @@ -50,6 +50,11 @@ type MCPServerConfig struct { // non namespacé). Permet de garder un serveur connecté tout en masquant // certains de ses outils. Vide = tous les outils exposés. DisabledTools []string `json:"disabledTools,omitempty"` + + // CallTimeoutSec : délai d'un appel d'outil de ce serveur, en secondes. + // 0 = défaut (mcpCallTimeout). Pour un serveur dont les outils sont longs + // par nature (build, scraping, requête lourde). + CallTimeoutSec int `json:"callTimeoutSec,omitempty"` } // ToolDisabled indique si un outil (nom réel) est masqué pour ce serveur. diff --git a/internal/loki/parity_mcp_test.go b/internal/loki/parity_mcp_test.go index 837a7ab..0356272 100644 --- a/internal/loki/parity_mcp_test.go +++ b/internal/loki/parity_mcp_test.go @@ -299,9 +299,9 @@ func TestPariteMCP(t *testing.T) { results := map[string]string{} for _, n := range []string{"echo", "multi", "media", "link", "struct", "fail", "empty", "big", "titled", "secret", "dot_tool", "nope"} { name := mcpExposedName("alpha", n) - results[name] = mcpCall(name, callArgs[n]) + results[name] = mcpCall(context.Background(), name, callArgs[n]) } - results["mcp__zeta__x"] = mcpCall("mcp__zeta__x", nil) + results["mcp__zeta__x"] = mcpCall(context.Background(), "mcp__zeta__x", nil) fx["calls"] = results fx["prompt_line_live"] = mcpPromptLine(mcpTools()) diff --git a/internal/loki/store.go b/internal/loki/store.go index 607319e..46bf301 100644 --- a/internal/loki/store.go +++ b/internal/loki/store.go @@ -222,6 +222,10 @@ func cacheBust(bucket string) { cacheMu.Lock() delete(caches, bucket) cacheMu.Unlock() + if bucket == bkState { + webKeyCacheBust() + retryStateBust() + } } // dbStamp renvoie la date et la taille du fichier de base — de quoi repérer diff --git a/internal/loki/stream_hotpath_test.go b/internal/loki/stream_hotpath_test.go index c2ffb75..fec499c 100644 --- a/internal/loki/stream_hotpath_test.go +++ b/internal/loki/stream_hotpath_test.go @@ -340,15 +340,24 @@ func TestRetryScanStartWindows(t *testing.T) { // Surcharge retry_patterns : ses motifs ne sont pas bornés, donc pas de fenêtre. func TestTextualToolCallFromCustomFullScan(t *testing.T) { retryPatterns() + retryMu.Lock() saved := retryRes savedCustom := retryCustom - defer func() { retryRes, retryCustom = saved, savedCustom }() retryRes, retryCustom = []*regexp.Regexp{regexp.MustCompile(`(?mi)\Aoups`)}, true + retryChecked = time.Now() // pas de relecture pendant le test + retryMu.Unlock() + defer func() { + retryMu.Lock() + retryRes, retryCustom = saved, savedCustom + retryMu.Unlock() + }() s := "oups.\n" + strings.Repeat("Une phrase, puis une autre.\n", 400) if !textualToolCallFrom(s, len(s)-10) { t.Fatal("motif ancré en tête manqué") } + retryMu.Lock() retryCustom = false + retryMu.Unlock() if textualToolCallFrom(s, len(s)-10) { t.Fatal("la fenêtre aurait dû le manquer (le test ne prouve rien sinon)") } diff --git a/internal/loki/web_auth.go b/internal/loki/web_auth.go index 5a959cf..18318f8 100644 --- a/internal/loki/web_auth.go +++ b/internal/loki/web_auth.go @@ -10,6 +10,8 @@ import ( "net/url" "os" "strings" + "sync" + "time" ) // web_auth.go protège l'API de pilotage (loki web) quand elle est exposée sur @@ -19,7 +21,8 @@ import ( // La clé de pilotage est volontairement DISTINCTE de .api_key (qui, elle, // protège llama-server / les complétions). On veut pouvoir donner à un client // un accès aux complétions sans lui donner le droit de redémarrer la machine, -// et inversement. Elle est relue à chaque requête (pas de cache) pour qu'un +// et inversement. Elle est relue à chaque requête, +// depuis un cache invalidé à chaque écriture (voir readWebKeyErr), pour qu'un // changement de clé prenne effet sans redémarrer le serveur web. // readWebKeyErr renvoie la clé de pilotage en distinguant « aucune clé » d'une @@ -27,9 +30,62 @@ import ( // 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. +// +// requireWebAuth la consulte à CHAQUE requête : sans cache, chacune payait une +// ouverture complète de la base sous dbMu (vignettes chargées deux par deux, +// SSE, statut…), en concurrence avec l'écrivain des discussions (AJEAN #120). +// On garde la dernière lecture RÉUSSIE, invalidée comme cachedKV : écriture de +// ce process (cacheBust du bucket d'état), date/taille du fichier pour les +// autres process (la CLI), une seconde au plus. Une lecture ratée n'est jamais +// mise en cache : elle remonte, et l'API reste fermée. func readWebKeyErr() (string, error) { + path := dbPath() + mtime, size := dbStamp() + c := &webKeyCache + c.Lock() + if c.ok && c.path == path && c.size == size && c.mtime.Equal(mtime) && time.Since(c.when) < cacheMaxAge { + k := c.key + c.Unlock() + return k, nil + } + gen := c.gen + c.Unlock() b, err := getBytesErr(bkState, "web_key") - return string(b), err + if err != nil { + return "", err + } + // Date et taille relevées APRÈS la lecture : ouvrir la base peut à lui seul + // toucher le fichier (Windows), et un relevé d'avant ne correspondait alors + // jamais. Une écriture externe glissée entre la lecture et le relevé reste + // couverte par l'âge maximal. + mtime, size = dbStamp() + c.Lock() + // Une invalidation pendant la lecture : la valeur lue est peut-être déjà + // l'ancienne — ne pas la garder. + if c.gen == gen { + c.ok, c.path, c.key, c.when, c.mtime, c.size = true, path, string(b), time.Now(), mtime, size + } + c.Unlock() + return string(b), nil +} + +var webKeyCache struct { + sync.Mutex + ok bool + gen uint64 // incrémenté à chaque invalidation + path string // la base lue (LOKI_HOME change d'un test à l'autre) + key string + when time.Time + mtime time.Time + size int64 +} + +// webKeyCacheBust : appelé par cacheBust pour toute écriture du bucket d'état. +func webKeyCacheBust() { + webKeyCache.Lock() + webKeyCache.ok = false + webKeyCache.gen++ + webKeyCache.Unlock() } // readWebKey renvoie la clé de pilotage, ou "" si aucune n'est définie (ou diff --git a/internal/loki/web_auth_cache_test.go b/internal/loki/web_auth_cache_test.go new file mode 100644 index 0000000..bd2a9e2 --- /dev/null +++ b/internal/loki/web_auth_cache_test.go @@ -0,0 +1,88 @@ +package loki + +import ( + "testing" + "time" + + bolt "go.etcd.io/bbolt" +) + +// Une écriture de ce process est vue tout de suite : le cache ne retarde pas un +// changement de clé. +func TestWebKeyCacheInvalidatedByLocalWrite(t *testing.T) { + testHome(t) + if err := putStr(bkState, "web_key", "ancienne"); err != nil { + t.Fatal(err) + } + if k, err := readWebKeyErr(); err != nil || k != "ancienne" { + t.Fatalf("clé = %q, %v", k, err) + } + if err := putStr(bkState, "web_key", "nouvelle"); err != nil { + t.Fatal(err) + } + if k, err := readWebKeyErr(); err != nil || k != "nouvelle" { + t.Fatalf("clé après écriture = %q, %v ; attendu la nouvelle", k, err) + } +} + +// Une écriture d'un AUTRE process (sans cacheBust ici) finit vue : au plus +// cacheMaxAge plus tard. +func TestWebKeyCacheSeesExternalWrite(t *testing.T) { + testHome(t) + if err := putStr(bkState, "web_key", "a"); err != nil { + t.Fatal(err) + } + if k, _ := readWebKeyErr(); k != "a" { + t.Fatalf("clé = %q", k) + } + // Écriture directe, comme le ferait la CLI : pas de cacheBust. + err := withDB(func(d *bolt.DB) error { + return d.Update(func(tx *bolt.Tx) error { + return tx.Bucket([]byte(bkState)).Put([]byte("web_key"), []byte("b")) + }) + }) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(cacheMaxAge + 500*time.Millisecond) + for { + if k, _ := readWebKeyErr(); k == "b" { + return + } + if time.Now().After(deadline) { + t.Fatal("écriture externe jamais vue") + } + time.Sleep(50 * time.Millisecond) + } +} + +// Sans changement, les lectures suivantes n'ouvrent plus la base. +func TestWebKeyCacheServesRepeatedReads(t *testing.T) { + testHome(t) + if err := putStr(bkState, "web_key", "k"); err != nil { + t.Fatal(err) + } + if k, _ := readWebKeyErr(); k != "k" { + t.Fatalf("clé = %q", k) + } + // Base verrouillée par « un autre process » : une lecture qui l'ouvrirait + // attendrait la borne de 5 s puis échouerait. + dbMu.Lock() + type res struct { + k string + err error + } + ch := make(chan res, 1) + go func() { k, err := readWebKeyErr(); ch <- res{k, err} }() + select { + case r := <-ch: + dbMu.Unlock() + if r.err != nil || r.k != "k" { + t.Fatalf("clé = %q, %v", r.k, r.err) + } + case <-time.After(300 * time.Millisecond): + dbMu.Unlock() + <-ch + t.Fatal("lecture servie par la base, pas par le cache") + } +}