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.