From 01c11be4ae763ab17ee197006a65fe8f825becdd Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Fri, 16 Oct 2026 14:23:06 +0200 Subject: [PATCH] =?UTF-8?q?Correctifs=20repris=20d'AJEAN=20et=20d'OpenFox?= =?UTF-8?q?=20:=20LSP,=20shell,=20cl=C3=A9=20web,=20MCP,=20recall,=20compa?= =?UTF-8?q?ctage,=20relances?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LSP (lsp.go) : un signal resté dans le tampon par une republication tardive faisait rendre « aucun diagnostic » à l'édition suivante, avant même que le serveur ait vu la nouvelle version. Le canal est vidé, une publication d'une version antérieure est ignorée, et check attend 350 ms de calme après la dernière publication (liste vide puis analyse : c'est l'analyse qui compte), toujours borné par lspDiagWait. Au passage : interblocage latent levé (writeMsg → markDead reprenait mu, déjà tenu par check) — dead devient atomique. - run_shell : ErrWaitDelay (commande finie, process laissé en arrière-plan sur les tubes) rend la sortie au lieu de « WaitDelay expired » qui l'effaçait (AJEAN 7403a8a0). - Clé de pilotage web : lue depuis un cache à chaque requête au lieu d'ouvrir la base sous dbMu (AJEAN #120). Lectures réussies seulement, invalidé par toute écriture du bucket d'état, par la date/taille du fichier (autre process) et au bout d'une seconde. - MCP : l'appel suit le contexte du tour (le bouton stop l'annule), délai par serveur (callTimeoutSec) et allongé par l'argument timeout de l'outil, plafonné à 30 min ; un délai dépassé n'est plus relancé (OpenFox 2ef584b8). - Recall : les blocs d'une compaction s'archivent en une seule transaction au lieu d'une par bloc (AJEAN #130). - Compactage en cours de tour : après un essai qui n'a rien réduit, pas de nouvel essai avant +5 % de la fenêtre (sauf à 90 %), pour les tâches et sous-agents aussi ; la cause d'un résumé raté est journalisée (AJEAN 68659e73). - Motifs de relance : un motif vide (qui correspondait à tout) est écarté, une surcharge sans motif utilisable retombe sur l'embarqué, et un changement de retry_patterns agit sans redémarrer (OpenFox 8e9b4869, c52c5bf7). Co-Authored-By: Claude Opus 5.5 --- internal/loki/chat_compact.go | 62 ++++++- internal/loki/chat_compact_retry_test.go | 32 ++++ internal/loki/chat_recall.go | 100 +++++++---- internal/loki/chat_recall_test.go | 44 +++++ internal/loki/chat_stuck_test.go | 10 +- internal/loki/chat_tools.go | 11 ++ internal/loki/code_retry.go | 94 ++++++++-- internal/loki/code_retry_patterns_test.go | 47 +++++ internal/loki/facade_chat_test.go | 7 +- internal/loki/llm_client.go | 9 +- internal/loki/lsp.go | 108 +++++++++--- internal/loki/lsp_check_test.go | 200 ++++++++++++++++++++++ internal/loki/mcp_call_ctx_test.go | 96 +++++++++++ internal/loki/mcp_client.go | 76 +++++++- internal/loki/mcp_client_test.go | 3 +- internal/loki/mcp_config.go | 5 + internal/loki/parity_mcp_test.go | 4 +- internal/loki/store.go | 4 + internal/loki/stream_hotpath_test.go | 11 +- internal/loki/web_auth.go | 60 ++++++- internal/loki/web_auth_cache_test.go | 88 ++++++++++ 21 files changed, 967 insertions(+), 104 deletions(-) create mode 100644 internal/loki/chat_compact_retry_test.go create mode 100644 internal/loki/code_retry_patterns_test.go create mode 100644 internal/loki/lsp_check_test.go create mode 100644 internal/loki/mcp_call_ctx_test.go create mode 100644 internal/loki/web_auth_cache_test.go 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") + } +}