From 8a61879bfde5da9e331d654b53180e7e8e7af1ba Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 01:17:49 +0000 Subject: [PATCH] =?UTF-8?q?Lot=2029=20:=20t=C3=A9l=C3=A9m=C3=A9trie=20des?= =?UTF-8?q?=20compl=C3=A9tions=20de=20la=20fa=C3=A7ade?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Quand la façade tient la conversation (LOKI_CHAT=rust), ses complétions rejoignent l'anneau de télémétrie du Go (perf_log.go) au lieu de passer inaperçues : /api/perf/summary les compte, l'optimiseur en tire ses poids, et l'interface reçoit à nouveau la perte de cache. - chat/perf.rs : perfRecFromStats, perfRecFromWire (lu sans échec possible, mêmes tolérances que perfNum, timings d'un autre type présents mais vides comme le Go) et perfPrefix (FNV-1a cumulé) ; - la boucle du tour range chaque complétion lue, numérotée dans le tour, avec ses empreintes si elle est complète ; une complétion coupée par un stop est rangée incomplète ; la dernière statistique porte la nature, la perte, ce qui s'est intercalé et l'alerte ; - natures : main (tour), verify (vérification du mode Code), compact (résumé de compaction, rangé dès que la réponse se lit) ; - le Go (/internal/v1/perf/record) range le relevé comme les siens et calcule l'alerte au seuil du preset actif, jamais pour un preset externe. Tests : relevés et empreintes (chat/perf.rs), télémétrie de la boucle (complète, coupée), du vrai hôte et des résumés (tests/lot6.rs), TestFacadePerf côté Go. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv --- ...-rust-migration-lot29-telemetrie-design.md | 75 +++++ internal/loki/facade_perf.go | 37 +++ internal/loki/facade_perf_test.go | 61 ++++ internal/loki/web_internal.go | 2 + rust/loki-rs/src/chat/compact.rs | 5 +- rust/loki-rs/src/chat/generate.rs | 7 +- rust/loki-rs/src/chat/host.rs | 28 +- rust/loki-rs/src/chat/mod.rs | 4 +- rust/loki-rs/src/chat/perf.rs | 279 ++++++++++++++++++ rust/loki-rs/src/chat/turn.rs | 61 +++- rust/loki-rs/src/chat/turn/tests.rs | 89 +++++- rust/loki-rs/src/chat/verify.rs | 1 + rust/loki-rs/tests/common/fakego.rs | 10 + rust/loki-rs/tests/lot6.rs | 87 ++++++ rust/loki-rs/tests/parity.rs | 1 + 15 files changed, 731 insertions(+), 16 deletions(-) create mode 100644 docs/superpowers/specs/2026-10-16-rust-migration-lot29-telemetrie-design.md create mode 100644 internal/loki/facade_perf.go create mode 100644 internal/loki/facade_perf_test.go create mode 100644 rust/loki-rs/src/chat/perf.rs diff --git a/docs/superpowers/specs/2026-10-16-rust-migration-lot29-telemetrie-design.md b/docs/superpowers/specs/2026-10-16-rust-migration-lot29-telemetrie-design.md new file mode 100644 index 0000000..a677c4c --- /dev/null +++ b/docs/superpowers/specs/2026-10-16-rust-migration-lot29-telemetrie-design.md @@ -0,0 +1,75 @@ +# Lot 29 — Télémétrie des complétions de la façade + +## Contexte + +Le Go range chaque complétion dans un anneau en mémoire (`perf_log.go`). +Pour chacune, il note : + +- combien du prompt venait du cache, combien a été recalculé ; +- à quelle vitesse ; +- ce que le décodage spéculatif a gardé. + +Il expose des agrégats sur `/api/perf/summary`. Il calcule aussi la **perte +de cache** d'une complétion : ce que le cache aurait dû reprendre de la +précédente complétion de la même discussion, et n'a pas repris. Il nomme ce +qui s'est intercalé entre les deux, et signale la perte à l'interface +au-delà d'un seuil tiré du preset. + +Quand la façade tient la conversation (`LOKI_CHAT=rust`), ses complétions +n'apparaissaient nulle part. Le résumé des performances ignorait les tours ; +l'optimiseur tirait ses poids d'une télémétrie vide ; l'interface ne +recevait jamais la perte de cache. + +## Le lot + +Les complétions de la façade rejoignent l'anneau du Go : + +- `chat/perf.rs` porte la partie « relevé » : + - `perfRecFromStats` (complétion streamée) ; + - `perfRecFromWire` (réponse non streamée, lue sans échec possible, avec + les mêmes tolérances que `perfNum`) ; + - `perfPrefix` (empreintes FNV-1a cumulées des messages envoyés). +- `TurnHost::perf` range une complétion. La boucle du tour l'appelle : + - après chaque complétion lue, complète ou non, numérotée dans le tour ; + - avec les empreintes seulement si elle est complète ; + - pour une complétion coupée par un stop, rangée incomplète. + + La dernière statistique d'une complétion complète porte la nature, la + perte de cache, ce qui s'est intercalé et l'alerte, comme celle du Go. +- La nature : `main` pour un tour, `verify` pour la vérification du mode + Code, `compact` pour un résumé de compaction. Le résumé est rangé dès que + sa réponse se lit, sans empreintes. +- Le Go (`/internal/v1/perf/record`) range le relevé comme les siens : + numéro, instant, perte de cache, ce qui s'est intercalé. Il calcule + l'alerte au seuil du preset actif (`perfLostAlertAt`), jamais pour un + preset externe. + +`/api/perf/summary` reste au Go, qui tient l'anneau. + +## Tests + +- `chat/perf.rs` : + - relevé d'une réponse entière : nombres en chaîne, valeurs hors bornes, + octet parasite, cache d'OpenAI, timings d'un autre type ; + - empreintes cumulées, FNV-1a de référence ; + - relevé au format du Go. +- La boucle du tour : + - deux complétions numérotées, nature et totaux, empreintes qui se + prolongent ; + - dernière statistique avec la perte vue par le Go ; + - complétion coupée rangée incomplète, sans statistique finale. +- Le vrai hôte (`tests/lot6.rs`) : + - complétions d'un tour rangées par le Go, rattachées à la discussion ; + - résumés de compaction (rangés si la réponse se lit, pas sinon). +- Le Go (`TestFacadePerf`) : + - première complétion, perte et ce qui s'est intercalé, alerte ; + - preset externe sans alerte, complétion incomplète ; + - champs envoyés en trop ignorés, corps illisible, GET refusé. + +## Écarts assumés + +- Les empreintes sont calculées sur le JSON des messages tel que Rust + l'écrit (sans l'échappement HTML du Go). Elles ne se comparent qu'entre + complétions de la même discussion et de la même nature, toutes produites + par la façade. +- Une complétion coupée par un stop est rangée sans ses compteurs partiels. diff --git a/internal/loki/facade_perf.go b/internal/loki/facade_perf.go new file mode 100644 index 0000000..e483efb --- /dev/null +++ b/internal/loki/facade_perf.go @@ -0,0 +1,37 @@ +package loki + +import ( + "net/http" + "time" +) + +// facade_perf.go — la télémétrie des complétions de la façade (lot 29 de la +// migration). Quand la façade tient la conversation (LOKI_CHAT=rust), ses +// complétions — tours, vérifications, résumés de compaction — rejoignent +// l'anneau du Go (perf_log.go) : /api/perf/summary les compte, la perte de +// cache se calcule contre la complétion précédente de la même discussion, et +// l'alerte suit le seuil du preset actif (perfLostAlertAt), comme pour un +// tour du Go. + +// handleInternalPerfRecord : POST {rec, prefix, external} → {seq, lost, +// lost_after, alert}. rec : une complétion (perfRec, sans at ni seq) ; +// prefix : les empreintes cumulées des messages envoyés (perfPrefix), vide +// pour une complétion incomplète ; external : preset externe (jamais +// d'alerte). +func handleInternalPerfRecord(w http.ResponseWriter, r *http.Request) { + var in struct { + Rec perfRec `json:"rec"` + Prefix []uint64 `json:"prefix"` + External bool `json:"external"` + } + if !readInternalJSON(w, r, &in) { + return + } + in.Rec.At, in.Rec.Seq, in.Rec.Lost, in.Rec.LostAfter = time.Time{}, 0, nil, "" + rec := perfRecord(in.Rec, in.Prefix) + alert := false + if rec.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{"seq": rec.Seq, "lost": rec.Lost, "lost_after": rec.LostAfter, "alert": alert}) +} diff --git a/internal/loki/facade_perf_test.go b/internal/loki/facade_perf_test.go new file mode 100644 index 0000000..9daf4fd --- /dev/null +++ b/internal/loki/facade_perf_test.go @@ -0,0 +1,61 @@ +package loki + +import ( + "encoding/json" + "net/http/httptest" + "strings" + "testing" +) + +// La télémétrie des complétions de la façade rejoint l'anneau du Go : perte +// de cache calculée contre la complétion précédente de la même discussion, +// ce qui s'est intercalé, et l'alerte au seuil du preset actif. +func TestFacadePerf(t *testing.T) { + testHome(t) + post := func(body string) (int, map[string]any) { + rec := httptest.NewRecorder() + handleInternalPerfRecord(rec, httptest.NewRequest("POST", "/internal/v1/perf/record", strings.NewReader(body))) + var out map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &out) + return rec.Code, out + } + // Première complétion : rien à comparer. Ce que la façade enverrait en + // trop (at, seq, lost) est ignoré. + code, out := post(`{"rec":{"kind":"main","conv":"fp1","iter":0,"complete":true,"total":5000,"cached":0,"new":5000, + "seq":99,"lost":7,"lost_after":"x"},"prefix":[11,22]}`) + if code != 200 || out["lost"] != nil || out["lost_after"] != "" || out["alert"] != false { + t.Fatalf("première : %d %v", code, out) + } + seq := out["seq"].(float64) + if seq == 99 || seq <= 0 { + t.Fatalf("seq : %v", seq) + } + // 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(`{"rec":{"kind":"main","conv":"fp1","iter":1,"complete":true,"total":6000,"cached":100,"new":5900},"prefix":[11,22,33]}`) + if code != 200 || out["lost"] != float64(4900) || out["lost_after"] != perfTask || out["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(`{"rec":{"kind":"main","conv":"fp1","iter":2,"complete":true,"total":7000,"cached":100},"prefix":[11,22,33,44],"external":true}`) + if code != 200 || out["lost"] != float64(5900) || out["alert"] != false { + t.Fatalf("externe : %d %v", code, out) + } + // Incomplète : ni perte ni alerte. + code, out = post(`{"rec":{"kind":"main","conv":"fp1","iter":3,"complete":false,"cached":0}}`) + if code != 200 || out["lost"] != nil || out["alert"] != false { + t.Fatalf("incomplète : %d %v", code, out) + } + if code, _ := post(`{"rec":`); code != 400 { + t.Fatalf("corps illisible : %d", code) + } + rec := httptest.NewRecorder() + handleInternalPerfRecord(rec, httptest.NewRequest("GET", "/internal/v1/perf/record", nil)) + if rec.Code != 405 { + t.Fatalf("GET : %d", rec.Code) + } +} diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index 3aad6e4..7744ee0 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -135,6 +135,8 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalBenchPerf(w, r) case "/internal/v1/tune/context": handleInternalTuneContext(w, r) + case "/internal/v1/perf/record": + handleInternalPerfRecord(w, r) case "/internal/v1/tasks/run": handleInternalTaskRun(w, r) case "/internal/v1/push/send": diff --git a/rust/loki-rs/src/chat/compact.rs b/rust/loki-rs/src/chat/compact.rs index 6c354fd..b21f67c 100644 --- a/rust/loki-rs/src/chat/compact.rs +++ b/rust/loki-rs/src/chat/compact.rs @@ -853,6 +853,9 @@ struct WireMsg { tool_calls: Option>, } +/// Réponse lue mais sans choix. +pub const ERR_EMPTY_SUMMARY: &str = "résumé: réponse vide"; + /// Décodage d'une réponse non streamée de /v1/chat/completions, comme /// `postSummary` : seule la PREMIÈRE valeur JSON compte (un octet parasite /// après elle est ignoré), aucun choix → « résumé: réponse vide ». @@ -866,7 +869,7 @@ pub fn parse_summary_response(raw: &[u8]) -> Result Some(Ok(r)) => r, }; let Some(ch) = resp.choices.and_then(|c| c.into_iter().next()) else { - return Err(SummaryError::Other("résumé: réponse vide".into())); + return Err(SummaryError::Other(ERR_EMPTY_SUMMARY.into())); }; Ok(SummaryChoice { content: ch.message.content.unwrap_or_default(), diff --git a/rust/loki-rs/src/chat/generate.rs b/rust/loki-rs/src/chat/generate.rs index 5a48246..cfa10e8 100644 --- a/rust/loki-rs/src/chat/generate.rs +++ b/rust/loki-rs/src/chat/generate.rs @@ -23,8 +23,9 @@ //! ([`super::verify`]) ; toute compaction rend l'état des critères au résumé. //! //! Écarts assumés (restent au Go) : PROJ_SNAPSHOT et KEEP_TURN_IMAGES -//! (refusés par `/turn/view`), préchauffage, télémétrie, motifs -//! `retry_patterns` personnalisés. +//! (refusés par `/turn/view`), préchauffage, motifs `retry_patterns` +//! personnalisés. La télémétrie des complétions (tours, vérifications, +//! résumés) rejoint l'anneau du Go (`/internal/v1/perf/record`). use std::collections::VecDeque; use std::future::Future; @@ -508,6 +509,7 @@ impl ChatService { cfg: cfg.clone(), patterns: self.shared.patterns.clone(), port, + perf: Some((store.clone(), self.sessions.conv().id())), }; TurnEnv { cfg, @@ -797,6 +799,7 @@ impl ChatService { echo: echo.clone(), compact: ccfg.clone(), patterns: self.shared.patterns.clone(), + perf_kind: super::perf::KIND_MAIN, }; let me = self.clone(); let host = GoTurnHost { diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs index 337ed94..e6d8b80 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -170,6 +170,9 @@ pub struct EngineSummarizer { pub cfg: HashMap, pub patterns: Arc, pub port: u16, + /// Télémétrie des résumés (kind=compact), rattachée à cette discussion : + /// `None` sans base (tests). + pub perf: Option<(Store, String)>, } impl EngineSummarizer { @@ -200,7 +203,19 @@ impl EngineSummarizer { body: String::from_utf8_lossy(b).trim().to_string(), }); } - parse_summary_response(&raw) + let out = parse_summary_response(&raw); + // Télémétrie de la compaction : réponse non streamée, timings et usage + // à la racine, lus sans échec possible — dès que la réponse se lit. + let decoded = match &out { + Ok(_) => true, + Err(SummaryError::Other(e)) => e == super::compact::ERR_EMPTY_SUMMARY, + Err(_) => false, + }; + if decoded && let Some((store, conv)) = &self.perf { + let rec = super::perf::PerfRec::from_wire(super::perf::KIND_COMPACT, conv, &raw); + super::perf::record(store, &rec, &[], self.ep.external).await; + } + out } } @@ -393,6 +408,17 @@ impl TurnHost for GoTurnHost { } } + async fn perf( + &self, + rec: &super::perf::PerfRec, + prefix: &[u64], + external: bool, + ) -> super::perf::PerfNoted { + let mut rec = rec.clone(); + rec.conv = self.conv_id.clone(); + super::perf::record(&self.shared.store, &rec, prefix, external).await + } + async fn exec_tool(&self, name: &str, args: &Map, label: &str) -> ToolOutcome { // Outils fichiers et shell : ici (lot 7), l'état partagé restant au Go. if crate::tools::is_local(name) { diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs index e2edcdc..b5c67fa 100644 --- a/rust/loki-rs/src/chat/mod.rs +++ b/rust/loki-rs/src/chat/mod.rs @@ -18,7 +18,8 @@ //! - [`host`] : le tour branché sur le moteur, les outils du Go et la base ; //! - [`history`] : début et fin d'un tour côté historique ; //! - [`generate`] : un tour de bout en bout (envoi, file, fin de tour) ; -//! - [`tplprobe`] : sonde de gabarit (rendus de `/apply-template`). +//! - [`tplprobe`] : sonde de gabarit (rendus de `/apply-template`) ; +//! - [`perf`] : télémétrie des complétions (relevés rangés par le Go). pub mod compact; pub mod conv; @@ -29,6 +30,7 @@ pub mod host; pub mod journal; pub mod lease; pub mod nudges; +pub mod perf; pub mod persist; pub mod sessions; pub mod stream; diff --git a/rust/loki-rs/src/chat/perf.rs b/rust/loki-rs/src/chat/perf.rs new file mode 100644 index 0000000..fa63196 --- /dev/null +++ b/rust/loki-rs/src/chat/perf.rs @@ -0,0 +1,279 @@ +//! Télémétrie des complétions de la façade (lot 29) — port de la partie +//! « relevé » de `perf_log.go` : `perfRecFromStats`, `perfRecFromWire`, +//! `perfPrefix`. L'anneau, le calcul de la perte de cache et son seuil +//! d'alerte restent au Go (`/internal/v1/perf/record`) : les complétions de +//! la façade rejoignent celles du Go dans `/api/perf/summary`. +//! +//! Des identifiants et des compteurs, jamais de texte. Rien ne retourne au +//! moteur. + +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +use super::types::{Message, StatsEvent}; +use crate::store::Store; + +/// Nature d'une complétion. +pub const KIND_MAIN: &str = "main"; +pub const KIND_VERIFY: &str = "verify"; +pub const KIND_COMPACT: &str = "compact"; + +fn is_zero_i(v: &i64) -> bool { + *v == 0 +} + +fn is_zero_f(v: &f64) -> bool { + *v == 0.0 +} + +/// `perfRec` (sans `at` ni `seq`, posés par le Go). +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct PerfRec { + pub kind: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub conv: String, + pub iter: i64, + pub complete: bool, + /// usage.prompt_tokens : la taille du prompt entier. + #[serde(skip_serializing_if = "is_zero_i")] + pub total: i64, + /// cache_n ; `None` : inconnu. + #[serde(skip_serializing_if = "Option::is_none")] + pub cached: Option, + /// prompt_n : les jetons réellement recalculés. + #[serde(skip_serializing_if = "is_zero_i")] + pub new: i64, + #[serde(skip_serializing_if = "is_zero_f")] + pub pp_ms: f64, + #[serde(skip_serializing_if = "is_zero_f")] + pub pp_tps: f64, + #[serde(rename = "gen", skip_serializing_if = "is_zero_i")] + pub generated: i64, + #[serde(skip_serializing_if = "is_zero_f")] + pub tg_tps: f64, + #[serde(skip_serializing_if = "Option::is_none")] + pub draft_n: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub draft_accepted: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub ttft_ms: Option, +} + +impl PerfRec { + /// `perfRecFromStats` : la complétion telle que la boucle l'a vue. + pub fn from_stats(kind: &str, conv: &str, iter: i64, complete: bool, s: &StatsEvent) -> Self { + PerfRec { + kind: kind.into(), + conv: conv.into(), + iter, + 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, + generated: s.gen_tokens, + tg_tps: s.gen_per_second, + draft_n: s.draft_n, + draft_accepted: s.draft_accepted, + ttft_ms: s.ttft_ms, + } + } + + /// `perfRecFromWire` : une réponse NON streamée (compaction), lue d'un + /// bloc. Jamais d'échec : un compteur illisible reste inconnu. + pub fn from_wire(kind: &str, conv: &str, raw: &[u8]) -> Self { + let mut r = PerfRec { + kind: kind.into(), + conv: conv.into(), + ..Default::default() + }; + // Seule la première valeur JSON compte (`json.Decoder`). + let v = serde_json::Deserializer::from_slice(raw) + .into_iter::() + .next() + .and_then(Result::ok) + .unwrap_or(Value::Null); + // Une valeur non nulle d'un autre type (chaîne, nombre) alloue quand + // même la structure côté Go avant l'erreur de type : présente, vide. + let has_timings = v.get("timings").is_some_and(|t| !t.is_null()); + let timings = v.get("timings").filter(|t| t.is_object()); + if let Some(t) = timings { + let n = |k: &str| perf_num(t.get(k)); + r.new = n("prompt_n").unwrap_or(0.0) as i64; + r.pp_ms = n("prompt_ms").unwrap_or(0.0); + r.pp_tps = n("prompt_per_second").unwrap_or(0.0); + r.generated = n("predicted_n").unwrap_or(0.0) as i64; + r.tg_tps = n("predicted_per_second").unwrap_or(0.0); + r.cached = n("cache_n").map(|f| f as i64); + r.draft_n = n("draft_n").map(|f| f as i64); + r.draft_accepted = n("draft_n_accepted").map(|f| f as i64); + } + let usage = v.get("usage").filter(|u| u.is_object()); + if let Some(u) = usage { + r.total = perf_num(u.get("prompt_tokens")).unwrap_or(0.0) as i64; + if r.generated == 0 { + r.generated = perf_num(u.get("completion_tokens")).unwrap_or(0.0) as i64; + } + } + if r.cached.is_none() { + r.cached = usage + .and_then(|u| u.get("prompt_tokens_details")) + .filter(|d| d.is_object()) + .and_then(|d| perf_num(d.get("cached_tokens"))) + .map(|f| f as i64); + } + r.complete = r.total > 0 || has_timings; + r + } +} + +/// `perfNum` : un nombre (ou une chaîne qui en contient un), fini, positif +/// et < 1e12 ; sinon inconnu. +fn perf_num(v: Option<&Value>) -> Option { + let f = match v? { + Value::Number(n) => n.as_f64()?, + Value::String(s) => s.parse::().ok()?, + _ => return None, + }; + (f.is_finite() && (0.0..1e12).contains(&f)).then_some(f) +} + +/// FNV-1a 64 bits. +fn fnv1a(h: &mut u64, b: &[u8]) { + for &c in b { + *h ^= u64::from(c); + *h = h.wrapping_mul(0x0000_0100_0000_01b3); + } +} + +/// `perfPrefix` : empreintes cumulées des messages envoyés (l'empreinte i +/// couvre les messages 0..i). Une requête PROLONGE la précédente de la même +/// discussion quand l'empreinte de la précédente est un de ses préfixes. +pub fn prefix(msgs: &[Message]) -> Vec { + let mut h = 0xcbf2_9ce4_8422_2325u64; + msgs.iter() + .map(|m| { + fnv1a(&mut h, &serde_json::to_vec(m).unwrap_or_default()); + fnv1a(&mut h, &[0]); + h + }) + .collect() +} + +/// Ce que le Go a tiré de la complétion rangée. +#[derive(Clone, Debug, Default, PartialEq, Deserialize)] +pub struct PerfNoted { + #[serde(default)] + pub lost: Option, + #[serde(default)] + pub lost_after: String, + #[serde(default)] + pub alert: bool, +} + +/// Range la complétion dans l'anneau du Go. Le Go injoignable : rien (la +/// télémétrie ne fait jamais échouer un tour). +pub async fn record(store: &Store, rec: &PerfRec, prefix: &[u64], external: bool) -> PerfNoted { + match store + .internal_post( + "/internal/v1/perf/record", + &json!({ "rec": rec, "prefix": prefix, "external": external }), + ) + .await + { + Ok((code, v)) if code.is_success() => serde_json::from_value(v).unwrap_or_default(), + _ => PerfNoted::default(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn releve_d_une_reponse_entiere() { + let r = PerfRec::from_wire( + KIND_COMPACT, + "c1", + br#"{"timings":{"prompt_n":"120","prompt_ms":50.5,"prompt_per_second":2376,"predicted_n":30,"predicted_per_second":40,"cache_n":900,"draft_n":-1},"usage":{"prompt_tokens":1020,"completion_tokens":31}} parasite"#, + ); + assert_eq!( + r, + PerfRec { + kind: "compact".into(), + conv: "c1".into(), + complete: true, + total: 1020, + cached: Some(900), + new: 120, + pp_ms: 50.5, + pp_tps: 2376.0, + generated: 30, + tg_tps: 40.0, + ..Default::default() + } + ); + // Sans timings : le cache d'OpenAI, les jetons générés de l'usage. + let r = PerfRec::from_wire( + KIND_COMPACT, + "", + br#"{"usage":{"prompt_tokens":10,"completion_tokens":3,"prompt_tokens_details":{"cached_tokens":4}}}"#, + ); + assert_eq!( + (r.complete, r.total, r.generated, r.cached), + (true, 10, 3, Some(4)) + ); + // Rien de lisible : incomplète, tout inconnu. + let r = PerfRec::from_wire(KIND_COMPACT, "", b"pas du json"); + assert!(!r.complete); + assert_eq!(r.cached, None); + let r = PerfRec::from_wire( + KIND_COMPACT, + "", + br#"{"timings":null,"usage":{"prompt_tokens":1e13}}"#, + ); + assert_eq!((r.complete, r.total), (false, 0)); + // Des timings d'un autre type : présents mais vides, comme le Go. + let r = PerfRec::from_wire(KIND_COMPACT, "", br#"{"timings":"x"}"#); + assert_eq!((r.complete, r.new), (true, 0)); + } + + #[test] + fn empreintes_cumulees() { + let m = |c: &str| Message { + role: "user".into(), + content: Some(json!(c)), + ..Default::default() + }; + let a = prefix(&[m("a"), m("b")]); + let b = prefix(&[m("a"), m("b"), m("c")]); + assert_eq!(a.len(), 2); + assert_eq!(b[..2], a[..]); + assert_ne!(prefix(&[m("x"), m("b")])[1], a[1]); + assert!(prefix(&[]).is_empty()); + // FNV-1a de référence : « a » seul. + let mut h = 0xcbf2_9ce4_8422_2325u64; + fnv1a(&mut h, b"a"); + assert_eq!(h, 0xaf63_dc4c_8601_ec8c); + } + + #[test] + fn releve_au_format_du_go() { + let s = StatsEvent { + prompt_tokens: 5, + prompt_tokens_total: 100, + cache_tokens: Some(95), + gen_tokens: 7, + gen_per_second: 20.5, + ttft_ms: Some(30), + ..Default::default() + }; + let r = PerfRec::from_stats(KIND_MAIN, "c", 2, true, &s); + assert_eq!( + serde_json::to_value(&r).unwrap(), + json!({ "kind": "main", "conv": "c", "iter": 2, "complete": true, "total": 100, + "cached": 95, "new": 5, "gen": 7, "tg_tps": 20.5, "ttft_ms": 30 }) + ); + } +} diff --git a/rust/loki-rs/src/chat/turn.rs b/rust/loki-rs/src/chat/turn.rs index b7638d7..033f224 100644 --- a/rust/loki-rs/src/chat/turn.rs +++ b/rust/loki-rs/src/chat/turn.rs @@ -23,12 +23,15 @@ //! `cb` (comme le `ChatCallback` Go) ; le tour rend ses messages à ajouter à //! l'historique ([`TurnResult::extra`]). //! +//! Télémétrie : chaque complétion lue est rangée (`perfRecord`, par +//! [`TurnHost::perf`]) ; la dernière statistique d'une complétion complète +//! porte la nature et la perte de cache calculée par le Go. +//! //! Non portés (clés opt-in, ou machinerie du moteur local restée au Go) : //! KEEP_TURN_IMAGES, PROJ_SNAPSHOT, SIDE_SLOT (slot et rejeux « cache KV -//! plein »), SLOT_PERSIST, préchauffage, télémétrie `perfRecord`, comptage -//! des requêtes en vol du moteur. Avec l'une -//! de ces clés, le Go garde la conversation (`facadeChatBlocker`, décidé au -//! démarrage). +//! plein »), SLOT_PERSIST, préchauffage, comptage des requêtes en vol du +//! moteur. Avec l'une de ces clés, le Go garde la conversation +//! (`facadeChatBlocker`, décidé au démarrage). use std::collections::HashMap; use std::future::Future; @@ -54,6 +57,7 @@ use super::nudges::{ MAX_NUDGES, THINK_NUDGE_FIRST, THINK_NUDGE_STUCK, append_nudge, budget_nudge, dedupable_tool, log_budget, repeated_call_result, steer_system, tool_result_preview, }; +use super::perf::PerfRec; use super::stream::{ CacheHintFn, MAX_PATTERN_RETRIES, PreflightFn, RetryPatterns, StreamConfig, StreamParser, call_args, retry_corrective, tool_call_label, @@ -181,6 +185,16 @@ pub trait TurnHost: Send + Sync { fn cache_hint(&self) -> Option { None } + /// Range la télémétrie d'une complétion (`perfRecord`), rattachée à la + /// discussion du tour ; rend la perte de cache calculée et l'alerte. + fn perf( + &self, + _rec: &super::perf::PerfRec, + _prefix: &[u64], + _external: bool, + ) -> impl Future + Send { + async { super::perf::PerfNoted::default() } + } /// Garde d'écriture du mode Code (`codeWriteGuard`). fn preflight(&self) -> Option { None @@ -202,6 +216,8 @@ pub struct TurnConfig { pub echo: EchoPolicy, pub compact: CompactConfig, pub patterns: Arc, + /// Nature des complétions pour la télémétrie (`perfMain`, `perfVerify`). + pub perf_kind: &'static str, } /// Fin du tour : messages à ajouter à l'historique, et l'erreur s'il y en a @@ -348,6 +364,8 @@ struct Loop { echo_tpl_retry: bool, tool_runs: i64, budget_nudges: i64, + /// Complétions rangées dans la télémétrie (`perfIter`). + perf_iter: i64, } /// `runChatLoop` : le tour, de la première requête à la réponse finale. @@ -415,8 +433,9 @@ pub async fn run_turn( let calls_on = !tc.tools.is_empty() && !s.disable_tools && !s.tool_choice_none; let pol = turn_echo_policy(ep.external || s.echo_off_turn, || tc.echo.clone()); host.prepare_images(&messages).await; - let (body, echo_sent) = { + let (body, echo_sent, sent_prefix) = { let (sent, echo_sent) = wire_messages(&messages, s.tool_choice_none, &pol); + let sent_prefix = super::perf::prefix(&sent); let payload = build_chat_payload( ep, &sent, @@ -432,7 +451,11 @@ pub async fn run_turn( &tc.cfg, &|r| host.image_data_url(r), ); - (serde_json::to_vec(&payload).unwrap_or_default(), echo_sent) + ( + serde_json::to_vec(&payload).unwrap_or_default(), + echo_sent, + sent_prefix, + ) }; let t_req = Instant::now(); let reply = match cancel.or_stop(host.send(ep, body)).await { @@ -603,14 +626,32 @@ pub async fn run_turn( } } // Arrêt pendant la lecture : le tour s'arrête là, sans erreur, et rien - // de plus ne part vers l'écran. + // de plus ne part vers l'écran. La complétion coupée reste dans la + // télémétrie, incomplète. if stopped { + let rec = + PerfRec::from_stats(tc.perf_kind, "", s.perf_iter, false, &Default::default()); + host.perf(&rec, &[], ep.external).await; return TurnResult { extra, err: None }; } let out = parser.finish(&mut emit); - if out.is_complete(read_err.is_some()) { - // Dernier événement de la complétion : copie complète des stats. - cb(StreamEvent::Stats(out.stats.clone())); + // Télémétrie : une complétion coupée avant son chunk final n'a ni cache + // ni total fiables — rangée incomplète, sans calcul de perte. + let complete = out.is_complete(read_err.is_some()); + let rec = PerfRec::from_stats(tc.perf_kind, "", s.perf_iter, complete, &out.stats); + let noted = host + .perf(&rec, if complete { &sent_prefix } else { &[] }, ep.external) + .await; + s.perf_iter += 1; + if complete { + // Dernier événement de la complétion : copie complète des stats, + // avec la perte de cache vue par la télémétrie. + let mut st = out.stats.clone(); + st.kind = tc.perf_kind.into(); + st.lost = noted.lost; + st.lost_after = noted.lost_after; + st.lost_alert = noted.alert; + cb(StreamEvent::Stats(st)); // Sonde de gabarit, en tâche de fond et après coup : rien n'en // revient dans la requête suivante. if !ep.external && !s.disable_tools && !s.tool_choice_none { diff --git a/rust/loki-rs/src/chat/turn/tests.rs b/rust/loki-rs/src/chat/turn/tests.rs index e9040c5..34641bf 100644 --- a/rust/loki-rs/src/chat/turn/tests.rs +++ b/rust/loki-rs/src/chat/turn/tests.rs @@ -2,7 +2,7 @@ use super::*; use crate::chat::compact::CompactConfig; -use crate::chat::types::ToolFunction; +use crate::chat::types::{StatsEvent, ToolFunction}; use crate::chat::wire::resolve_chat_endpoint; use serde_json::json; use std::collections::VecDeque; @@ -36,6 +36,10 @@ struct Fake { unstable: Mutex, /// Complétions signalées à la sonde : (outils, kwargs, effort). completions: Mutex>, + /// Télémétrie rangée : (relevé, empreintes, externe) ; et ce que le Go + /// en tire. + perfs: Mutex, bool)>>, + noted: Mutex, } impl Fake { @@ -143,6 +147,18 @@ impl TurnHost for Fake { fn prefix_unstable(&self, _t: &[Tool], _k: Option<&Map>, _e: &str) -> bool { *self.unstable.lock().unwrap() } + async fn perf( + &self, + rec: &crate::chat::perf::PerfRec, + prefix: &[u64], + external: bool, + ) -> crate::chat::perf::PerfNoted { + self.perfs + .lock() + .unwrap() + .push((rec.clone(), prefix.to_vec(), external)); + self.noted.lock().unwrap().clone() + } } fn content(s: &str) -> String { @@ -190,6 +206,7 @@ fn config(tools: &[&str]) -> TurnConfig { echo: EchoPolicy::default(), compact: CompactConfig::default(), patterns: RetryPatterns::builtin(), + perf_kind: "main", } } @@ -623,3 +640,73 @@ async fn flux_externe_coupe_repris_apres_le_texte_affiche() { let (r, _) = run(&host, &config(&[]), &Cancel::never()).await; assert!(r.err.unwrap().contains("a été coupé en cours de route")); } + +/// Télémétrie : chaque complétion rangée, numérotée dans le tour, avec les +/// empreintes des messages envoyés ; la dernière statistique porte la +/// nature et la perte de cache vue par le Go. +#[tokio::test] +async fn telemetrie_des_completions() { + use crate::chat::perf::PerfNoted; + let host = Fake::new(vec![ + tool_step("c1", "glob", r#"{"pattern":"*.go"}"#), + answer("Trouvé"), + ]); + *host.noted.lock().unwrap() = PerfNoted { + lost: Some(40), + lost_after: "foreign".into(), + alert: true, + }; + let mut tc = config(&["glob"]); + tc.perf_kind = "verify"; + let (r, evs) = run(&host, &tc, &Cancel::never()).await; + assert_eq!(r.err, None); + let perfs = host.perfs.lock().unwrap().clone(); + assert_eq!(perfs.len(), 2); + for (i, (rec, prefix, external)) in perfs.iter().enumerate() { + assert_eq!(rec.kind, "verify"); + assert_eq!(rec.iter, i as i64); + assert!(rec.complete); + assert_eq!((rec.total, rec.generated), (100, 5)); + assert!(!external); + assert!(!prefix.is_empty()); + } + // La seconde requête prolonge la première : mêmes empreintes en tête. + let (a, b) = (&perfs[0].1, &perfs[1].1); + assert_eq!(a.len(), 1); + assert_eq!(b.len(), 3); + assert_eq!(b[..1], a[..]); + let last: Vec<&StatsEvent> = evs + .iter() + .filter_map(|e| match e { + StreamEvent::Stats(s) if !s.kind.is_empty() => Some(s), + _ => None, + }) + .collect(); + assert_eq!(last.len(), 2); + assert_eq!( + ( + last[1].kind.as_str(), + last[1].lost, + last[1].lost_after.as_str(), + last[1].lost_alert + ), + ("verify", Some(40), "foreign", true) + ); +} + +/// Une complétion coupée avant son chunk final : rangée incomplète, sans +/// empreintes ; un stop pendant la lecture aussi. +#[tokio::test] +async fn telemetrie_incomplete() { + let host = Fake::new(vec![Script::Cut(vec![content("Bon")])]); + let (_, evs) = run(&host, &config(&[]), &Cancel::never()).await; + let perfs = host.perfs.lock().unwrap().clone(); + assert_eq!(perfs.len(), 1); + assert!(!perfs[0].0.complete); + assert!(perfs[0].1.is_empty()); + assert!( + !evs.iter() + .any(|e| matches!(e, StreamEvent::Stats(s) if !s.kind.is_empty())), + "pas de statistique finale" + ); +} diff --git a/rust/loki-rs/src/chat/verify.rs b/rust/loki-rs/src/chat/verify.rs index 17e1808..9efbae9 100644 --- a/rust/loki-rs/src/chat/verify.rs +++ b/rust/loki-rs/src/chat/verify.rs @@ -408,6 +408,7 @@ impl ChatService { echo: env.echo.clone(), compact: env.ccfg.clone(), patterns: self.shared.patterns.clone(), + perf_kind: super::perf::KIND_VERIFY, }; let host = GoTurnHost { shared: self.shared.clone(), diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 1a64491..d8304e8 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -125,6 +125,8 @@ pub struct FakeGo { /// pour une action donnée. pub service_os: Mutex>, pub service_os_fail: Mutex>, + /// /internal/v1/perf/record reçus (corps entiers). + pub perf_records: Mutex>, } impl FakeGo { @@ -500,6 +502,14 @@ async fn handle(st: Arc, req: Request) -> Response } resp(200, r#"{"ran":true,"error":""}"#) } + "/internal/v1/perf/record" => { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); + let mut log = st.perf_records.lock().unwrap(); + log.push(v); + let out = serde_json::json!({"seq": log.len(), "lost": 12, "lost_after": "task", "alert": false}); + resp(200, out.to_string()) + } "/internal/v1/tune/context" => { let v = serde_json::json!({ "busy": *st.tune_ctx_busy.lock().unwrap(), diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index eb29526..8c45588 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -680,6 +680,7 @@ async fn tour_complet_par_le_go() { cfg: cfg.clone(), patterns: shared.patterns.clone(), port: 8080, + perf: None, }, compact: TurnCompact { cfg: CompactConfig::default(), @@ -710,6 +711,7 @@ async fn tour_complet_par_le_go() { echo: Default::default(), compact: CompactConfig::default(), patterns: shared.patterns.clone(), + perf_kind: "main", }; let question = Message { role: "user".into(), @@ -761,6 +763,38 @@ 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]["rec"]["complete"], false); + assert_eq!(perfs[0]["prefix"], serde_json::json!([])); + let r1 = &perfs[1]["rec"]; + assert_eq!( + ( + r1["conv"].clone(), + r1["kind"].clone(), + r1["iter"].clone(), + r1["total"].clone() + ), + ( + serde_json::json!("c7"), + serde_json::json!("main"), + serde_json::json!(1), + serde_json::json!(50) + ) + ); + assert_eq!(perfs[1]["prefix"].as_array().unwrap().len(), 3); + assert_eq!(perfs[1]["external"], false); + let last = evs + .iter() + .rev() + .find_map(|e| match e { + StreamEvent::Stats(s) if !s.kind.is_empty() => Some(s.clone()), + _ => None, + }) + .unwrap(); + assert_eq!((last.lost, last.lost_after.as_str()), (Some(12), "task")); } #[tokio::test] @@ -978,6 +1012,7 @@ async fn hote_et_sonde_de_gabarit() { cfg, patterns: shared.patterns.clone(), port: 1, + perf: None, }, compact: TurnCompact { cfg: CompactConfig::default(), @@ -1768,3 +1803,55 @@ async fn parite_du_tour_complet() { failures.join("\n\n") ); } + +/// Lot 29 : le résumé d'une compaction rejoint la télémétrie du Go +/// (kind=compact, rattaché à la discussion), dès que la réponse se lit. +#[tokio::test] +async fn telemetrie_des_resumes() { + use loki_rs::chat::compact::{Summarizer, TranscriptRequest}; + let (st, go, _stop, _t) = store().await; + let (url, _seen) = fake_engine(vec![ + serde_json::json!({ + "choices": [{"message": {"content": "résumé"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 3000, "completion_tokens": 200}, + "timings": {"prompt_n": 3000, "prompt_ms": 1500, "predicted_n": 200, "predicted_per_second": 40} + }) + .to_string(), + r#"{"choices":[]}"#.into(), + "pas du json".into(), + ]) + .await; + let s = EngineSummarizer { + http: EngineHttp::new(), + ep: ChatEndpoint { + url, + model: "loki".into(), + key: String::new(), + external: false, + }, + cfg: Default::default(), + patterns: RetryPatterns::builtin(), + port: 8080, + perf: Some((st, "c9".into())), + }; + let req = || TranscriptRequest { + system: "s".into(), + transcript: "t".into(), + max_tokens: 500, + }; + assert_eq!( + s.summarize_transcript(req()).await.unwrap().content, + "résumé" + ); + assert!(s.summarize_transcript(req()).await.is_err(), "réponse vide"); + assert!(s.summarize_transcript(req()).await.is_err(), "illisible"); + let perfs = go.perf_records.lock().unwrap().clone(); + assert_eq!(perfs.len(), 2, "la réponse illisible n'est pas rangée"); + assert_eq!( + perfs[0]["rec"], + serde_json::json!({ "kind": "compact", "conv": "c9", "iter": 0, "complete": true, + "total": 3000, "new": 3000, "pp_ms": 1500.0, "gen": 200, "tg_tps": 40.0 }) + ); + assert_eq!(perfs[0]["prefix"], serde_json::json!([])); + assert_eq!(perfs[1]["rec"]["complete"], false); +} diff --git a/rust/loki-rs/tests/parity.rs b/rust/loki-rs/tests/parity.rs index 7f759ec..10f1924 100644 --- a/rust/loki-rs/tests/parity.rs +++ b/rust/loki-rs/tests/parity.rs @@ -195,6 +195,7 @@ async fn replay(fx: Fixture) -> Result<(), String> { echo, patterns: RetryPatterns::builtin(), cfg, + perf_kind: "main", }; let host = ParityHost { replies: Mutex::new(fx.replies.into()),