From 5edcad50a505ac1b884e7dcf8a27c1ba7fd1d932 Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Thu, 15 Oct 2026 21:04:05 +0200 Subject: [PATCH] =?UTF-8?q?Conversation=20tenue=20par=20Rust=20:=20t=C3=A9?= =?UTF-8?q?l=C3=A9m=C3=A9trie=20et=20requ=C3=AAtes=20en=20vol=20comme=20le?= =?UTF-8?q?=20Go?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Trois choses manquaient au tour tenu par la façade, actives même sans clé opt-in, et qu'il fallait avant d'en faire le défaut : - la télémétrie de chaque complétion (perfRecord) : /api/perf/summary et l'alerte « cache perdu » posée sur les stats finales d'une étape. La boucle Rust la remet au Go (/internal/v1/turn/perf) au même moment que la boucle Go, avec ses empreintes de messages ; le Go range, calcule la perte et rend nature, jetons perdus et alerte, posés sur la copie finale des stats ; - la sonde du gabarit, lancée dans les mêmes conditions (requête de forme normale du moteur local) ; - le comptage des requêtes en vol : chaque requête au moteur local ouvre un ticket (/internal/v1/oai/begin), marqué servi à la réponse 200 (engineServed, engineMarkMain pour une étape de la discussion), rendu en fin de lecture du flux. Le slot de la discussion n'est ainsi ni effacé pendant une requête ni pris pour un autre. Co-Authored-By: Claude Opus 5.5 --- internal/loki/web_internal.go | 4 + internal/loki/web_internal_oai.go | 39 ++++++- internal/loki/web_internal_turnperf.go | 72 ++++++++++++ internal/loki/web_internal_turnperf_test.go | 67 +++++++++++ rust/loki-rs/src/chat/generate.rs | 1 + rust/loki-rs/src/chat/host.rs | 122 +++++++++++++++++++- rust/loki-rs/src/chat/turn.rs | 114 ++++++++++++++++-- rust/loki-rs/src/chat/verify.rs | 2 + rust/loki-rs/tests/common/fakego.rs | 41 ++++++- rust/loki-rs/tests/lot6.rs | 41 +++++++ 10 files changed, 483 insertions(+), 20 deletions(-) create mode 100644 internal/loki/web_internal_turnperf.go create mode 100644 internal/loki/web_internal_turnperf_test.go diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index 5fad81b..d81c80b 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -69,6 +69,10 @@ func withInternalAPI(next http.Handler) http.Handler { handleInternalOAIBegin(w, r) case "/internal/v1/oai/end": handleInternalOAIEnd(w, r) + case "/internal/v1/oai/served": + handleInternalOAIServed(w, r) + case "/internal/v1/turn/perf": + handleInternalTurnPerf(w, r) case "/internal/v1/tool/exec": handleInternalToolExec(w, r) case "/internal/v1/tool/context": diff --git a/internal/loki/web_internal_oai.go b/internal/loki/web_internal_oai.go index 71da86b..44bd020 100644 --- a/internal/loki/web_internal_oai.go +++ b/internal/loki/web_internal_oai.go @@ -31,6 +31,11 @@ const oaiTicketMaxAge = 6 * time.Hour type oaiTicket struct { end func() when time.Time + // seq / mainConv : requête d'un tour de discussion tenu par la façade — + // une fois acceptée par le moteur (served), le slot porte cette discussion + // (engineMarkMain). Vides pour un client /v1. + seq uint64 + mainConv string } var oaiTickets = struct { @@ -62,16 +67,20 @@ func handleInternalOAIBegin(w http.ResponseWriter, r *http.Request) { Read bool `json:"read"` // GET/HEAD : ne calcule rien dans un slot Side bool `json:"side"` // corps routé vers le second slot PostV1 bool `json:"post_v1"` // complétion d'un client externe + // MainConv : étape d'un tour de cette discussion, tenue par la façade + // (LOKI_CHAT=rust) — comptée comme la requête principale du Go. + MainConv string `json:"main_conv"` } if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&in) != nil { sendJSON(w, http.StatusBadRequest, map[string]any{"error": "POST JSON attendu"}) return } var end func() + var seq uint64 if in.Read { - end, _ = engineRequestBeginSide(func(*prewarmRun) bool { return true }, true) + end, seq = engineRequestBeginSide(func(*prewarmRun) bool { return true }, true) } else { - end, _ = engineRequestBeginSide(nil, in.Side) + end, seq = engineRequestBeginSide(nil, in.Side) } if in.PostV1 { refuse := func(retry, msg, code string) { @@ -101,7 +110,7 @@ func handleInternalOAIBegin(w http.ResponseWriter, r *http.Request) { } oaiTickets.next++ id := oaiTickets.next - oaiTickets.m[id] = oaiTicket{end: end, when: now} + oaiTickets.m[id] = oaiTicket{end: end, when: now, seq: seq, mainConv: in.MainConv} oaiTickets.mu.Unlock() sendJSON(w, http.StatusOK, map[string]any{"ticket": id}) } @@ -132,3 +141,27 @@ func oaiInflight() int { defer oaiTickets.mu.Unlock() return len(oaiTickets.m) } + +// handleInternalOAIServed : la requête du ticket a été acceptée par le moteur +// (réponse 200) — engineServed, et pour l'étape d'un tour de discussion, +// engineMarkMain : le slot porte désormais cette discussion. Le ticket reste +// ouvert jusqu'à end (fin de lecture du flux). +func handleInternalOAIServed(w http.ResponseWriter, r *http.Request) { + var in struct { + Ticket uint64 `json:"ticket"` + } + if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&in) != nil { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "POST JSON attendu"}) + return + } + oaiTickets.mu.Lock() + t, ok := oaiTickets.m[in.Ticket] + oaiTickets.mu.Unlock() + if !ok { + http.NotFound(w, r) + return + } + engineServed() + engineMarkMain(t.seq, t.mainConv) + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/loki/web_internal_turnperf.go b/internal/loki/web_internal_turnperf.go new file mode 100644 index 0000000..b4a169d --- /dev/null +++ b/internal/loki/web_internal_turnperf.go @@ -0,0 +1,72 @@ +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…], +// "probe":{"tools":[…],"kwargs":{…},"effort":""}} +// → {"kind":"main","lost":N,"lost_after":"…","lost_alert":true} +// +// 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. probe : présent quand +// la boucle Go aurait lancé la sonde du gabarit (requête de forme normale du +// fil principal, moteur local). + +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"` + Probe *struct { + Tools []Tool `json:"tools"` + Kwargs map[string]any `json:"kwargs"` + Effort string `json:"effort"` + } `json:"probe"` + } + 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())) + } + if in.Complete && !in.External && in.Probe != nil { + tplProbeKick(in.Probe.Tools, in.Probe.Kwargs, in.Probe.Effort) + } + 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 new file mode 100644 index 0000000..0a5548a --- /dev/null +++ b/internal/loki/web_internal_turnperf_test.go @@ -0,0 +1,67 @@ +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) + } +} + +// 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 a81882f..a63f0ef 100644 --- a/rust/loki-rs/src/chat/generate.rs +++ b/rust/loki-rs/src/chat/generate.rs @@ -811,6 +811,7 @@ impl ChatService { kwargs: reasoning.kwargs.clone().unwrap_or_default(), }, queued: Box::new(move || me.drain_queued(epoch)), + perf_kind: "main", }; // Chaque événement devient un delta du journal. diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs index 0a78b28..f66f163 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -33,7 +33,7 @@ use super::compact::{ use super::engine::{EngineHttp, classify, friendly_error}; use super::persist::{Persister, StoreSink, ToolResJob}; use super::stream::{CacheHintFn, PreflightFn, RetryPatterns}; -use super::turn::{CompactPhase, Reply, SendError, ToolOutcome, TurnHost}; +use super::turn::{CompactPhase, PerfMarks, PerfRec, Reply, SendError, ToolOutcome, TurnHost}; use super::types::{AskEvent, DiffLine, Message, Tool}; use super::wire::{ ChatEndpoint, ChatPayloadOpts, EchoRefusals, EffortFallbacks, IMG_REF_SCHEME, @@ -314,6 +314,65 @@ pub struct GoTurnHost { pub compact: TurnCompact, /// Messages mis en file par l'utilisateur pendant le tour. pub queued: Box Vec + Send + Sync>, + /// Nature des complétions pour la télémétrie : `main` (tour de la + /// discussion), `verify` (passe de vérification du mode Code). + pub perf_kind: &'static str, +} + +/// Rend le ticket « en vol » au Go quand la réponse du moteur est lue ou +/// abandonnée (`engineRequestBeginSide` … `end`). +struct GateTicket { + store: Store, + ticket: u64, +} + +impl Drop for GateTicket { + fn drop(&mut self) { + let (store, ticket) = (self.store.clone(), self.ticket); + if let Ok(rt) = tokio::runtime::Handle::try_current() { + rt.spawn(async move { + let _ = store + .internal_json( + hyper::Method::POST, + "/internal/v1/oai/end", + Some(&json!({ "ticket": ticket })), + ) + .await; + }); + } + } +} + +impl GoTurnHost { + /// Ouvre un ticket « en vol » auprès du Go : le slot de la discussion + /// n'est pas effacé pendant la requête, et un préchauffage est annulé. + /// `None` si le Go ne répond pas — la requête part quand même. + async fn gate_begin(&self) -> Option { + let main_conv = if self.perf_kind == "main" { + self.conv_id.as_str() + } else { + "" + }; + match self + .shared + .store + .internal_json( + hyper::Method::POST, + "/internal/v1/oai/begin", + Some(&json!({ "main_conv": main_conv })), + ) + .await + { + Ok(v) => v["ticket"].as_u64().map(|ticket| GateTicket { + store: self.shared.store.clone(), + ticket, + }), + Err(e) => { + eprintln!("[loki-rs] moteur : requête non comptée en vol ({e})"); + None + } + } + } } /// `toolResID` : « .<24 hex> ». @@ -372,10 +431,35 @@ pub fn see_image_message(label: &str, img: Value) -> Message { impl TurnHost for GoTurnHost { async fn send(&self, ep: &ChatEndpoint, body: Vec) -> Result { + // Moteur local : requête comptée en vol jusqu'à la fin de lecture du + // flux (le ticket voyage avec le corps). Rien pour une API externe. + let gate = if ep.external { + None + } else { + self.gate_begin().await + }; match self.shared.http.post(&ep.url, &ep.key, body).await { Ok(resp) => { let status = resp.status().as_u16(); - let body = resp.bytes_stream().map(|r| r.map_err(|e| e.to_string())); + if status == 200 + && let Some(g) = &gate + { + // Le slot porte désormais cette requête (engineServed, + // engineMarkMain pour une étape de la discussion). + let _ = self + .shared + .store + .internal_json( + hyper::Method::POST, + "/internal/v1/oai/served", + Some(&json!({ "ticket": g.ticket })), + ) + .await; + } + let body = resp.bytes_stream().map(move |r| { + let _keep = &gate; + r.map_err(|e| e.to_string()) + }); Ok(Reply { status, body: Box::pin(body), @@ -510,6 +594,40 @@ impl TurnHost for GoTurnHost { None } + async fn perf_record(&self, rec: PerfRec) -> PerfMarks { + let probe = rec + .probe + .as_ref() + .map(|p| json!({ "tools": p.tools, "kwargs": p.kwargs, "effort": p.effort })); + let body = json!({ + "kind": self.perf_kind, + "conv": self.conv_id, + "iter": rec.iter, + "complete": rec.complete, + "external": rec.external, + "stats": rec.stats, + "prefix": rec.prefix, + "probe": probe, + }); + match self + .shared + .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() + } + } + } + fn preflight(&self) -> Option { None } diff --git a/rust/loki-rs/src/chat/turn.rs b/rust/loki-rs/src/chat/turn.rs index 7eb782f..3d18b75 100644 --- a/rust/loki-rs/src/chat/turn.rs +++ b/rust/loki-rs/src/chat/turn.rs @@ -25,10 +25,11 @@ //! //! Non portés (clés opt-in, ou machinerie du moteur local restée au Go) : //! KEEP_TURN_IMAGES, PROJ_SNAPSHOT, NUDGE_IN_TOOL, SIDE_SLOT (slot et -//! rejeux « cache KV plein »), SLOT_PERSIST, préchauffage, sonde de gabarit, -//! 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). +//! rejeux « cache KV plein »), SLOT_PERSIST, préchauffage. Avec l'une de +//! ces clés, le Go garde la conversation (`facadeChatBlocker`, décidé au +//! démarrage). La télémétrie (`perfRecord`, sonde du gabarit) et le comptage +//! des requêtes en vol du moteur passent par le Go ([`TurnHost::perf_record`], +//! et l'hôte autour de [`TurnHost::send`]). use std::collections::HashMap; use std::future::Future; @@ -59,8 +60,8 @@ use super::stream::{ call_args, retry_corrective, tool_call_label, }; use super::types::{ - AskEvent, DiffLine, Message, ReasoningEcho, StreamEvent, Tool, ToolCall, ToolCallFunc, - ToolUsedEvent, + AskEvent, DiffLine, Message, ReasoningEcho, StatsEvent, StreamEvent, Tool, ToolCall, + ToolCallFunc, ToolUsedEvent, }; use super::wire::{ ChatEndpoint, ChatPayloadOpts, EchoPolicy, TOOLS_OFF_HINT, TurnReasoning, build_chat_payload, @@ -171,6 +172,62 @@ pub trait TurnHost: Send + Sync { fn preflight(&self) -> Option { None } + /// Télémétrie d'une complétion (`perfRecord`) : la rend au Go, qui la + /// range dans l'anneau de `/api/perf/summary` et dit quoi poser sur la + /// copie finale des stats (nature, jetons perdus, alerte). + fn perf_record(&self, _rec: PerfRec) -> impl Future + Send { + async { PerfMarks::default() } + } +} + +/// Une complétion telle que la boucle l'a vue, pour la télémétrie. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct PerfRec { + /// Rang de la complétion dans le tour (`perfIter`). + pub iter: i64, + pub complete: bool, + pub external: bool, + pub stats: StatsEvent, + /// Empreintes cumulées des messages envoyés ([`perf_prefix`]), si complète. + pub prefix: Vec, + /// Requête de forme normale du moteur local : de quoi lancer la sonde du + /// gabarit (`tplProbeKick`). + pub probe: Option, +} + +#[derive(Debug, Clone, PartialEq, Default)] +pub struct PerfProbe { + pub tools: Vec, + pub kwargs: Option>, + pub effort: String, +} + +/// Ce que la télémétrie pose sur la copie finale des stats. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct PerfMarks { + pub kind: String, + pub lost: Option, + pub lost_after: String, + pub lost_alert: bool, +} + +/// `perfPrefix` : empreinte FNV-1a 64 cumulée de chaque message envoyé (son +/// JSON suivi d'un octet nul). Comparée seulement aux empreintes des envois +/// précédents de la même discussion, toutes calculées ici. +pub fn perf_prefix(msgs: &[Message]) -> Vec { + const OFFSET: u64 = 0xcbf2_9ce4_8422_2325; + const PRIME: u64 = 0x0000_0100_0000_01b3; + let fnv = |h: u64, b: u8| (h ^ u64::from(b)).wrapping_mul(PRIME); + let mut h = OFFSET; + let mut out = Vec::with_capacity(msgs.len()); + for m in msgs { + for b in serde_json::to_vec(m).unwrap_or_default() { + h = fnv(h, b); + } + h = fnv(h, 0); + out.push(h); + } + out } /// Ce que le tour sait avant de commencer (lu UNE fois). @@ -334,6 +391,8 @@ struct Loop { echo_tpl_retry: bool, tool_runs: i64, budget_nudges: i64, + /// `perfIter` : complétions de ce tour, complètes ou non. + perf_iter: i64, } /// `runChatLoop` : le tour, de la première requête à la réponse finale. @@ -389,7 +448,7 @@ 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, prefix) = { let (sent, echo_sent) = wire_messages(&messages, s.tool_choice_none, &pol); let payload = build_chat_payload( ep, @@ -406,7 +465,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, + perf_prefix(&sent), + ) }; let t_req = Instant::now(); let reply = match cancel.or_stop(host.send(ep, body)).await { @@ -582,9 +645,38 @@ pub async fn run_turn( 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())); + let complete = out.is_complete(read_err.is_some()); + // Télémétrie (perf_log.go) : une complétion coupée avant son chunk + // final est rangée comme incomplète, sans calcul de perte. Sonde du + // gabarit : fil d'une requête de forme normale, moteur local. + let marks = host + .perf_record(PerfRec { + iter: s.perf_iter, + complete, + external: ep.external, + stats: out.stats.clone(), + prefix: if complete { prefix } else { Vec::new() }, + probe: (complete && !ep.external && !s.disable_tools && !s.tool_choice_none).then( + || PerfProbe { + tools: tc.tools.clone(), + kwargs: kwargs.clone(), + effort: effort.clone(), + }, + ), + }) + .await; + s.perf_iter += 1; + if complete { + // Dernier événement de la complétion : copie complète des stats, + // avec ce que la télémétrie en dit (nature, perte, alerte). + let mut st = out.stats.clone(); + if !marks.kind.is_empty() { + st.kind = marks.kind; + } + st.lost = marks.lost; + st.lost_after = marks.lost_after; + st.lost_alert = marks.lost_alert; + cb(StreamEvent::Stats(st)); if !ep.external { s.peak_gen = s.peak_gen.max(out.stats.gen_tokens); if s.last_est > 0 && out.stats.prompt_tokens_total > 0 { diff --git a/rust/loki-rs/src/chat/verify.rs b/rust/loki-rs/src/chat/verify.rs index 17e1808..d479231 100644 --- a/rust/loki-rs/src/chat/verify.rs +++ b/rust/loki-rs/src/chat/verify.rs @@ -428,6 +428,8 @@ impl ChatService { // Les messages en file attendent la fin du tour (runChatTools sans // injection, comme le Go). queued: Box::new(Vec::new), + // Passe de vérification : comptée à part, comme perfVerify. + perf_kind: if isolated { "verify" } else { "main" }, }; let mut out = PassOut::default(); let mut cb = |ev: StreamEvent| -> bool { diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 19b92dd..f77daf0 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -358,15 +358,48 @@ async fn handle(st: Arc, req: Request) -> Response "/internal/v1/oai/begin" => { 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!( - "begin read={} side={} post_v1={}", - v["read"], v["side"], v["post_v1"] - )); + if let Some(conv) = v["main_conv"].as_str() { + st.oai_log + .lock() + .unwrap() + .push(format!("begin main_conv={conv}")); + } else { + st.oai_log.lock().unwrap().push(format!( + "begin read={} side={} post_v1={}", + v["read"], v["side"], v["post_v1"] + )); + } if let Some(r) = st.refuse.lock().unwrap().clone() { return resp(200, serde_json::json!({ "refuse": r }).to_string()); } resp(200, r#"{"ticket":7}"#) } + "/internal/v1/oai/served" => { + 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!("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={} probe={}", + 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), + !v["probe"].is_null() + )); + resp( + 200, + r#"{"kind":"main","lost":3,"lost_after":"tâche","lost_alert":true}"#, + ) + } "/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 c63b16b..d6e6b56 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -645,6 +645,7 @@ async fn tour_complet_par_le_go() { kwargs: Default::default(), }, queued: Box::new(Vec::new), + perf_kind: "main", }; let bash = Tool { kind: "function".into(), @@ -862,6 +863,46 @@ async fn tour_de_bout_en_bout() { // Bail rendu, fin signalée, discussion enregistrée. assert_eq!(*go.lease.lock().unwrap(), None); let id = svc.sessions.conv().id(); + // Chaque requête au moteur local est comptée en vol par le Go (ticket + // ouvert, servi, rendu), et sa complétion rangée dans la télémétrie ; + // la copie finale des stats porte ce que le Go en dit. + until(|| { + go.oai_log + .lock() + .unwrap() + .iter() + .filter(|l| l.starts_with("end")) + .count() + == 2 + }) + .await; + let log = go.oai_log.lock().unwrap().clone(); + let want_begin = format!("begin main_conv={id}"); + assert_eq!( + log.iter().filter(|l| **l == want_begin).count(), + 2, + "{log:?}" + ); + assert_eq!( + log.iter().filter(|l| *l == "served 7").count(), + 2, + "{log:?}" + ); + // Appel d'outil sans usage ni timings : complétion incomplète (comme le + // Go : ni empreintes, ni sonde) ; la réponse finale, avec usage, complète. + for want in [ + format!("perf kind=main conv={id} iter=0 complete=false prefix=0 probe=false"), + format!("perf kind=main conv={id} iter=1 complete=true prefix=4 probe=true"), + ] { + assert!(log.contains(&want), "{want} absent de {log:?}"); + } + let finals: Vec<_> = deltas(&svc, "stats") + .into_iter() + .filter(|s| s["lost_alert"] == true) + .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"); 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.