diff --git a/docs/superpowers/specs/2026-10-16-rust-migration-lot39-anneau-telemetrie-design.md b/docs/superpowers/specs/2026-10-16-rust-migration-lot39-anneau-telemetrie-design.md new file mode 100644 index 0000000..916baef --- /dev/null +++ b/docs/superpowers/specs/2026-10-16-rust-migration-lot39-anneau-telemetrie-design.md @@ -0,0 +1,92 @@ +# Lot 39 — Anneau de télémétrie tenu par la façade + +## Contexte + +Chaque complétion qui passe par le moteur est rangée dans un anneau +(`perf_log.go`). L'anneau : + +- calcule la perte de cache d'une complétion contre la précédente de la même + discussion et de la même nature ; +- nomme ce qui s'est intercalé (sous-agent, client externe…) ; +- signale une perte au-delà du seuil du preset (stats finales) ; +- nourrit `/api/perf/summary` et les poids de l'optimiseur. + +Depuis le lot 29, la façade envoyait chaque complétion de ses tours au Go +(`/internal/v1/turn/perf`). C'était un aller-retour par complétion pour un +anneau tenu ailleurs. + +## Le lot + +**Rust** (`src/perfring.rs`, port de `perf_log.go`) : + +- l'anneau : 5000 entrées, 128 discussions suivies, 500 entrées remontées au + plus pour nommer ce qui s'est intercalé ; +- la perte : seulement si la complétion prolonge strictement la précédente + (empreintes cumulées des messages), cache connu, complétion allée au bout ; +- la ligne `[perf]` (LOKI_PERF_LOG) ; +- le seuil d'alerte, dans le même ordre de priorité qu'au lancement : + - `-cms` d'EXTRA_ARGS ; + - `LLAMA_ARG_CHECKPOINT_MIN_SPACING_NT` ; + - `CKPT_MIN_STEP` ; + - 2048 sur un modèle hybride ; + - plus le micro-batch ; +- le résumé de `/api/perf/summary` (par nature, par profondeur, brouillon, + dernier verdict de la sonde de gabarit). + +`ChatShared` porte l'anneau. Les tours, les passes de vérification, les +résumés de compaction et le préchauffage de la façade y rangent leurs +complétions directement. Les marques (perte, ce qui s'est intercalé, +alerte) partent sur les stats finales comme avant. + +La façade sert : + +- `POST /internal/facade/v1/perf/record` : une complétion faite par le Go + (sous-agent, tâche IA, bench, compaction et préchauffage de ses tours, + client externe de `/v1`), rendue telle que rangée ; +- `POST /internal/facade/v1/perf/snapshot` : l'anneau, pour les poids de + l'optimiseur ; +- `GET /api/perf/summary`. + +**Go** (`facade_perf.go`) : quand la façade tient la conversation, +`perfRecord` range chez elle, et `perfSnapshot` relit son anneau +(optimiseur, contexte de `/internal/v1/tune/context`). `/api/perf/summary` +lui est remis (`facadeChatRoutes`). Façade injoignable : l'anneau du Go +prend le relais. Sans façade, ou la conversation rendue au Go +(`LOKI_CHAT=go`), rien ne change. + +La route `/internal/v1/turn/perf` est retirée : plus personne ne l'appelle. + +## Parité + +`internal/loki/parity_perf_test.go` tire du Go un journal +(`tests/parity_perf/perf.json`). Quatorze complétions y sont rangées, avec +les empreintes de leurs messages ; elles couvrent : + +- prolongement ; +- perte après un sous-agent et un client externe ; +- cache inconnu ; +- divergence ; +- autre discussion ; +- complétion coupée ; +- compaction, bench, vérification. + +Pour chaque complétion, le journal garde ce qui est rangé, la ligne +`[perf]` et l'alerte. Il porte aussi le résumé, le résumé vide, et onze +seuils d'alerte. `src/perfring/tests.rs` le rejoue, avec l'anneau borné et +l'heure posée. + +`tests/lot6.rs` : + +- `cache_perdu_mesure_par_la_facade` : deux tours de la même discussion ; + la perte est mesurée par la façade et signalée dans les stats finales ; +- `anneau_de_la_facade_pour_le_go` : ce que le Go confie (perte après un + client externe), ce qu'il relit, le jeton exigé, `/api/perf/summary` + servi par la façade ; +- les tests du lot 29 lisent désormais l'anneau de la façade. + +`internal/loki/facade_perf_test.go` (`TestFacadePerfAnneau`) : + +- complétion confiée à la façade ; +- anneau relu ; +- repli sur l'anneau du Go quand la façade est en panne ; +- route remise. diff --git a/internal/loki/backend_tune_run.go b/internal/loki/backend_tune_run.go index f3b0343..5d81e71 100644 --- a/internal/loki/backend_tune_run.go +++ b/internal/loki/backend_tune_run.go @@ -552,7 +552,7 @@ func runTune(ctx context.Context, opts tuneOpts, tr *tuneTracker) (out *tuneResu res := &tuneResult{PresetID: preset.ID, PresetName: preset.Name, BaseFP: configFingerprint(cfg), PresetFP: presetFingerprint([]byte(content)), OptIn: opts.OptIn, At: time.Now().Unix(), - Weights: tuneWeightsFrom(perfLog.snapshot())} + Weights: tuneWeightsFrom(perfSnapshot())} // Référence : la ligne composée telle quelle, la sonde de la machine. tr.setPhase("référence : composition de la ligne de commande") diff --git a/internal/loki/facade_chat.go b/internal/loki/facade_chat.go index e4f8d05..4c72ed1 100644 --- a/internal/loki/facade_chat.go +++ b/internal/loki/facade_chat.go @@ -89,6 +89,8 @@ var facadeChatRoutes = map[string]bool{ "/api/chat/export": true, "/api/chat/mode": true, "/api/chat/criteria": true, + // Lot 39 : l'anneau de télémétrie est celui de la façade. + "/api/perf/summary": true, } // withFacadeChat remet la requête à la façade quand elle tient la diff --git a/internal/loki/facade_perf.go b/internal/loki/facade_perf.go new file mode 100644 index 0000000..5067d9e --- /dev/null +++ b/internal/loki/facade_perf.go @@ -0,0 +1,74 @@ +package loki + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "os" +) + +// facade_perf.go — l'anneau de télémétrie tenu par la façade (lot 39 de la +// migration Rust). Quand elle tient la conversation, toutes les complétions y +// sont rangées : +// +// - celles de ses tours, directement ; +// - celles que le Go fait encore (sous-agents, tâches IA, bench, compaction +// et préchauffage de ses tours, clients externes de /v1), par +// /internal/facade/v1/perf/record. +// +// /api/perf/summary est servi par elle (facadeChatRoutes) et l'optimiseur lit +// l'anneau par perfSnapshot. Façade injoignable : la complétion est rangée +// ici, comme sans façade — la télémétrie ne fait jamais échouer un tour. + +// facadePerfJSON appelle l'API de télémétrie de la façade et décode sa +// réponse. +func facadePerfJSON(path string, body, out any) error { + b, err := json.Marshal(body) + if err != nil { + return err + } + req, err := http.NewRequest(http.MethodPost, "http://"+facadeAddr()+path, bytes.NewReader(b)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set(internalTokenHeader, internalToken()) + resp, err := facadeClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("façade : %s", resp.Status) + } + return json.NewDecoder(resp.Body).Decode(out) +} + +// facadePerfRecord range une complétion dans l'anneau de la façade et la rend +// telle que rangée (numéro, perte de cache, ce qui s'est intercalé). +func facadePerfRecord(r perfRec, prefix []uint64) (perfRec, bool) { + var out struct { + Rec perfRec `json:"rec"` + } + err := facadePerfJSON("/internal/facade/v1/perf/record", map[string]any{"rec": r, "prefix": prefix}, &out) + if err != nil { + fmt.Fprintf(os.Stderr, "[perf] complétion non transmise à la façade : %v\n", err) + return perfRec{}, false + } + return out.Rec, true +} + +// perfSnapshot : l'anneau, de la plus ancienne complétion à la plus récente — +// celui de la façade quand elle le tient. +func perfSnapshot() []perfRec { + if facadeChat() { + var out struct { + Recs []perfRec `json:"recs"` + } + if err := facadePerfJSON("/internal/facade/v1/perf/snapshot", map[string]any{}, &out); err == nil { + return out.Recs + } + } + return perfLog.snapshot() +} diff --git a/internal/loki/facade_perf_test.go b/internal/loki/facade_perf_test.go new file mode 100644 index 0000000..51b580e --- /dev/null +++ b/internal/loki/facade_perf_test.go @@ -0,0 +1,47 @@ +package loki + +import ( + "strings" + "testing" +) + +// Lot 39 : quand la façade tient la conversation, les complétions du Go vont +// dans son anneau, et l'optimiseur le lit ; façade en panne, l'anneau du Go +// prend le relais. /api/perf/summary lui est remis. +func TestFacadePerfAnneau(t *testing.T) { + testHome(t) + freshPerf(t) + f := demarrerFacade(t) + f.internalStatus = 200 + f.internalBody = `{"rec":{"at":"2026-10-16T08:00:00Z","seq":7,"kind":"subagent","conv":"c1","iter":0,"complete":true,"total":10,"cached":2,"lost":3,"lost_after":"main"}}` + got := perfRecord(perfRec{Kind: perfSubagent, Conv: "c1", Complete: true, Total: 10, Cached: iptr(2)}, []uint64{1, 2}) + if got.Seq != 7 || got.Lost == nil || *got.Lost != 3 || got.LostAfter != "main" { + t.Fatalf("rangée par la façade : %+v", got) + } + if n := len(perfLog.snapshot()); n != 0 { + t.Fatalf("rangée aussi ici : %d", n) + } + vu := f.recu() + if len(vu) != 1 || !strings.HasPrefix(vu[0], "POST /internal/facade/v1/perf/record jeton ") || + !strings.Contains(vu[0], `"prefix":[1,2]`) || !strings.Contains(vu[0], `"kind":"subagent"`) { + t.Fatalf("envoi : %v", vu) + } + + f.internalBody = `{"recs":[{"at":"2026-10-16T08:00:00Z","seq":1,"kind":"main","iter":0,"complete":true,"total":40}]}` + if s := perfSnapshot(); len(s) != 1 || s[0].Total != 40 { + t.Fatalf("anneau de la façade : %+v", s) + } + + // Façade en panne : l'anneau d'ici. + f.internalStatus = 500 + got = perfRecord(perfRec{Kind: perfForeign}, nil) + if got.Seq != 1 || got.Kind != perfForeign { + t.Fatalf("repli : %+v", got) + } + if s := perfSnapshot(); len(s) != 1 || s[0].Kind != perfForeign { + t.Fatalf("repli de l'anneau : %+v", s) + } + if !facadeChatRoutes["/api/perf/summary"] { + t.Fatal("/api/perf/summary n'est pas remis à la façade") + } +} diff --git a/internal/loki/facade_tune.go b/internal/loki/facade_tune.go index 50e38a7..14bb01e 100644 --- a/internal/loki/facade_tune.go +++ b/internal/loki/facade_tune.go @@ -66,5 +66,5 @@ func handleInternalTuneContext(w http.ResponseWriter, r *http.Request) { bgJobsRunning()) } sendJSON(w, http.StatusOK, map[string]any{"busy": busy, "main_active": engineNeedsStop(), - "perf": perfLog.snapshot()}) + "perf": perfSnapshot()}) } diff --git a/internal/loki/parity_perf_test.go b/internal/loki/parity_perf_test.go new file mode 100644 index 0000000..0a37936 --- /dev/null +++ b/internal/loki/parity_perf_test.go @@ -0,0 +1,142 @@ +//go:build linux + +package loki + +import ( + "bytes" + "encoding/json" + "os" + "path/filepath" + "strconv" + "testing" + "time" +) + +// parity_perf_test.go — journal de référence de l'anneau de télémétrie (lot 39 +// de la migration Rust) : une suite de complétions rangées (chacune avec les +// empreintes des messages envoyés), ce que l'anneau en a tiré (numéro, perte +// de cache, ce qui s'est intercalé), la ligne [perf], le résumé de +// /api/perf/summary, puis les seuils d'alerte d'une série de presets. Rangé +// dans rust/loki-rs/tests/parity_perf/perf.json ; src/perfring/tests.rs le +// rejoue. +// +// LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteAnneauPerf + +const parityPerfFile = "../../rust/loki-rs/tests/parity_perf/perf.json" + +type parityPerfStep struct { + Rec json.RawMessage `json:"rec"` + Msgs []string `json:"msgs,omitempty"` // contenus des messages envoyés (rôle user) + Prefix []string `json:"prefix,omitempty"` + Out json.RawMessage `json:"out"` + Line string `json:"line"` + Alert int `json:"alert_1000"` // perfAlert au seuil 1000 : 1 vrai, 0 faux +} + +type parityPerfAlert struct { + Cfg map[string]string `json:"cfg"` + Env string `json:"env,omitempty"` // LLAMA_ARG_CHECKPOINT_MIN_SPACING_NT + Want int `json:"want"` +} + +type parityPerfJournal struct { + Steps []parityPerfStep `json:"steps"` + Summary json.RawMessage `json:"summary"` + Empty json.RawMessage `json:"empty"` + Alerts []parityPerfAlert `json:"alerts"` +} + +func TestPariteAnneauPerf(t *testing.T) { + testHome(t) + freshPerf(t) + at := time.Date(2026, 10, 16, 8, 0, 0, 0, time.UTC) + var j parityPerfJournal + rec := func(r perfRec, msgs ...string) { + var prefix []uint64 + if msgs != nil { + ms := make([]Message, len(msgs)) + for i, c := range msgs { + ms[i] = Message{Role: "user", Content: c} + } + prefix = perfPrefix(ms) + } + in, _ := json.Marshal(r) + r.At = at + at = at.Add(time.Second) + got := perfLog.record(r, prefix) + out, _ := json.Marshal(got) + st := parityPerfStep{Rec: in, Msgs: msgs, Out: out, Line: perfLine(got)} + for _, p := range prefix { + st.Prefix = append(st.Prefix, strconv.FormatUint(p, 10)) + } + if perfAlert(got, 1000) { + st.Alert = 1 + } + j.Steps = append(j.Steps, st) + } + main := func(conv string, iter, total int, cached *int) perfRec { + return perfRec{Kind: perfMain, Conv: conv, Iter: iter, Complete: true, Total: total, Cached: cached} + } + full := perfRec{Kind: perfMain, Conv: "c1", Iter: 0, Complete: true, Total: 4000, Cached: iptr(3000), + New: 1000, PPms: 2000.4, PPtps: 499.9, Gen: 50, TGtps: 40.25, DraftN: iptr(40), DraftAcc: iptr(30), TTFTms: ptr64(2100)} + rec(full, "a") + rec(main("c1", 1, 4600, iptr(4000)), "a", "b", "c") // prolonge : rien de perdu + rec(perfRec{Kind: perfSubagent, Conv: "c1", Complete: true, Total: 300, Cached: iptr(0), New: 300, PPms: 600, PPtps: 500}, "z") + rec(perfRec{Kind: perfForeign}) // client externe : vu, pas mesuré + rec(main("c1", 2, 5200, iptr(200)), "a", "b", "c", "d") // perte après le sous-agent et le client + rec(main("c1", 3, 5200, nil), "a", "b", "c", "d", "e") // cache inconnu + rec(main("c1", 4, 5300, iptr(10)), "x", "b") // diverge + rec(main("c2", 0, 20000, iptr(0)), "q") // autre discussion, gros prompt sans cache + rec(perfRec{Kind: perfMain, Conv: "c2", Iter: 1, Cached: iptr(0)}, "q", "r") // complétion coupée + rec(main("c2", 1, 21000, iptr(19000)), "q", "r") // prolonge malgré la coupure + rec(perfRec{Kind: perfCompact, Conv: "c2", Complete: true, Total: 9000, Cached: iptr(8800), New: 200, PPms: 100, PPtps: 2000, Gen: 300, TGtps: 30, TTFTms: ptr64(900)}) + rec(perfRec{Kind: perfBench, Complete: true, Total: 33000, New: 33000, PPms: 30000, PPtps: 1100, Gen: 64, TGtps: 20.5}) + rec(perfRec{Kind: perfVerify, Conv: "c1", Iter: 0, Complete: true, Total: 12000, Cached: iptr(0), New: 12000, PPms: 9000.5, PPtps: 1333.3, Gen: 2, TGtps: 9.75, DraftN: iptr(0)}) + rec(main("c1", 0, 6000, iptr(5900)), "a", "b", "c", "d", "e", "f") // prolonge la dernière terminée de c1 + + sum, _ := json.Marshal(perfSummarize(perfLog.snapshot())) + j.Summary = sum + empty, _ := json.Marshal(perfSummarize(nil)) + j.Empty = empty + + for _, c := range []parityPerfAlert{ + {Cfg: map[string]string{}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "2048"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "4096", "UBATCH": "1024"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "beaucoup"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": " 300 ", "UBATCH": "100"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "2048", "EXTRA_ARGS": "--checkpoint-min-step 8192 -ub 256"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "2048", "UBATCH": "1024", "EXTRA_ARGS": "-ub=2048"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "-5", "UBATCH": "0"}}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "2048"}, Env: "3000"}, + {Cfg: map[string]string{"CKPT_MIN_STEP": "2048"}, Env: "rien"}, + {Cfg: map[string]string{"MODEL": "absent.gguf"}}, + } { + t.Setenv("LLAMA_ARG_CHECKPOINT_MIN_SPACING_NT", c.Env) + c.Want = perfLostAlertAt(c.Cfg) + j.Alerts = append(j.Alerts, c) + } + + got, err := json.MarshalIndent(j, "", " ") + if err != nil { + t.Fatal(err) + } + got = append(got, '\n') + if os.Getenv("LOKI_PARITY_UPDATE") == "1" { + if err := os.MkdirAll(filepath.Dir(parityPerfFile), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(parityPerfFile, got, 0o644); err != nil { + t.Fatal(err) + } + return + } + want, err := os.ReadFile(parityPerfFile) + if err != nil { + t.Fatalf("journal de référence absent (%v) : LOKI_PARITY_UPDATE=1 pour le créer", err) + } + if !bytes.Equal(got, want) { + t.Errorf("l'anneau de télémétrie ne produit plus le journal rangé dans %s.\n"+ + "LOKI_PARITY_UPDATE=1 pour le régénérer si le changement est voulu.", parityPerfFile) + } +} diff --git a/internal/loki/perf_log.go b/internal/loki/perf_log.go index 36aee20..0e3e63c 100644 --- a/internal/loki/perf_log.go +++ b/internal/loki/perf_log.go @@ -343,7 +343,15 @@ func (p *perfStore) snapshot() []perfRec { // perfRecord range une complétion et, si LOKI_PERF_LOG est posée, l'écrit sur // stderr. Une ligne par complétion, c'est des dizaines par tour d'agent : pas // par défaut, les journaux Docker ont déjà celles de llama-server. +// +// Quand la façade tient la conversation, l'anneau est le sien (facade_perf.go) : +// elle range et écrit la ligne. func perfRecord(r perfRec, prefix []uint64) perfRec { + if facadeChat() { + if got, ok := facadePerfRecord(r, prefix); ok { + return got + } + } r = perfLog.record(r, prefix) if perfLogOn() { fmt.Fprintln(os.Stderr, perfLine(r)) diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index 70a3672..ce5a857 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -71,8 +71,6 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalOAIEnd(w, r) case "/internal/v1/oai/served": handleInternalOAIServed(w, r) - case "/internal/v1/turn/perf": - handleInternalTurnPerf(w, r) case "/internal/v1/turn/projsnap/refresh": handleInternalTurnProjSnapRefresh(w, r) case "/internal/v1/turn/slot": diff --git a/internal/loki/web_internal_turnperf.go b/internal/loki/web_internal_turnperf.go deleted file mode 100644 index afa14bf..0000000 --- a/internal/loki/web_internal_turnperf.go +++ /dev/null @@ -1,64 +0,0 @@ -package loki - -import ( - "encoding/json" - "io" - "net/http" -) - -// Télémétrie d'une complétion d'un tour tenu par la façade Rust -// (LOKI_CHAT=rust). La boucle Go range chaque complétion dans l'anneau de -// perf_log.go (perfRecord) : c'est lui qui nourrit /api/perf/summary et -// l'alerte « cache perdu » posée sur les stats finales d'une complétion. La -// boucle Rust lui envoie la même chose, ici, au même moment. -// -// POST /internal/v1/turn/perf -// {"kind":"main","conv":"c…","iter":0,"complete":true,"external":false, -// "stats":{…StatsEvent…},"prefix":[…empreintes…]} -// → {"kind":"main","lost":N,"lost_after":"…","lost_alert":true} -// -// kind : main (tour), verify (passe de vérification du mode Code) ou compact -// (résumé de compaction, relevé dans la réponse entière par la façade). -// prefix : empreintes cumulées des messages envoyés (perfPrefix), calculées -// par l'appelant — elles n'ont à être cohérentes qu'avec ses propres envois, -// d'une complétion à la suivante de la même discussion. La sonde du gabarit -// n'en part pas : la façade la tient (facadeTplCaps). - -const internalTurnPerfMaxBody = 8 << 20 - -func handleInternalTurnPerf(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - w.Header().Set("Allow", http.MethodPost) - sendJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "POST attendu"}) - return - } - var in struct { - Kind string `json:"kind"` - Conv string `json:"conv"` - Iter int `json:"iter"` - Complete bool `json:"complete"` - External bool `json:"external"` - Stats StatsEvent `json:"stats"` - Prefix []uint64 `json:"prefix"` - } - raw, err := io.ReadAll(io.LimitReader(r.Body, internalTurnPerfMaxBody+1)) - if err != nil || len(raw) > internalTurnPerfMaxBody || json.Unmarshal(raw, &in) != nil { - sendJSON(w, http.StatusBadRequest, map[string]any{"error": "télémétrie illisible"}) - return - } - if in.Kind == "" { - in.Kind = perfMain - } - var prefix []uint64 - if in.Complete { - prefix = in.Prefix - } - rec := perfRecord(perfRecFromStats(perfTag{kind: in.Kind, conv: in.Conv}, in.Iter, in.Complete, in.Stats), prefix) - alert := false - if in.Complete && !in.External && ((rec.Lost != nil && *rec.Lost > 0) || (rec.Cached != nil && *rec.Cached == 0)) { - alert = perfAlert(rec, perfLostAlertAt(ReadConfig())) - } - sendJSON(w, http.StatusOK, map[string]any{ - "kind": rec.Kind, "lost": rec.Lost, "lost_after": rec.LostAfter, "lost_alert": alert, - }) -} diff --git a/internal/loki/web_internal_turnperf_test.go b/internal/loki/web_internal_turnperf_test.go deleted file mode 100644 index 43a9912..0000000 --- a/internal/loki/web_internal_turnperf_test.go +++ /dev/null @@ -1,124 +0,0 @@ -package loki - -import ( - "encoding/json" - "fmt" - "testing" -) - -// La complétion d'un tour tenu par Rust arrive dans l'anneau de télémétrie, -// comme celle de la boucle Go : /api/perf/summary la compte. -func TestTurnPerfDansLAnneau(t *testing.T) { - testHome(t) - t.Setenv(internalTokenEnv, jetonTest) - avant := perfSummarize(perfLog.snapshot()) - n0 := 0 - if k := avant.Kinds[perfMain]; k != nil { - n0 = k.Completions - } - body := `{"kind":"main","conv":"c-perf","iter":0,"complete":true,"external":false, - "stats":{"prompt_tokens":120,"prompt_tokens_total":120,"gen_tokens":8,"cache_tokens":0}, - "prefix":[1,2,3]}` - w := appelOAIInterne(t, "POST", "/internal/v1/turn/perf", body) - if w.Code != 200 { - t.Fatalf("code %d : %s", w.Code, w.Body.String()) - } - var got map[string]any - if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got["kind"] != "main" { - t.Fatalf("réponse : %s", w.Body.String()) - } - apres := perfSummarize(perfLog.snapshot()) - if k := apres.Kinds[perfMain]; k == nil || k.Completions != n0+1 { - t.Fatalf("complétion non rangée : %+v", apres.Kinds[perfMain]) - } - if w := appelOAIInterne(t, "POST", "/internal/v1/turn/perf", "{"); w.Code != 400 { - t.Fatalf("corps illisible : %d", w.Code) - } -} - -// Perte de cache calculée contre la complétion précédente de la même -// discussion, ce qui s'est intercalé, alerte au seuil du preset actif ; un -// preset externe ou une complétion incomplète ne la signalent jamais. -func TestTurnPerfPerteEtAlerte(t *testing.T) { - testHome(t) - t.Setenv(internalTokenEnv, jetonTest) - post := func(body string) (int, map[string]any) { - w := appelOAIInterne(t, "POST", "/internal/v1/turn/perf", body) - var out map[string]any - _ = json.Unmarshal(w.Body.Bytes(), &out) - return w.Code, out - } - code, out := post(`{"kind":"main","conv":"fp1","iter":0,"complete":true, - "stats":{"prompt_tokens":5000,"prompt_tokens_total":5000,"cache_tokens":0},"prefix":[11,22]}`) - if code != 200 || out["lost"] != nil || out["lost_after"] != "" { - t.Fatalf("première : %d %v", code, out) - } - // Une tâche passe entre deux. - perfRecord(perfRec{Kind: perfTask, Conv: "t", Complete: true, Total: 10}, nil) - // La suivante prolonge la première, mais n'en reprend que 100 jetons. - code, out = post(`{"kind":"main","conv":"fp1","iter":1,"complete":true, - "stats":{"prompt_tokens":5900,"prompt_tokens_total":6000,"cache_tokens":100},"prefix":[11,22,33]}`) - if code != 200 || out["lost"] != float64(4900) || out["lost_after"] != perfTask || out["lost_alert"] != true { - t.Fatalf("perte : %d %v", code, out) - } - if recs := perfLog.snapshot(); recs[len(recs)-1].Conv != "fp1" || recs[len(recs)-1].Iter != 1 || recs[len(recs)-1].New != 5900 { - t.Fatalf("relevé rangé : %+v", recs[len(recs)-1]) - } - // Preset externe : la perte est rangée, jamais signalée. - code, out = post(`{"kind":"main","conv":"fp1","iter":2,"complete":true,"external":true, - "stats":{"prompt_tokens_total":7000,"cache_tokens":100},"prefix":[11,22,33,44]}`) - if code != 200 || out["lost"] != float64(5900) || out["lost_alert"] != false { - t.Fatalf("externe : %d %v", code, out) - } - // Incomplète : ses empreintes sont ignorées, ni perte ni alerte. - code, out = post(`{"kind":"main","conv":"fp1","iter":3,"complete":false, - "stats":{"cache_tokens":0},"prefix":[11,22,33,44,55]}`) - if code != 200 || out["lost"] != nil || out["lost_alert"] != false { - t.Fatalf("incomplète : %d %v", code, out) - } - if recs := perfLog.snapshot(); recs[len(recs)-1].Complete { - t.Fatalf("incomplète rangée complète : %+v", recs[len(recs)-1]) - } - // Résumé de compaction : sa nature est gardée. - code, out = post(`{"kind":"compact","conv":"fp1","complete":true,"stats":{"prompt_tokens_total":900}}`) - if code != 200 || out["kind"] != perfCompact { - t.Fatalf("compaction : %d %v", code, out) - } - // Sans nature : un tour. - if _, out = post(`{"conv":"fp2","complete":true,"stats":{}}`); out["kind"] != perfMain { - t.Fatalf("sans nature : %v", out) - } - if w := appelOAIInterne(t, "GET", "/internal/v1/turn/perf", ""); w.Code != 405 { - t.Fatalf("GET : %d", w.Code) - } -} - -// Une étape de tour acceptée par le moteur : le slot porte la discussion -// (engineSlotHolds), comme engineMarkMain dans la boucle Go. -func TestOAIServedMarqueLeSlot(t *testing.T) { - testHome(t) - t.Setenv(internalTokenEnv, jetonTest) - w := appelOAIInterne(t, "POST", "/internal/v1/oai/begin", `{"main_conv":"c-slot"}`) - var b struct { - Ticket uint64 `json:"ticket"` - } - if err := json.Unmarshal(w.Body.Bytes(), &b); err != nil || b.Ticket == 0 { - t.Fatalf("begin : %s", w.Body.String()) - } - if engineSlotHolds("c-slot") { - t.Fatal("slot marqué avant que le moteur ait accepté la requête") - } - tk := fmt.Sprintf(`{"ticket":%d}`, b.Ticket) - if w := appelOAIInterne(t, "POST", "/internal/v1/oai/served", tk); w.Code != 204 { - t.Fatalf("served : %d", w.Code) - } - if !engineSlotHolds("c-slot") { - t.Fatal("slot non marqué après served") - } - if w := appelOAIInterne(t, "POST", "/internal/v1/oai/end", tk); w.Code != 204 { - t.Fatalf("end : %d", w.Code) - } - if w := appelOAIInterne(t, "POST", "/internal/v1/oai/served", tk); w.Code != 404 { - t.Fatalf("served après end : %d", w.Code) - } -} diff --git a/rust/loki-rs/src/chat/generate.rs b/rust/loki-rs/src/chat/generate.rs index e72974f..e1b8bcb 100644 --- a/rust/loki-rs/src/chat/generate.rs +++ b/rust/loki-rs/src/chat/generate.rs @@ -29,8 +29,8 @@ //! Les images gardées (KEEP_TURN_IMAGES, lot 32) //! partent en début et en fin de tour quand il le faut //! ([`ChatService::kept_images_turn_start`]). La télémétrie des complétions -//! (tours, vérifications, résumés) rejoint l'anneau du Go -//! (`/internal/v1/turn/perf`) ; les motifs `retry_patterns` personnalisés +//! (tours, vérifications, résumés) rejoint l'anneau tenu ici +//! ([`crate::perfring`], lot 39) ; les motifs `retry_patterns` personnalisés //! sont lus au démarrage ([`retry_patterns`]). Chaque requête au moteur local //! part sur le slot que rend le Go (SIDE_SLOT, rechargement SLOT_PERSIST, //! lot 31). @@ -702,7 +702,11 @@ impl ChatService { cfg: cfg.clone(), patterns: self.shared.patterns.clone(), port, - go: Some((store.clone(), self.sessions.conv().id())), + go: Some(( + store.clone(), + self.sessions.conv().id(), + self.shared.perf.clone(), + )), }; TurnEnv { cfg, @@ -1296,6 +1300,13 @@ impl ChatShared { store: crate::store::Store, img_dir: std::path::PathBuf, ) -> Self { + // LOKI_HOME : le dossier des images en est un sous-dossier. + let perf = std::sync::Arc::new(crate::perfring::PerfRing::new( + img_dir + .parent() + .map(|p| p.to_path_buf()) + .unwrap_or_default(), + )); ChatShared { http: EngineHttp::new(), store, @@ -1307,6 +1318,7 @@ impl ChatShared { patterns: super::stream::RetryPatterns::builtin(), tpl: std::sync::Arc::new(super::tplprobe::TplProbe::new()), prewarm: Default::default(), + perf, mcp: None, tasks: None, go_version: Default::default(), diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs index 45a8c17..25ab5c5 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -179,9 +179,10 @@ pub struct EngineSummarizer { pub cfg: HashMap, pub patterns: Arc, pub port: u16, - /// Le Go et la discussion des résumés : télémétrie (kind=compact), - /// requêtes en vol et slot du moteur. `None` sans base (tests). - pub go: Option<(Store, String)>, + /// Le Go et la discussion des résumés (requêtes en vol, slot du + /// moteur), et l'anneau de télémétrie (kind=compact). `None` sans base + /// (tests). + pub go: Option<(Store, String, Arc)>, } impl EngineSummarizer { @@ -191,7 +192,7 @@ impl EngineSummarizer { async fn post(&self, mut payload: Value, main: bool) -> Result { let mut _gate = None; if !self.ep.external - && let Some((store, _)) = &self.go + && let Some((store, _, _)) = &self.go { let slot = engine_slot_for(store, "", main).await; if let Some(n) = slot @@ -235,7 +236,7 @@ impl EngineSummarizer { Err(SummaryError::Other(e)) => e == super::compact::ERR_EMPTY_SUMMARY, Err(_) => false, }; - if decoded && let Some((store, conv)) = &self.go { + if decoded && let Some((store, conv, ring)) = &self.go { let (complete, stats) = super::perf::wire_stats(&raw); let rec = PerfRec { complete, @@ -243,7 +244,7 @@ impl EngineSummarizer { stats, ..Default::default() }; - super::perf::record(store, super::perf::KIND_COMPACT, conv, &rec).await; + super::perf::record(store, ring, super::perf::KIND_COMPACT, conv, &rec).await; } out } @@ -333,6 +334,9 @@ pub struct ChatShared { pub tpl: Arc, /// Préchauffage (PREWARM) : celui en vol, capacités des discussions. pub prewarm: super::prewarm::Prewarm, + /// L'anneau de télémétrie des complétions (lot 39) : celles de la + /// façade, et celles que le Go lui confie. + pub perf: Arc, /// Le pool MCP, quand la façade le tient (LOKI_MCP=rust) : les appels /// `mcp__*` du tour y vont directement (lot 36). pub mcp: Option>, @@ -935,7 +939,14 @@ impl TurnHost for GoTurnHost { } async fn perf_record(&self, rec: PerfRec) -> PerfMarks { - super::perf::record(&self.shared.store, self.perf_kind, &self.conv_id, &rec).await + super::perf::record( + &self.shared.store, + &self.shared.perf, + self.perf_kind, + &self.conv_id, + &rec, + ) + .await } fn preflight(&self) -> Option { diff --git a/rust/loki-rs/src/chat/perf.rs b/rust/loki-rs/src/chat/perf.rs index 7db38a9..3c63b41 100644 --- a/rust/loki-rs/src/chat/perf.rs +++ b/rust/loki-rs/src/chat/perf.rs @@ -1,19 +1,19 @@ -//! Télémétrie des complétions de la façade (lot 29). Toutes rejoignent -//! l'anneau du Go (`perf_log.go`) par `/internal/v1/turn/perf` : le Go les -//! range, calcule la perte de cache contre la précédente de la même -//! discussion et l'alerte au seuil du preset actif. `/api/perf/summary` -//! reste au Go. +//! Télémétrie des complétions de la façade (lots 29 et 39). Toutes +//! rejoignent l'anneau tenu ici ([`crate::perfring`]) : il les range, +//! calcule la perte de cache contre la précédente de la même discussion et +//! l'alerte au seuil du preset actif ; il nourrit `/api/perf/summary`. //! //! Ce module porte ce que la façade relève elle-même : la réponse entière -//! d'un résumé de compaction (`perfRecFromWire`), et l'envoi au Go. Des -//! identifiants et des compteurs, jamais de texte. Rien ne retourne au -//! moteur. +//! d'un résumé de compaction (`perfRecFromWire`), et le passage des stats +//! d'une complétion à l'anneau (`perfRecFromStats`). Des identifiants et +//! des compteurs, jamais de texte. Rien ne retourne au moteur. -use serde_json::{Value, json}; +use serde_json::Value; use super::turn::{PerfMarks, PerfRec}; use super::types::StatsEvent; -use crate::store::Store; +use crate::perfring::{PerfRing, perf_alert}; +use crate::store::{BK_STATE, Store}; /// Nature d'une complétion. pub const KIND_MAIN: &str = "main"; @@ -75,32 +75,56 @@ fn perf_num(v: Option<&Value>) -> Option { (f.is_finite() && (0.0..1e12).contains(&f)).then_some(f) } -/// Range la complétion dans l'anneau du Go et rend ce qu'il en tire. Le Go -/// injoignable : rien (la télémétrie ne fait jamais échouer un tour). -pub async fn record(store: &Store, kind: &str, conv: &str, rec: &PerfRec) -> PerfMarks { - let body = json!({ - "kind": kind, - "conv": conv, - "iter": rec.iter, - "complete": rec.complete, - "external": rec.external, - "stats": rec.stats, - "prefix": rec.prefix, - }); - match store - .internal_json(hyper::Method::POST, "/internal/v1/turn/perf", Some(&body)) - .await - { - Ok(v) => PerfMarks { - kind: v["kind"].as_str().unwrap_or("").to_string(), - lost: v["lost"].as_i64(), - lost_after: v["lost_after"].as_str().unwrap_or("").to_string(), - lost_alert: v["lost_alert"].as_bool().unwrap_or(false), - }, - Err(e) => { - eprintln!("[loki-rs] télémétrie : {e}"); - PerfMarks::default() - } +/// Range la complétion dans l'anneau et rend ce qu'il en tire : la perte +/// de cache, ce qui s'est intercalé, et l'alerte au seuil du preset actif +/// (jamais sur une API externe). La télémétrie ne fait jamais échouer un +/// tour. +pub async fn record( + store: &Store, + ring: &PerfRing, + kind: &str, + conv: &str, + rec: &PerfRec, +) -> PerfMarks { + let kind = if kind.is_empty() { KIND_MAIN } else { kind }; + let s = &rec.stats; + let r = crate::perfring::PerfRec { + kind: kind.to_string(), + conv: conv.to_string(), + iter: rec.iter, + complete: rec.complete, + total: s.prompt_tokens_total, + cached: s.cache_tokens, + new: s.prompt_tokens, + pp_ms: s.prompt_ms, + pp_tps: s.prompt_per_second, + gen_tokens: s.gen_tokens, + tg_tps: s.gen_per_second, + draft_n: s.draft_n, + draft_accepted: s.draft_accepted, + ttft_ms: s.ttft_ms, + ..Default::default() + }; + let prefix: &[u64] = if rec.complete { &rec.prefix } else { &[] }; + let got = ring.record(r, prefix); + let mut alert = false; + if rec.complete && !rec.external && (got.lost.is_some_and(|l| l > 0) || got.cached == Some(0)) { + let cfg = store.config().await; + let stored = crate::gguf::parse_stored_model_dirs( + store + .get(BK_STATE, crate::gguf::STATE_MODEL_DIRS_KEY) + .await + .ok() + .flatten() + .as_deref(), + ); + alert = perf_alert(&got, ring.lost_alert_at(&cfg, &stored)); + } + PerfMarks { + kind: got.kind, + lost: got.lost, + lost_after: got.lost_after, + lost_alert: alert, } } @@ -109,6 +133,7 @@ mod tests { use super::super::turn::perf_prefix; use super::super::types::Message; use super::*; + use serde_json::json; #[test] fn releve_d_une_reponse_entiere() { diff --git a/rust/loki-rs/src/chat/prewarm.rs b/rust/loki-rs/src/chat/prewarm.rs index 65f35be..b0cec26 100644 --- a/rust/loki-rs/src/chat/prewarm.rs +++ b/rust/loki-rs/src/chat/prewarm.rs @@ -500,7 +500,8 @@ impl ChatService { stats, ..Default::default() }; - perf::record(store, perf::KIND_PREWARM, &conv_id, &rec).await; + perf::record(store, &self.shared.perf, perf::KIND_PREWARM, &conv_id, &rec) + .await; } None => {} } diff --git a/rust/loki-rs/src/chatapi.rs b/rust/loki-rs/src/chatapi.rs index b3c6eec..3bd7833 100644 --- a/rust/loki-rs/src/chatapi.rs +++ b/rust/loki-rs/src/chatapi.rs @@ -51,6 +51,7 @@ pub fn handles(path: &str) -> bool { | "/api/conversations/switch" | "/api/conversations/rename" | "/api/conversations/delete" + | "/api/perf/summary" ) } @@ -68,6 +69,8 @@ pub fn is_internal(path: &str) -> bool { | "/internal/facade/v1/chat/tpl" | "/internal/facade/v1/chat/prewarm" | "/internal/facade/v1/chat/prewarm/cancel" + | "/internal/facade/v1/perf/record" + | "/internal/facade/v1/perf/snapshot" ) } @@ -109,6 +112,34 @@ pub async fn serve_internal(chat: &Arc, req: Request) -> None => send_json(StatusCode::OK, &json!({ "ok": false })), } } + // Télémétrie (lot 39) : une complétion faite par le Go (sous-agent, + // tâche IA, bench, compaction de ses tours, client externe de /v1), + // rangée dans l'anneau tenu ici ; rendue telle que rangée (numéro, + // perte de cache, ce qui s'est intercalé). + "/internal/facade/v1/perf/record" => match read_json(req.into_body()).await { + Ok(v) => { + let rec: crate::perfring::PerfRec = + serde_json::from_value(v.get("rec").cloned().unwrap_or_default()) + .unwrap_or_default(); + let prefix: Vec = v + .get("prefix") + .and_then(Value::as_array) + .map(|a| a.iter().filter_map(Value::as_u64).collect()) + .unwrap_or_default(); + let got = chat.shared.perf.record(rec, &prefix); + send_json(StatusCode::OK, &json!({ "rec": got })) + } + Err(_) => send_json( + StatusCode::BAD_REQUEST, + &json!({ "error": "télémétrie illisible" }), + ), + }, + // L'anneau, de la plus ancienne à la plus récente (poids de + // l'optimiseur). + "/internal/facade/v1/perf/snapshot" => send_json( + StatusCode::OK, + &json!({ "recs": chat.shared.perf.snapshot() }), + ), // PREWARM (lot 33) : déclencheur venu du Go (fin d'une tâche // planifiée), annulation relayée (une autre requête est partie). "/internal/facade/v1/chat/prewarm" => { @@ -202,6 +233,23 @@ pub async fn serve( } let path = parts.uri.path().to_string(); match path.as_str() { + // `handlePerfSummary` : des agrégats seulement, aucun texte. + "/api/perf/summary" => { + if parts.method != Method::GET { + return send_json( + StatusCode::METHOD_NOT_ALLOWED, + &json!({ "error": "GET seulement" }), + ); + } + let mut s = crate::perfring::summarize(&chat.shared.perf.snapshot()); + let cfg = chat.shared.store.config().await; + s.template = chat + .shared + .tpl + .caps_current(crate::engstatus::is_external(&cfg)) + .map(|r| r.to_json()); + send_json(StatusCode::OK, &json!(s)) + } "/api/chat" => match read_json(body).await { Ok(v) => subscribe(chat, &v), Err(e) => plain(StatusCode::BAD_REQUEST, &e), diff --git a/rust/loki-rs/src/lib.rs b/rust/loki-rs/src/lib.rs index a4bf833..4d14062 100644 --- a/rust/loki-rs/src/lib.rs +++ b/rust/loki-rs/src/lib.rs @@ -33,6 +33,7 @@ pub mod memory; pub mod models; pub mod oai; pub mod paths; +pub mod perfring; pub mod presets; pub mod presets_api; pub mod projects_api; diff --git a/rust/loki-rs/src/perfring.rs b/rust/loki-rs/src/perfring.rs new file mode 100644 index 0000000..3f8d482 --- /dev/null +++ b/rust/loki-rs/src/perfring.rs @@ -0,0 +1,545 @@ +//! L'anneau de télémétrie des complétions (lot 39) — port de `perf_log.go`, +//! tenu par la façade quand elle tient la conversation. +//! +//! Chaque complétion qui passe par le moteur y est rangée : celles des tours +//! de la façade directement, celles que le Go fait encore (sous-agents, +//! tâches IA, bench, compaction de ses tours, clients externes de /v1) par +//! `/internal/facade/v1/perf/record`. L'anneau calcule la perte de cache +//! d'une complétion contre la précédente de la même discussion et de la même +//! nature, nomme ce qui s'est intercalé, et nourrit `/api/perf/summary` et +//! les poids de l'optimiseur. Des identifiants et des compteurs, jamais de +//! texte. + +use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::path::PathBuf; +use std::sync::Mutex; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::bench::ser_go_f64; + +/// `perfRingMax` : ~1 Mio en mémoire au pire. +pub const RING_MAX: usize = 5000; +/// `perfConvMax` : conversations suivies pour le calcul de la perte. +pub const CONV_MAX: usize = 128; +/// `perfBetween` : entrées remontées au plus pour nommer ce qui s'est +/// intercalé. +pub const BETWEEN: usize = 500; + +fn is_zero_i(v: &i64) -> bool { + *v == 0 +} + +fn is_zero_f(v: &f64) -> bool { + *v == 0.0 +} + +/// `perfRec` : une complétion (mêmes champs JSON que le Go). +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct PerfRec { + /// `time.Time` du Go (RFC 3339) ; vide ou nul : posé à l'enregistrement. + #[serde(default)] + pub at: String, + #[serde(default)] + pub seq: u64, + #[serde(default)] + pub kind: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub conv: String, + #[serde(default)] + pub iter: i64, + #[serde(default)] + pub complete: bool, + /// usage.prompt_tokens : la taille du prompt entier. + #[serde(default, skip_serializing_if = "is_zero_i")] + pub total: i64, + /// cache_n ; absent si inconnu. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cached: Option, + /// prompt_n : les jetons réellement recalculés. + #[serde(default, skip_serializing_if = "is_zero_i")] + pub new: i64, + #[serde( + default, + skip_serializing_if = "is_zero_f", + serialize_with = "ser_go_f64" + )] + pub pp_ms: f64, + #[serde( + default, + skip_serializing_if = "is_zero_f", + serialize_with = "ser_go_f64" + )] + pub pp_tps: f64, + #[serde(default, rename = "gen", skip_serializing_if = "is_zero_i")] + pub gen_tokens: i64, + #[serde( + default, + skip_serializing_if = "is_zero_f", + serialize_with = "ser_go_f64" + )] + pub tg_tps: f64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub draft_n: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub draft_accepted: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ttft_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lost: Option, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub lost_after: String, +} + +/// `time.Time` nul du Go, tel que l'encode son JSON. +const GO_ZERO_TIME: &str = "0001-01-01T00:00:00Z"; + +/// `perfConvState` : la dernière complétion terminée d'une conversation (et +/// d'une nature). +#[derive(Clone, Copy, Debug)] +struct ConvState { + seq: u64, + n: usize, + hash: u64, + total: i64, +} + +#[derive(Default)] +struct Inner { + seq: u64, + ring: Vec, + /// Prochaine case à écraser quand l'anneau est plein. + head: usize, + convs: HashMap, +} + +/// `perfStore`. +pub struct PerfRing { + inner: Mutex, + /// LOKI_HOME : le modèle du preset, pour le seuil d'alerte d'un hybride. + pub home: PathBuf, + /// L'heure (RFC 3339) — remplaçable en test. + pub clock: fn() -> String, +} + +/// Maintenant, comme `time.Now()` encodé par le Go. +pub fn now_rfc3339() -> String { + crate::chat::tplprobe::rfc3339(std::time::SystemTime::now()) +} + +impl PerfRing { + pub fn new(home: PathBuf) -> Self { + PerfRing { + inner: Mutex::new(Inner::default()), + home, + clock: now_rfc3339, + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, Inner> { + self.inner.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// `record` : range la complétion et calcule sa perte de cache + /// + /// `lost = max(0, total précédent − cache_n)` + /// + /// seulement quand le cache est connu, que la complétion est allée au + /// bout, et que ses messages PROLONGENT strictement ceux de la précédente + /// complétion de la même conversation et de la même nature. Ailleurs, la + /// perte reste inconnue plutôt que fausse. `prefix` : empreintes cumulées + /// des messages envoyés (`perfPrefix`). + pub fn record(&self, mut r: PerfRec, prefix: &[u64]) -> PerfRec { + let mut p = self.lock(); + p.seq += 1; + r.seq = p.seq; + if r.at.is_empty() || r.at == GO_ZERO_TIME { + r.at = (self.clock)(); + } + let key = format!("{}\0{}", r.kind, r.conv); + if r.complete && !prefix.is_empty() { + if let Some(prev) = p.convs.get(&key).copied() + && let Some(cached) = r.cached + && prev.total > 0 + && prev.n > 0 + && prefix.len() > prev.n + && prefix[prev.n - 1] == prev.hash + { + let lost = (prev.total - cached).max(0); + r.lost = Some(lost); + if lost > 0 { + r.lost_after = p.between(prev.seq, &key); + } + } + if r.total > 0 { + p.convs.insert( + key, + ConvState { + seq: r.seq, + n: prefix.len(), + hash: prefix[prefix.len() - 1], + total: r.total, + }, + ); + p.evict_convs(); + } + } + if p.ring.len() < RING_MAX { + p.ring.push(r.clone()); + } else { + let h = p.head; + p.ring[h] = r.clone(); + p.head = (h + 1) % RING_MAX; + } + drop(p); + if perf_log_on() { + eprintln!("{}", perf_line(&r)); + } + r + } + + /// `snapshot` : copie de l'anneau, de la plus ancienne à la plus récente. + pub fn snapshot(&self) -> Vec { + let p = self.lock(); + let mut out = Vec::with_capacity(p.ring.len()); + out.extend_from_slice(&p.ring[p.head..]); + out.extend_from_slice(&p.ring[..p.head]); + out + } + + /// Conversations suivies (tests). + pub fn convs_len(&self) -> usize { + self.lock().convs.len() + } + + /// `perfLostAlertAt` du preset actif `cfg` (le modèle résolu comme au + /// lancement : dossiers rangés, `LOKI_MODEL_DIRS`). + pub fn lost_alert_at(&self, cfg: &HashMap, stored_dirs: &[String]) -> i64 { + lost_alert_at(cfg, &|model| { + let env = std::env::var(crate::gguf::MODEL_DIRS_ENV).ok(); + let p = crate::gguf::resolve_serve_model_path( + model, + &self.home, + stored_dirs, + env.as_deref(), + ) + .ok()?; + let g = crate::gguf::gguf_meta(&p).ok()?; + Some(crate::serve::ckpt::gguf_hybrid(Some(&g))) + }) + } +} + +impl Inner { + /// `between` : les natures des complétions passées par le moteur depuis + /// `seq` (un sous-agent, un client externe…), cause probable d'une perte. + fn between(&self, seq: u64, key: &str) -> String { + let mut seen = BTreeSet::new(); + let n = self.ring.len(); + for i in 0..n.min(BETWEEN) { + let idx = ((self.head as i64 - 1 - i as i64).rem_euclid(n as i64)) as usize; + let e = &self.ring[idx]; + if e.seq <= seq { + break; + } + if format!("{}\0{}", e.kind, e.conv) != key { + seen.insert(e.kind.clone()); + } + } + seen.into_iter().collect::>().join(",") + } + + /// `evictConvs` : on oublie la conversation la plus ancienne. + fn evict_convs(&mut self) { + while self.convs.len() > CONV_MAX { + let Some(old) = self + .convs + .iter() + .min_by_key(|(_, s)| s.seq) + .map(|(k, _)| k.clone()) + else { + return; + }; + self.convs.remove(&old); + } + } +} + +/// `perfLogOn` : LOKI_PERF_LOG posée (une ligne par complétion sur stderr). +pub fn perf_log_on() -> bool { + let v = std::env::var("LOKI_PERF_LOG").unwrap_or_default(); + let v = v.trim().to_lowercase(); + !v.is_empty() && !matches!(v.as_str(), "0" | "false" | "off" | "no") +} + +/// `perfLine`. +pub fn perf_line(r: &PerfRec) -> String { + let opt = |p: Option| p.map_or_else(|| "?".to_string(), |v| v.to_string()); + let draft = match r.draft_n { + Some(n) => format!("{}/{n}", opt(r.draft_accepted)), + None => "?".into(), + }; + let mut lost = opt(r.lost); + if !r.lost_after.is_empty() { + lost.push_str(&format!("({})", r.lost_after)); + } + let conv = if r.conv.is_empty() { "-" } else { &r.conv }; + format!( + "[perf] kind={} conv={conv} iter={} total={} cached={} new={} pp_ms={:.0} pp_tps={:.1} gen={} tg_tps={:.1} draft={draft} ttft_ms={} lost={lost} complete={}", + r.kind, + r.iter, + r.total, + opt(r.cached), + r.new, + r.pp_ms, + r.pp_tps, + r.gen_tokens, + r.tg_tps, + opt(r.ttft_ms), + r.complete + ) +} + +/// `perfLostAlertAt` : la perte au-delà de laquelle l'interface la signale. +/// Sur un modèle hybride, le cache ne reprend qu'à un point de reprise : +/// perdre jusqu'à un espacement de points plus un micro-batch est normal. +/// `hybrid` : le modèle du preset est-il hybride (`None` : illisible) ? +pub fn lost_alert_at(cfg: &HashMap, hybrid: &dyn Fn(&str) -> Option) -> i64 { + const BASE: i64 = 1000; + let get = |k: &str| cfg.get(k).map_or("", String::as_str); + let extra = crate::models::split_args(get("EXTRA_ARGS")); + let num = |vals: &[String]| -> i64 { + vals.iter() + .filter_map(|v| v.trim().parse::().ok()) + .find(|n| *n > 0) + .unwrap_or(0) + }; + let env = std::env::var("LLAMA_ARG_CHECKPOINT_MIN_SPACING_NT").unwrap_or_default(); + let mut step = num(&[ + crate::serve::flag_value(&extra, &["-cms", "--checkpoint-min-step"]), + env, + get("CKPT_MIN_STEP").to_string(), + ]); + let model = get("MODEL").trim(); + if step == 0 && !model.is_empty() && hybrid(model) == Some(true) { + step = 2048; + } + if step == 0 { + return BASE; + } + let mut ub = num(&[ + crate::serve::flag_value(&extra, &["-ub", "--ubatch-size"]), + get("UBATCH").to_string(), + ]); + if ub == 0 { + ub = 512; + } + BASE.max(step + ub) +} + +/// `perfAlert` : une perte au-delà du seuil, ou un gros prompt repris de +/// zéro alors que le cache est connu. +pub fn perf_alert(r: &PerfRec, threshold: i64) -> bool { + if r.lost.is_some_and(|l| l > threshold) { + return true; + } + r.cached == Some(0) && r.total >= 8 * threshold +} + +// --------------------------------------------------------------------------- +// /api/perf/summary +// --------------------------------------------------------------------------- + +fn ser_opt_f64(v: &Option, s: S) -> Result { + match v { + Some(f) => ser_go_f64(f, s), + None => s.serialize_none(), + } +} + +/// `perfKindSum`. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct KindSum { + pub completions: i64, + pub incomplete: i64, + /// Premières complétions (iter 0) : un tour, un sous-agent… + pub turns: i64, + pub new_tokens: i64, + #[serde(serialize_with = "ser_go_f64")] + pub prefill_sec: f64, + #[serde(skip_serializing_if = "is_zero_f", serialize_with = "ser_go_f64")] + pub new_tokens_per_turn: f64, + #[serde(skip_serializing_if = "is_zero_f", serialize_with = "ser_go_f64")] + pub prefill_sec_per_turn: f64, + pub lost_events: i64, + pub lost_tokens: i64, + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "ser_opt_f64" + )] + pub ttft_median_ms: Option, + #[serde(skip_serializing_if = "is_zero_i")] + pub draft_n: i64, + #[serde(skip_serializing_if = "is_zero_i")] + pub draft_accepted: i64, + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "ser_opt_f64" + )] + pub draft_rate: Option, +} + +/// `perfDepthSum`. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct DepthSum { + pub bucket: String, + pub n: i64, + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "ser_opt_f64" + )] + pub pp_median_tps: Option, + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "ser_opt_f64" + )] + pub tg_median_tps: Option, +} + +/// `perfSummary`. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct Summary { + pub entries: i64, + #[serde(skip_serializing_if = "Option::is_none")] + pub since: Option, + /// Σ cached / Σ total sur les complétions où les deux sont connus. + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "ser_opt_f64" + )] + pub cache_hit_ratio: Option, + pub kinds: BTreeMap, + pub depth: Vec, + pub draft_n: i64, + pub draft_accepted: i64, + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "ser_opt_f64" + )] + pub draft_rate: Option, + /// Dernier verdict de la sonde de gabarit, absent tant qu'aucune n'a + /// tourné. + #[serde(skip_serializing_if = "Option::is_none")] + pub template: Option, +} + +/// `perfDepthBuckets`. +pub const DEPTH_BUCKETS: [(&str, i64); 4] = [ + ("0-8k", 8 << 10), + ("8-16k", 16 << 10), + ("16-32k", 32 << 10), + ("32k+", i64::MAX), +]; + +/// `perfMedian`. +fn median(mut v: Vec) -> Option { + if v.is_empty() { + return None; + } + v.sort_by(f64::total_cmp); + let n = v.len(); + Some(if n.is_multiple_of(2) { + (v[n / 2 - 1] + v[n / 2]) / 2.0 + } else { + v[n / 2] + }) +} + +/// `perfSummarize`. +pub fn summarize(recs: &[PerfRec]) -> Summary { + let mut s = Summary { + entries: recs.len() as i64, + since: recs.first().map(|r| r.at.clone()), + ..Default::default() + }; + let (mut sum_cached, mut sum_total) = (0i64, 0i64); + let mut pp: Vec> = vec![Vec::new(); DEPTH_BUCKETS.len()]; + let mut tg: Vec> = vec![Vec::new(); DEPTH_BUCKETS.len()]; + let mut ttft: HashMap> = HashMap::new(); + for r in recs { + let k = s.kinds.entry(r.kind.clone()).or_default(); + if !r.complete { + k.incomplete += 1; + continue; + } + k.completions += 1; + if r.iter == 0 { + k.turns += 1; + } + k.new_tokens += r.new; + k.prefill_sec += r.pp_ms / 1000.0; + if let Some(l) = r.lost.filter(|l| *l > 0) { + k.lost_events += 1; + k.lost_tokens += l; + } + if let Some(t) = r.ttft_ms { + ttft.entry(r.kind.clone()).or_default().push(t as f64); + } + if let Some(c) = r.cached + && r.total > 0 + { + sum_cached += c.min(r.total); + sum_total += r.total; + } + let depth = if r.total == 0 { r.new } else { r.total }; + for (i, (_, max)) in DEPTH_BUCKETS.iter().enumerate() { + if depth < *max { + // Débit de lecture : seulement sur un vrai prefill. + if r.pp_tps > 0.0 && r.new >= 64 { + pp[i].push(r.pp_tps); + } + if r.tg_tps > 0.0 && r.gen_tokens > 1 { + tg[i].push(r.tg_tps); + } + break; + } + } + if let Some(n) = r.draft_n { + s.draft_n += n; + k.draft_n += n; + if let Some(a) = r.draft_accepted { + s.draft_accepted += a; + k.draft_accepted += a; + } + } + } + for (kind, k) in s.kinds.iter_mut() { + if k.turns > 0 { + k.new_tokens_per_turn = k.new_tokens as f64 / k.turns as f64; + k.prefill_sec_per_turn = k.prefill_sec / k.turns as f64; + } + k.ttft_median_ms = median(ttft.remove(kind).unwrap_or_default()); + if k.draft_n > 0 { + k.draft_rate = Some(k.draft_accepted as f64 / k.draft_n as f64); + } + } + if sum_total > 0 { + s.cache_hit_ratio = Some(sum_cached as f64 / sum_total as f64); + } + if s.draft_n > 0 { + s.draft_rate = Some(s.draft_accepted as f64 / s.draft_n as f64); + } + for (i, (name, _)) in DEPTH_BUCKETS.iter().enumerate() { + s.depth.push(DepthSum { + bucket: name.to_string(), + n: pp[i].len().max(tg[i].len()) as i64, + pp_median_tps: median(std::mem::take(&mut pp[i])), + tg_median_tps: median(std::mem::take(&mut tg[i])), + }); + } + s +} + +#[cfg(test)] +mod tests; diff --git a/rust/loki-rs/src/perfring/tests.rs b/rust/loki-rs/src/perfring/tests.rs new file mode 100644 index 0000000..da94a6c --- /dev/null +++ b/rust/loki-rs/src/perfring/tests.rs @@ -0,0 +1,125 @@ +//! L'anneau rejoué contre le journal du Go (`parity_perf_test.go` → +//! `tests/parity_perf/perf.json`), et les cas de `perf_log_test.go`. + +use std::collections::HashMap; + +use serde_json::{Value, json}; + +use super::*; + +fn journal() -> Value { + serde_json::from_str(include_str!("../../tests/parity_perf/perf.json")).expect("journal") +} + +fn ring() -> PerfRing { + PerfRing::new(std::path::PathBuf::from("/nulle-part")) +} + +#[test] +fn parite_avec_le_go() { + let j = journal(); + let r = ring(); + for (i, st) in j["steps"].as_array().unwrap().iter().enumerate() { + let mut rec: PerfRec = serde_json::from_value(st["rec"].clone()).unwrap(); + // L'heure posée par le journal (le Go la fixe avant de ranger). + rec.at = st["out"]["at"].as_str().unwrap().to_string(); + let prefix: Vec = st["prefix"] + .as_array() + .map(|a| { + a.iter() + .map(|p| p.as_str().unwrap().parse().unwrap()) + .collect() + }) + .unwrap_or_default(); + let got = r.record(rec, &prefix); + assert_eq!(serde_json::to_value(&got).unwrap(), st["out"], "étape {i}"); + assert_eq!(perf_line(&got), st["line"].as_str().unwrap(), "étape {i}"); + assert_eq!( + perf_alert(&got, 1000), + st["alert_1000"] == json!(1), + "étape {i}" + ); + } + let snap = r.snapshot(); + assert_eq!( + serde_json::to_value(summarize(&snap)).unwrap(), + j["summary"] + ); + assert_eq!(serde_json::to_value(summarize(&[])).unwrap(), j["empty"]); +} + +#[test] +fn seuils_d_alerte_du_go() { + let j = journal(); + for c in j["alerts"].as_array().unwrap() { + let cfg: HashMap = serde_json::from_value(c["cfg"].clone()).unwrap(); + // L'environnement est lu par lost_alert_at : un test à la fois ici. + let env = c["env"].as_str().unwrap_or(""); + // SAFETY : les tests de ce module ne lisent cette variable qu'ici. + unsafe { std::env::set_var("LLAMA_ARG_CHECKPOINT_MIN_SPACING_NT", env) }; + let got = lost_alert_at(&cfg, &|_| None); + assert_eq!(got, c["want"].as_i64().unwrap(), "{c}"); + } + unsafe { std::env::remove_var("LLAMA_ARG_CHECKPOINT_MIN_SPACING_NT") }; + // Un modèle hybride sans espacement posé : 2048 + micro-batch. + let cfg: HashMap = [("MODEL".to_string(), "h.gguf".to_string())].into(); + assert_eq!(lost_alert_at(&cfg, &|m| Some(m == "h.gguf")), 2560); + assert_eq!(lost_alert_at(&cfg, &|_| Some(false)), 1000); + // Le seuil : une perte au-delà, ou un gros prompt repris de zéro. + let rec = |lost: Option, cached: Option, total: i64| PerfRec { + lost, + cached, + total, + ..Default::default() + }; + assert!(!perf_alert(&rec(Some(900), None, 0), 1000)); + assert!(perf_alert(&rec(Some(1500), None, 0), 1000)); + assert!(!perf_alert(&rec(Some(2000), None, 0), 2560)); + assert!(perf_alert(&rec(None, Some(0), 9000), 1000)); + assert!(!perf_alert(&rec(None, Some(0), 500), 1000)); + assert!(!perf_alert(&rec(None, None, 9000), 1000)); +} + +/// `TestPerfRingBounded` : l'anneau et l'état par conversation sont bornés, +/// l'ordre gardé. +#[test] +fn anneau_borne() { + let r = ring(); + for i in 0..RING_MAX + 10 { + r.record( + PerfRec { + kind: "main".into(), + conv: char::from(b'a' + (i % 26) as u8).to_string() + &(i % 200).to_string(), + complete: true, + total: 10, + ..Default::default() + }, + &[i as u64], + ); + } + let recs = r.snapshot(); + assert_eq!(recs.len(), RING_MAX); + for w in recs.windows(2) { + assert_eq!(w[1].seq, w[0].seq + 1); + } + assert_eq!(recs.last().unwrap().seq, (RING_MAX + 10) as u64); + assert!(r.convs_len() <= CONV_MAX, "{}", r.convs_len()); +} + +/// L'heure posée à l'enregistrement quand elle manque (ou nulle, venue du +/// Go). +#[test] +fn heure_posee() { + let mut r = ring(); + r.clock = || "2026-10-16T09:00:00Z".to_string(); + let got = r.record(PerfRec::default(), &[]); + assert_eq!(got.at, "2026-10-16T09:00:00Z"); + let got = r.record( + PerfRec { + at: "0001-01-01T00:00:00Z".into(), + ..Default::default() + }, + &[], + ); + assert_eq!((got.at.as_str(), got.seq), ("2026-10-16T09:00:00Z", 2)); +} diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 4a9ac11..c76ae73 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -128,8 +128,6 @@ pub struct FakeGo { /// pour une action donnée. pub service_os: Mutex>, pub service_os_fail: Mutex>, - /// /internal/v1/turn/perf reçus (corps entiers). - pub perf_records: Mutex>, /// /internal/v1/turn/slot : second slot en service (le fil de la /// discussion sur le 0, le reste sur le 1 ; sinon aucun) ; demandes /// reçues (« conv main »). @@ -781,26 +779,6 @@ async fn handle(st: Arc, req: Request) -> Response .push(format!("served {}", v["ticket"])); resp(204, "") } - "/internal/v1/turn/perf" => { - let body = req.into_body().collect().await.unwrap().to_bytes(); - let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); - st.oai_log.lock().unwrap().push(format!( - "perf kind={} conv={} iter={} complete={} prefix={}", - v["kind"].as_str().unwrap_or(""), - v["conv"].as_str().unwrap_or(""), - v["iter"], - v["complete"], - v["prefix"].as_array().map_or(0, Vec::len), - )); - // Comme le Go : la nature reçue (un tour sans nature). - let kind = v["kind"] - .as_str() - .filter(|k| !k.is_empty()) - .unwrap_or("main"); - let out = serde_json::json!({"kind": kind, "lost": 3, "lost_after": "tâche", "lost_alert": true}); - st.perf_records.lock().unwrap().push(v); - resp(200, out.to_string()) - } "/internal/v1/oai/end" => { let body = req.into_body().collect().await.unwrap().to_bytes(); let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index 758cf70..4d0d8b6 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -667,6 +667,7 @@ async fn tour_complet_par_le_go() { patterns: RetryPatterns::builtin(), tpl: Arc::new(loki_rs::chat::tplprobe::TplProbe::new()), prewarm: Default::default(), + perf: Arc::new(loki_rs::perfring::PerfRing::new(Default::default())), mcp: None, tasks: None, go_version: Default::default(), @@ -771,29 +772,26 @@ async fn tour_complet_par_le_go() { }) .collect(); assert_eq!(answer, "Voilà."); - // Télémétrie : chaque complétion rangée par le Go, rattachée à la - // discussion ; la première, sans usage, reste incomplète. - let perfs = go.perf_records.lock().unwrap().clone(); - assert_eq!(perfs.len(), 2); - assert_eq!(perfs[0]["complete"], false); - assert_eq!(perfs[0]["prefix"], serde_json::json!([])); - let r1 = &perfs[1]; + // Télémétrie (lot 39) : chaque complétion rangée dans l'anneau de la + // façade, rattachée à la discussion ; la première, sans usage, reste + // incomplète. + let recs = shared.perf.snapshot(); + let got: Vec<_> = recs + .iter() + .map(|r| { + ( + r.kind.as_str(), + r.conv.as_str(), + r.iter, + r.complete, + r.total, + ) + }) + .collect(); assert_eq!( - ( - r1["conv"].clone(), - r1["kind"].clone(), - r1["iter"].clone(), - r1["stats"]["prompt_tokens_total"].clone() - ), - ( - serde_json::json!("c7"), - serde_json::json!("main"), - serde_json::json!(1), - serde_json::json!(50) - ) + got, + [("main", "c7", 0, false, 0), ("main", "c7", 1, true, 50)] ); - assert_eq!(perfs[1]["prefix"].as_array().unwrap().len(), 3); - assert_eq!(perfs[1]["external"], false); let last = evs .iter() .rev() @@ -802,9 +800,10 @@ async fn tour_complet_par_le_go() { _ => None, }) .unwrap(); + // Rien de comparable avant : la perte reste inconnue. assert_eq!( (last.kind.as_str(), last.lost, last.lost_after.as_str()), - ("main", Some(3), "tâche") + ("main", None, "") ); } @@ -992,22 +991,30 @@ async fn tour_de_bout_en_bout() { 2, "{log:?}" ); - // Appel d'outil sans usage ni timings : complétion incomplète (comme le - // Go : ni empreintes) ; la réponse finale, avec usage, complète. La - // sonde du gabarit reste à la façade (`sonde_de_gabarit_apres_un_tour`). - for want in [ - format!("perf kind=main conv={id} iter=0 complete=false prefix=0"), - format!("perf kind=main conv={id} iter=1 complete=true prefix=4"), - ] { - assert!(log.contains(&want), "{want} absent de {log:?}"); - } + // Appel d'outil sans usage ni timings : complétion incomplète ; la + // réponse finale, avec usage, complète — dans l'anneau de la façade + // (lot 39), rien par le Go. + assert!(!log.iter().any(|l| l.starts_with("perf ")), "{log:?}"); + let recs: Vec<_> = svc + .shared + .perf + .snapshot() + .into_iter() + .map(|r| (r.kind, r.conv, r.iter, r.complete)) + .collect(); + assert_eq!( + recs, + [ + ("main".to_string(), id.clone(), 0, false), + ("main".to_string(), id.clone(), 1, true) + ] + ); let finals: Vec<_> = deltas(&svc, "stats") .into_iter() - .filter(|s| s["lost_alert"] == true) + .filter(|s| s["kind"] == "main") .collect(); - assert_eq!(finals.len(), 1, "stats finales sans les marques du Go"); - assert_eq!(finals[0]["lost"], 3); - assert_eq!(finals[0]["kind"], "main"); + assert_eq!(finals.len(), 1, "stats finales sans leur nature"); + assert_eq!(finals[0].get("lost_alert"), None); let body = opened(&go, BK_CHAT, &format!("conv:{id}")).unwrap(); assert_eq!(decode_state(&body).messages.len(), 4); // Réessai du même envoi : reconnu. @@ -1943,6 +1950,7 @@ async fn telemetrie_des_resumes() { .to_string(), ]) .await; + let ring = Arc::new(loki_rs::perfring::PerfRing::new(Default::default())); let s = EngineSummarizer { http: EngineHttp::new(), ep: ChatEndpoint { @@ -1954,7 +1962,7 @@ async fn telemetrie_des_resumes() { cfg: Default::default(), patterns: RetryPatterns::builtin(), port: 8080, - go: Some((st, "c9".into())), + go: Some((st, "c9".into(), ring.clone())), }; *go.turn_side_live.lock().unwrap() = true; let req = || TranscriptRequest { @@ -2018,16 +2026,16 @@ async fn telemetrie_des_resumes() { .contains(&"begin main_conv=".to_string()) }) .await; - let perfs = go.perf_records.lock().unwrap().clone(); + let perfs = ring.snapshot(); assert_eq!(perfs.len(), 3, "la réponse illisible n'est pas rangée"); + let mut first = serde_json::to_value(&perfs[0]).unwrap(); + first.as_object_mut().unwrap().remove("at"); assert_eq!( - perfs[0], - serde_json::json!({ "kind": "compact", "conv": "c9", "iter": 0, "complete": true, - "external": false, "prefix": [], - "stats": { "prompt_tokens_total": 3000, "prompt_tokens": 3000, "prompt_ms": 1500.0, - "gen_tokens": 200, "gen_per_second": 40.0 } }) + first, + serde_json::json!({ "seq": 1, "kind": "compact", "conv": "c9", "iter": 0, "complete": true, + "total": 3000, "new": 3000, "pp_ms": 1500, "gen": 200, "tg_tps": 40 }) ); - assert_eq!(perfs[1]["complete"], false); + assert!(!perfs[1].complete); } /// Lot 32 : KEEP_TURN_IMAGES tenu par la façade. L'image relayée par un @@ -2202,11 +2210,12 @@ async fn prechauffage_par_la_facade() { let first_end = log.iter().find(|l| l.starts_with("end")).unwrap(); let v: serde_json::Value = serde_json::from_str(first_end.trim_start_matches("end ")).unwrap(); assert_eq!(v, serde_json::json!({"id": 5, "total": 60, "cached": 42})); - let perfs = go.perf_records.lock().unwrap().clone(); assert!( - perfs + svc.shared + .perf + .snapshot() .iter() - .any(|p| p["kind"] == "prewarm" && p["stats"]["prompt_tokens_total"] == 60) + .any(|p| p.kind == "prewarm" && p.total == 60) ); // La vue du préchauffage est demandée sans rien à ranger (PROJ_SNAPSHOT). assert!( @@ -2252,11 +2261,11 @@ async fn prechauffage_refuse_ou_annule() { r#"end {"cached":null,"id":5,"total":0}"# ); assert!( - go.perf_records - .lock() - .unwrap() + svc.shared + .perf + .snapshot() .iter() - .all(|p| p["kind"] != "prewarm") + .all(|p| p.kind != "prewarm") ); } @@ -2566,3 +2575,136 @@ async fn outil_task_par_la_facade() { )) ); } + +/// Lot 39 : l'anneau de télémétrie tenu par la façade. D'un tour à l'autre +/// de la même discussion, le cache perdu est mesuré ici et signalé dans les +/// stats finales (au-delà du seuil du preset). +#[tokio::test] +async fn cache_perdu_mesure_par_la_facade() { + let reply = |total: i64, cached: i64| { + sse(&[ + serde_json::json!({"choices":[{"delta":{"content":"vu"}}]}), + serde_json::json!({"choices":[{"delta":{},"finish_reason":"stop"}], + "usage":{"prompt_tokens":total,"completion_tokens":2}, + "timings":{"cache_n":cached,"prompt_n":total - cached,"prompt_ms":10.0,"prompt_per_second":100.0}}), + ]) + }; + let (url, _seen) = fake_engine(vec![reply(5000, 0), reply(6000, 200)]).await; + let (svc, _go, _stop, _t) = service(Some(&url)).await; + for (i, m) in ["un", "deux"].iter().enumerate() { + svc.send(serde_json::json!({"message": m, "cid": format!("p{i}")})) + .await + .unwrap(); + until_done(&svc, i + 1).await; + } + let id = svc.sessions.conv().id(); + let recs = svc.shared.perf.snapshot(); + let got: Vec<_> = recs + .iter() + .map(|r| (r.conv.as_str(), r.total, r.cached, r.lost)) + .collect(); + assert_eq!( + got, + [ + (id.as_str(), 5000, Some(0), None), + (id.as_str(), 6000, Some(200), Some(4800)) + ] + ); + let finals: Vec<_> = deltas(&svc, "stats") + .into_iter() + .filter(|s| s["kind"] == "main") + .collect(); + assert_eq!(finals.len(), 2); + assert_eq!(finals[0].get("lost_alert"), None); + assert_eq!( + (finals[1]["lost"].clone(), finals[1]["lost_alert"].clone()), + (serde_json::json!(4800), serde_json::json!(true)) + ); +} + +/// Lot 39 : ce que le Go confie à l'anneau de la façade, ce qu'il en relit, +/// et /api/perf/summary servi par elle. +#[tokio::test] +async fn anneau_de_la_facade_pour_le_go() { + let t = tempfile::tempdir().unwrap(); + let (facade, go, _stop, chat) = fakego::start_with( + t.path().into(), + fakego::Opts { + chat: true, + ..Default::default() + }, + ) + .await; + let chat = chat.unwrap(); + let tok = [(loki_rs::auth::INTERNAL_HEADER, fakego::JETON)]; + let rec = |total: i64, cached: Option, kind: &str| { + serde_json::json!({ "at": "0001-01-01T00:00:00Z", "seq": 0, "kind": kind, "conv": "c1", + "iter": 0, "complete": true, "total": total, "cached": cached }) + }; + let post = |body: serde_json::Value, path: &'static str| { + let b = body.to_string(); + async move { + let (code, out) = fakego::send(facade, "POST", path, &tok, &b).await; + assert_eq!(code, 200, "{}", String::from_utf8_lossy(&out)); + serde_json::from_slice::(&out).unwrap() + } + }; + let r = post( + serde_json::json!({ "rec": rec(3000, Some(0), "subagent"), "prefix": [11, 12] }), + "/internal/facade/v1/perf/record", + ) + .await; + assert_eq!( + (r["rec"]["seq"].clone(), r["rec"]["lost"].clone()), + (serde_json::json!(1), serde_json::Value::Null) + ); + assert_ne!(r["rec"]["at"], "0001-01-01T00:00:00Z", "heure posée"); + // Un client externe passe, puis la suite du sous-agent : perte, et ce qui + // s'est intercalé. + post( + serde_json::json!({ "rec": { "kind": "foreign" } }), + "/internal/facade/v1/perf/record", + ) + .await; + let r = post( + serde_json::json!({ "rec": rec(3500, Some(1000), "subagent"), "prefix": [11, 12, 13] }), + "/internal/facade/v1/perf/record", + ) + .await; + assert_eq!( + (r["rec"]["lost"].clone(), r["rec"]["lost_after"].clone()), + (serde_json::json!(2000), serde_json::json!("foreign")) + ); + let snap = post(serde_json::json!({}), "/internal/facade/v1/perf/snapshot").await; + let kinds: Vec<_> = snap["recs"] + .as_array() + .unwrap() + .iter() + .map(|r| r["kind"].as_str().unwrap().to_string()) + .collect(); + assert_eq!(kinds, ["subagent", "foreign", "subagent"]); + assert_eq!(chat.shared.perf.snapshot().len(), 3); + // Sans jeton : refusé. + let (code, _) = fakego::send( + facade, + "POST", + "/internal/facade/v1/perf/snapshot", + &[], + "{}", + ) + .await; + assert_ne!(code, 200); + // Le résumé, servi par la façade. + let (code, out) = fakego::send(facade, "GET", "/api/perf/summary", &[], "").await; + assert_eq!(code, 200); + let s: serde_json::Value = serde_json::from_slice(&out).unwrap(); + assert_eq!(s["entries"], 3); + assert_eq!(s["kinds"]["subagent"]["completions"], 2); + assert_eq!(s["kinds"]["subagent"]["lost_tokens"], 2000); + assert_eq!(s["kinds"]["foreign"]["incomplete"], 1); + assert_eq!(s["depth"].as_array().unwrap().len(), 4); + assert_eq!(s.get("template"), None); + let (code, _) = fakego::send(facade, "POST", "/api/perf/summary", &[], "").await; + assert_eq!(code, 405); + drop(go); +} diff --git a/rust/loki-rs/tests/parity_perf/perf.json b/rust/loki-rs/tests/parity_perf/perf.json new file mode 100644 index 0000000..f5a611a --- /dev/null +++ b/rust/loki-rs/tests/parity_perf/perf.json @@ -0,0 +1,681 @@ +{ + "steps": [ + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 4000, + "cached": 3000, + "new": 1000, + "pp_ms": 2000.4, + "pp_tps": 499.9, + "gen": 50, + "tg_tps": 40.25, + "draft_n": 40, + "draft_accepted": 30, + "ttft_ms": 2100 + }, + "msgs": [ + "a" + ], + "prefix": [ + "18380994072798496976" + ], + "out": { + "at": "2026-10-16T08:00:00Z", + "seq": 1, + "kind": "main", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 4000, + "cached": 3000, + "new": 1000, + "pp_ms": 2000.4, + "pp_tps": 499.9, + "gen": 50, + "tg_tps": 40.25, + "draft_n": 40, + "draft_accepted": 30, + "ttft_ms": 2100 + }, + "line": "[perf] kind=main conv=c1 iter=0 total=4000 cached=3000 new=1000 pp_ms=2000 pp_tps=499.9 gen=50 tg_tps=40.2 draft=30/40 ttft_ms=2100 lost=? complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c1", + "iter": 1, + "complete": true, + "total": 4600, + "cached": 4000 + }, + "msgs": [ + "a", + "b", + "c" + ], + "prefix": [ + "18380994072798496976", + "7707499087160536902", + "9701112914521480905" + ], + "out": { + "at": "2026-10-16T08:00:01Z", + "seq": 2, + "kind": "main", + "conv": "c1", + "iter": 1, + "complete": true, + "total": 4600, + "cached": 4000, + "lost": 0 + }, + "line": "[perf] kind=main conv=c1 iter=1 total=4600 cached=4000 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=0 complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "subagent", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 300, + "cached": 0, + "new": 300, + "pp_ms": 600, + "pp_tps": 500 + }, + "msgs": [ + "z" + ], + "prefix": [ + "2089108250578558367" + ], + "out": { + "at": "2026-10-16T08:00:02Z", + "seq": 3, + "kind": "subagent", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 300, + "cached": 0, + "new": 300, + "pp_ms": 600, + "pp_tps": 500 + }, + "line": "[perf] kind=subagent conv=c1 iter=0 total=300 cached=0 new=300 pp_ms=600 pp_tps=500.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "foreign", + "iter": 0, + "complete": false + }, + "out": { + "at": "2026-10-16T08:00:03Z", + "seq": 4, + "kind": "foreign", + "iter": 0, + "complete": false + }, + "line": "[perf] kind=foreign conv=- iter=0 total=0 cached=? new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=false", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c1", + "iter": 2, + "complete": true, + "total": 5200, + "cached": 200 + }, + "msgs": [ + "a", + "b", + "c", + "d" + ], + "prefix": [ + "18380994072798496976", + "7707499087160536902", + "9701112914521480905", + "8266243799656022901" + ], + "out": { + "at": "2026-10-16T08:00:04Z", + "seq": 5, + "kind": "main", + "conv": "c1", + "iter": 2, + "complete": true, + "total": 5200, + "cached": 200, + "lost": 4400, + "lost_after": "foreign,subagent" + }, + "line": "[perf] kind=main conv=c1 iter=2 total=5200 cached=200 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=4400(foreign,subagent) complete=true", + "alert_1000": 1 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c1", + "iter": 3, + "complete": true, + "total": 5200 + }, + "msgs": [ + "a", + "b", + "c", + "d", + "e" + ], + "prefix": [ + "18380994072798496976", + "7707499087160536902", + "9701112914521480905", + "8266243799656022901", + "5788732863052660068" + ], + "out": { + "at": "2026-10-16T08:00:05Z", + "seq": 6, + "kind": "main", + "conv": "c1", + "iter": 3, + "complete": true, + "total": 5200 + }, + "line": "[perf] kind=main conv=c1 iter=3 total=5200 cached=? new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c1", + "iter": 4, + "complete": true, + "total": 5300, + "cached": 10 + }, + "msgs": [ + "x", + "b" + ], + "prefix": [ + "4430711847302402693", + "10853208800609132871" + ], + "out": { + "at": "2026-10-16T08:00:06Z", + "seq": 7, + "kind": "main", + "conv": "c1", + "iter": 4, + "complete": true, + "total": 5300, + "cached": 10 + }, + "line": "[perf] kind=main conv=c1 iter=4 total=5300 cached=10 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c2", + "iter": 0, + "complete": true, + "total": 20000, + "cached": 0 + }, + "msgs": [ + "q" + ], + "prefix": [ + "18326388500208184512" + ], + "out": { + "at": "2026-10-16T08:00:07Z", + "seq": 8, + "kind": "main", + "conv": "c2", + "iter": 0, + "complete": true, + "total": 20000, + "cached": 0 + }, + "line": "[perf] kind=main conv=c2 iter=0 total=20000 cached=0 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=true", + "alert_1000": 1 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c2", + "iter": 1, + "complete": false, + "cached": 0 + }, + "msgs": [ + "q", + "r" + ], + "prefix": [ + "18326388500208184512", + "745432260197996294" + ], + "out": { + "at": "2026-10-16T08:00:08Z", + "seq": 9, + "kind": "main", + "conv": "c2", + "iter": 1, + "complete": false, + "cached": 0 + }, + "line": "[perf] kind=main conv=c2 iter=1 total=0 cached=0 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=false", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c2", + "iter": 1, + "complete": true, + "total": 21000, + "cached": 19000 + }, + "msgs": [ + "q", + "r" + ], + "prefix": [ + "18326388500208184512", + "745432260197996294" + ], + "out": { + "at": "2026-10-16T08:00:09Z", + "seq": 10, + "kind": "main", + "conv": "c2", + "iter": 1, + "complete": true, + "total": 21000, + "cached": 19000, + "lost": 1000 + }, + "line": "[perf] kind=main conv=c2 iter=1 total=21000 cached=19000 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=1000 complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "compact", + "conv": "c2", + "iter": 0, + "complete": true, + "total": 9000, + "cached": 8800, + "new": 200, + "pp_ms": 100, + "pp_tps": 2000, + "gen": 300, + "tg_tps": 30, + "ttft_ms": 900 + }, + "out": { + "at": "2026-10-16T08:00:10Z", + "seq": 11, + "kind": "compact", + "conv": "c2", + "iter": 0, + "complete": true, + "total": 9000, + "cached": 8800, + "new": 200, + "pp_ms": 100, + "pp_tps": 2000, + "gen": 300, + "tg_tps": 30, + "ttft_ms": 900 + }, + "line": "[perf] kind=compact conv=c2 iter=0 total=9000 cached=8800 new=200 pp_ms=100 pp_tps=2000.0 gen=300 tg_tps=30.0 draft=? ttft_ms=900 lost=? complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "bench", + "iter": 0, + "complete": true, + "total": 33000, + "new": 33000, + "pp_ms": 30000, + "pp_tps": 1100, + "gen": 64, + "tg_tps": 20.5 + }, + "out": { + "at": "2026-10-16T08:00:11Z", + "seq": 12, + "kind": "bench", + "iter": 0, + "complete": true, + "total": 33000, + "new": 33000, + "pp_ms": 30000, + "pp_tps": 1100, + "gen": 64, + "tg_tps": 20.5 + }, + "line": "[perf] kind=bench conv=- iter=0 total=33000 cached=? new=33000 pp_ms=30000 pp_tps=1100.0 gen=64 tg_tps=20.5 draft=? ttft_ms=? lost=? complete=true", + "alert_1000": 0 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "verify", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 12000, + "cached": 0, + "new": 12000, + "pp_ms": 9000.5, + "pp_tps": 1333.3, + "gen": 2, + "tg_tps": 9.75, + "draft_n": 0 + }, + "out": { + "at": "2026-10-16T08:00:12Z", + "seq": 13, + "kind": "verify", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 12000, + "cached": 0, + "new": 12000, + "pp_ms": 9000.5, + "pp_tps": 1333.3, + "gen": 2, + "tg_tps": 9.75, + "draft_n": 0 + }, + "line": "[perf] kind=verify conv=c1 iter=0 total=12000 cached=0 new=12000 pp_ms=9000 pp_tps=1333.3 gen=2 tg_tps=9.8 draft=?/0 ttft_ms=? lost=? complete=true", + "alert_1000": 1 + }, + { + "rec": { + "at": "0001-01-01T00:00:00Z", + "seq": 0, + "kind": "main", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 6000, + "cached": 5900 + }, + "msgs": [ + "a", + "b", + "c", + "d", + "e", + "f" + ], + "prefix": [ + "18380994072798496976", + "7707499087160536902", + "9701112914521480905", + "8266243799656022901", + "5788732863052660068", + "7575424036688383126" + ], + "out": { + "at": "2026-10-16T08:00:13Z", + "seq": 14, + "kind": "main", + "conv": "c1", + "iter": 0, + "complete": true, + "total": 6000, + "cached": 5900 + }, + "line": "[perf] kind=main conv=c1 iter=0 total=6000 cached=5900 new=0 pp_ms=0 pp_tps=0.0 gen=0 tg_tps=0.0 draft=? ttft_ms=? lost=? complete=true", + "alert_1000": 0 + } + ], + "summary": { + "entries": 14, + "since": "2026-10-16T08:00:00Z", + "cache_hit_ratio": 0.4680778032036613, + "kinds": { + "bench": { + "completions": 1, + "incomplete": 0, + "turns": 1, + "new_tokens": 33000, + "prefill_sec": 30, + "new_tokens_per_turn": 33000, + "prefill_sec_per_turn": 30, + "lost_events": 0, + "lost_tokens": 0 + }, + "compact": { + "completions": 1, + "incomplete": 0, + "turns": 1, + "new_tokens": 200, + "prefill_sec": 0.1, + "new_tokens_per_turn": 200, + "prefill_sec_per_turn": 0.1, + "lost_events": 0, + "lost_tokens": 0, + "ttft_median_ms": 900 + }, + "foreign": { + "completions": 0, + "incomplete": 1, + "turns": 0, + "new_tokens": 0, + "prefill_sec": 0, + "lost_events": 0, + "lost_tokens": 0 + }, + "main": { + "completions": 8, + "incomplete": 1, + "turns": 3, + "new_tokens": 1000, + "prefill_sec": 2.0004, + "new_tokens_per_turn": 333.3333333333333, + "prefill_sec_per_turn": 0.6668, + "lost_events": 2, + "lost_tokens": 5400, + "ttft_median_ms": 2100, + "draft_n": 40, + "draft_accepted": 30, + "draft_rate": 0.75 + }, + "subagent": { + "completions": 1, + "incomplete": 0, + "turns": 1, + "new_tokens": 300, + "prefill_sec": 0.6, + "new_tokens_per_turn": 300, + "prefill_sec_per_turn": 0.6, + "lost_events": 0, + "lost_tokens": 0 + }, + "verify": { + "completions": 1, + "incomplete": 0, + "turns": 1, + "new_tokens": 12000, + "prefill_sec": 9.0005, + "new_tokens_per_turn": 12000, + "prefill_sec_per_turn": 9.0005, + "lost_events": 0, + "lost_tokens": 0 + } + }, + "depth": [ + { + "bucket": "0-8k", + "n": 2, + "pp_median_tps": 499.95, + "tg_median_tps": 40.25 + }, + { + "bucket": "8-16k", + "n": 2, + "pp_median_tps": 1666.65, + "tg_median_tps": 19.875 + }, + { + "bucket": "16-32k", + "n": 0 + }, + { + "bucket": "32k+", + "n": 1, + "pp_median_tps": 1100, + "tg_median_tps": 20.5 + } + ], + "draft_n": 40, + "draft_accepted": 30, + "draft_rate": 0.75 + }, + "empty": { + "entries": 0, + "kinds": {}, + "depth": [ + { + "bucket": "0-8k", + "n": 0 + }, + { + "bucket": "8-16k", + "n": 0 + }, + { + "bucket": "16-32k", + "n": 0 + }, + { + "bucket": "32k+", + "n": 0 + } + ], + "draft_n": 0, + "draft_accepted": 0 + }, + "alerts": [ + { + "cfg": {}, + "want": 1000 + }, + { + "cfg": { + "CKPT_MIN_STEP": "2048" + }, + "want": 2560 + }, + { + "cfg": { + "CKPT_MIN_STEP": "4096", + "UBATCH": "1024" + }, + "want": 5120 + }, + { + "cfg": { + "CKPT_MIN_STEP": "beaucoup" + }, + "want": 1000 + }, + { + "cfg": { + "CKPT_MIN_STEP": " 300 ", + "UBATCH": "100" + }, + "want": 1000 + }, + { + "cfg": { + "CKPT_MIN_STEP": "2048", + "EXTRA_ARGS": "--checkpoint-min-step 8192 -ub 256" + }, + "want": 8448 + }, + { + "cfg": { + "CKPT_MIN_STEP": "2048", + "EXTRA_ARGS": "-ub=2048", + "UBATCH": "1024" + }, + "want": 4096 + }, + { + "cfg": { + "CKPT_MIN_STEP": "-5", + "UBATCH": "0" + }, + "want": 1000 + }, + { + "cfg": { + "CKPT_MIN_STEP": "2048" + }, + "env": "3000", + "want": 3512 + }, + { + "cfg": { + "CKPT_MIN_STEP": "2048" + }, + "env": "rien", + "want": 2560 + }, + { + "cfg": { + "MODEL": "absent.gguf" + }, + "want": 1000 + } + ] +} diff --git a/rust/loki-rs/tests/tools.rs b/rust/loki-rs/tests/tools.rs index cae9487..09637ac 100644 --- a/rust/loki-rs/tests/tools.rs +++ b/rust/loki-rs/tests/tools.rs @@ -93,10 +93,16 @@ async fn criteria_comme_le_go() { } let args = st["args"].as_object().cloned().unwrap_or_default(); let verifier = st["verifier"].as_bool() == Some(true); - let before = store.get_open("chat", &format!("crit:{conv}")).await.unwrap(); + let before = store + .get_open("chat", &format!("crit:{conv}")) + .await + .unwrap(); let out = exec(&store, conv, &args, verifier).await; assert_eq!(out.result, st["out"].as_str().unwrap(), "{name}"); - let raw = store.get_open("chat", &format!("crit:{conv}")).await.unwrap(); + let raw = store + .get_open("chat", &format!("crit:{conv}")) + .await + .unwrap(); let stored = raw .as_deref() .filter(|b| !b.is_empty()) @@ -110,7 +116,11 @@ async fn criteria_comme_le_go() { // L'écran reçoit la liste quand elle change (vide : effacée). match &out.notify { Some(list) => { - let want = if stored.is_null() { json!([]) } else { stored.clone() }; + let want = if stored.is_null() { + json!([]) + } else { + stored.clone() + }; assert_eq!(json!(list), want, "{name}"); } None => assert_eq!(raw, before, "{name} : rien ne change sans notification"),