Michael/Loki
1
0 Fork 0
mirror of https://github.com/R0m1k3/Loki.git synced 2026-10-22 06:28:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fusion de main : conversation Rust par défaut, télémétrie du tour

Browse Source 
main a reçu en parallèle la télémétrie du tour tenu par la façade
(/internal/v1/turn/perf), les tickets « en vol » du moteur et la
conversation tenue par la façade par défaut (LOKI_CHAT=go pour revenir
au Go). Le lot 29 faisait la même chose par /internal/v1/perf/record :
un seul chemin est gardé, celui de main.

- TurnHost::perf_record (main) remplace TurnHost::perf ; la nature est
  posée par l'hôte (verify pour la passe isolée seulement, main pour les
  corrections du builder, comme le Go).
- Les résumés de compaction passent par le même point d'entrée, nature
  compact : chat/perf.rs ne garde que le relevé de la réponse entière
  (perfRecFromWire) et l'envoi.
- La sonde du gabarit reste à la façade (lot 23) : /internal/v1/turn/perf
  ne la lance plus côté Go.
- facade_perf.go retiré ; ses cas (perte, intercalé, alerte, preset
  externe, incomplète, GET) rejoignent web_internal_turnperf_test.go.
- README : LOKI_CHAT défaut Rust, avec ce que les lots 19, 23, 29 et 30
  y ajoutent.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv
This commit is contained in:
Claude committed 2026-10-16 01:53:53 +00:00
2 parents e444202b4b 4c8ac93581
commit 89ed527c96
20 files changed
+713 -457
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

rust/loki-rs/src/chat/generate.rs
+2 -2
View File
@@ -25,7 +25,7 @@
//! Écarts assumés (restent au Go) : PROJ_SNAPSHOT et KEEP_TURN_IMAGES
//! (refusés par `/turn/view`), préchauffage. La télémétrie des complétions
//! (tours, vérifications, résumés) rejoint l'anneau du Go
//! (`/internal/v1/perf/record`) ; les motifs `retry_patterns` personnalisés
//! (`/internal/v1/turn/perf`) ; les motifs `retry_patterns` personnalisés
//! sont lus au démarrage ([`retry_patterns`]).
use std::collections::VecDeque;
@@ -800,7 +800,6 @@ 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 {
@@ -820,6 +819,7 @@ impl ChatService {
kwargs: reasoning.kwargs.clone().unwrap_or_default(),
},
queued: Box::new(move || me.drain_queued(epoch)),
perf_kind: super::perf::KIND_MAIN,
};
// Chaque événement devient un delta du journal.
rust/loki-rs/src/chat/host.rs
+98 -15
View File
@@ -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,
@@ -212,8 +212,14 @@ impl EngineSummarizer {
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;
let (complete, stats) = super::perf::wire_stats(&raw);
let rec = PerfRec {
complete,
external: self.ep.external,
stats,
..Default::default()
};
super::perf::record(store, super::perf::KIND_COMPACT, conv, &rec).await;
}
out
}
@@ -331,6 +337,65 @@ pub struct GoTurnHost {
pub compact: TurnCompact,
/// Messages mis en file par l'utilisateur pendant le tour.
pub queued: Box<dyn Fn() -> Vec<Message> + 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<GateTicket> {
let main_conv = if self.perf_kind == super::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` : « <discussion>.<24 hex> ».
@@ -389,10 +454,35 @@ pub fn see_image_message(label: &str, img: Value) -> Message {
impl TurnHost for GoTurnHost {
async fn send(&self, ep: &ChatEndpoint, body: Vec<u8>) -> Result<Reply, SendError> {
// 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),
@@ -408,17 +498,6 @@ 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<String, Value>, label: &str) -> ToolOutcome {
// Outils fichiers et shell : ici (lot 7), l'état partagé restant au Go.
if crate::tools::is_local(name) {
@@ -571,6 +650,10 @@ impl TurnHost for GoTurnHost {
None
}
async fn perf_record(&self, rec: PerfRec) -> PerfMarks {
super::perf::record(&self.shared.store, self.perf_kind, &self.conv_id, &rec).await
}
fn preflight(&self) -> Option<PreflightFn> {
None
}
rust/loki-rs/src/chat/perf.rs
+111 -210
View File
@@ -1,16 +1,18 @@
//! 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`.
//! 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.
//!
//! Des identifiants et des compteurs, jamais de texte. Rien ne retourne au
//! 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.
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use super::types::{Message, StatsEvent};
use super::turn::{PerfMarks, PerfRec};
use super::types::StatsEvent;
use crate::store::Store;
/// Nature d'une complétion.
@@ -18,114 +20,47 @@ 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<i64>,
/// 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<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft_accepted: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ttft_ms: Option<i64>,
}
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, au format des stats d'une complétion streamée. Jamais d'échec : un
/// compteur illisible reste inconnu. Rend aussi si la complétion est
/// complète (un usage, ou des timings).
pub fn wire_stats(raw: &[u8]) -> (bool, StatsEvent) {
let mut s = StatsEvent::default();
// Seule la première valeur JSON compte (`json.Decoder`).
let v = serde_json::Deserializer::from_slice(raw)
.into_iter::<Value>()
.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());
if let Some(t) = v.get("timings").filter(|t| t.is_object()) {
let n = |k: &str| perf_num(t.get(k));
s.prompt_tokens = n("prompt_n").unwrap_or(0.0) as i64;
s.prompt_ms = n("prompt_ms").unwrap_or(0.0);
s.prompt_per_second = n("prompt_per_second").unwrap_or(0.0);
s.gen_tokens = n("predicted_n").unwrap_or(0.0) as i64;
s.gen_per_second = n("predicted_per_second").unwrap_or(0.0);
s.cache_tokens = n("cache_n").map(|f| f as i64);
s.draft_n = n("draft_n").map(|f| f as i64);
s.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 {
s.prompt_tokens_total = perf_num(u.get("prompt_tokens")).unwrap_or(0.0) as i64;
if s.gen_tokens == 0 {
s.gen_tokens = perf_num(u.get("completion_tokens")).unwrap_or(0.0) as i64;
}
}
/// `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::<Value>()
.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
if s.cache_tokens.is_none() {
s.cache_tokens = 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);
}
(s.prompt_tokens_total > 0 || has_timings, s)
}
/// `perfNum` : un nombre (ou une chaîne qui en contient un), fini, positif
@@ -139,104 +74,82 @@ fn perf_num(v: Option<&Value>) -> Option<f64> {
(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<u64> {
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<i64>,
#[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 {
/// 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_post(
"/internal/v1/perf/record",
&json!({ "rec": rec, "prefix": prefix, "external": external }),
)
.internal_json(hyper::Method::POST, "/internal/v1/turn/perf", Some(&body))
.await
{
Ok((code, v)) if code.is_success() => serde_json::from_value(v).unwrap_or_default(),
_ => PerfNoted::default(),
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()
}
}
}
#[cfg(test)]
mod tests {
use super::super::turn::perf_prefix;
use super::super::types::Message;
use super::*;
#[test]
fn releve_d_une_reponse_entiere() {
let r = PerfRec::from_wire(
KIND_COMPACT,
"c1",
let (complete, s) = wire_stats(
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!(complete);
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,
s,
StatsEvent {
prompt_tokens_total: 1020,
cache_tokens: Some(900),
prompt_tokens: 120,
prompt_ms: 50.5,
prompt_per_second: 2376.0,
gen_tokens: 30,
gen_per_second: 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,
"",
let (complete, s) = wire_stats(
br#"{"usage":{"prompt_tokens":10,"completion_tokens":3,"prompt_tokens_details":{"cached_tokens":4}}}"#,
);
assert_eq!(
(r.complete, r.total, r.generated, r.cached),
(
complete,
s.prompt_tokens_total,
s.gen_tokens,
s.cache_tokens
),
(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));
let (complete, s) = wire_stats(b"pas du json");
assert!(!complete);
assert_eq!(s.cache_tokens, None);
let (complete, s) = wire_stats(br#"{"timings":null,"usage":{"prompt_tokens":1e13}}"#);
assert_eq!((complete, s.prompt_tokens_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));
let (complete, s) = wire_stats(br#"{"timings":"x"}"#);
assert_eq!((complete, s.prompt_tokens), (true, 0));
}
#[test]
@@ -246,34 +159,22 @@ mod tests {
content: Some(json!(c)),
..Default::default()
};
let a = prefix(&[m("a"), m("b")]);
let b = prefix(&[m("a"), m("b"), m("c")]);
let a = perf_prefix(&[m("a"), m("b")]);
let b = perf_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.
assert_ne!(perf_prefix(&[m("x"), m("b")])[1], a[1]);
assert!(perf_prefix(&[]).is_empty());
// FNV-1a de référence : le JSON du message, puis l'octet nul.
let mut h = 0xcbf2_9ce4_8422_2325u64;
fnv1a(&mut h, b"a");
for c in serde_json::to_vec(&m("a")).unwrap().into_iter().chain([0]) {
h = (h ^ u64::from(c)).wrapping_mul(0x0000_0100_0000_01b3);
}
assert_eq!(perf_prefix(&[m("a")]), vec![h]);
let mut h = 0xcbf2_9ce4_8422_2325u64;
for c in *b"a" {
h = (h ^ u64::from(c)).wrapping_mul(0x0000_0100_0000_01b3);
}
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 })
);
}
}
rust/loki-rs/src/chat/turn.rs
+81 -38
View File
@@ -23,15 +23,16 @@
//! `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.
//! Télémétrie : chaque complétion lue est rangée par le Go (`perfRecord`, par
//! [`TurnHost::perf_record`]) ; la dernière statistique d'une complétion
//! complète porte la nature et la perte de cache qu'il a calculées.
//!
//! 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, 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. Avec l'une de ces clés, le Go garde
//! la conversation (`facadeChatBlocker`, décidé au démarrage). Le comptage
//! des requêtes en vol du moteur passe par le Go (l'hôte autour de
//! [`TurnHost::send`]).
use std::collections::HashMap;
use std::future::Future;
@@ -57,14 +58,13 @@ 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,
};
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,
@@ -185,20 +185,56 @@ pub trait TurnHost: Send + Sync {
fn cache_hint(&self) -> Option<CacheHintFn> {
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<Output = super::perf::PerfNoted> + Send {
async { super::perf::PerfNoted::default() }
}
/// Garde d'écriture du mode Code (`codeWriteGuard`).
fn preflight(&self) -> Option<PreflightFn> {
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<Output = PerfMarks> + 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<u64>,
}
/// 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<i64>,
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<u64> {
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).
@@ -216,8 +252,6 @@ pub struct TurnConfig {
pub echo: EchoPolicy,
pub compact: CompactConfig,
pub patterns: Arc<RetryPatterns>,
/// 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
@@ -364,7 +398,7 @@ struct Loop {
echo_tpl_retry: bool,
tool_runs: i64,
budget_nudges: i64,
/// Complétions rangées dans la télémétrie (`perfIter`).
/// `perfIter` : complétions de ce tour, complètes ou non.
perf_iter: i64,
}
@@ -433,9 +467,8 @@ pub async fn run_turn<H: TurnHost>(
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, sent_prefix) = {
let (body, echo_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,
@@ -454,7 +487,7 @@ pub async fn run_turn<H: TurnHost>(
(
serde_json::to_vec(&payload).unwrap_or_default(),
echo_sent,
sent_prefix,
perf_prefix(&sent),
)
};
let t_req = Instant::now();
@@ -629,28 +662,38 @@ pub async fn run_turn<H: TurnHost>(
// 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;
host.perf_record(PerfRec {
iter: s.perf_iter,
external: ep.external,
..Default::default()
})
.await;
return TurnResult { extra, err: None };
}
let out = parser.finish(&mut emit);
// 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)
// 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.
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() },
})
.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.
// avec ce que la télémétrie en dit (nature, perte, alerte).
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;
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));
// Sonde de gabarit, en tâche de fond et après coup : rien n'en
// revient dans la requête suivante.
rust/loki-rs/src/chat/turn/tests.rs
+21 -30
View File
@@ -36,10 +36,9 @@ struct Fake {
unstable: Mutex<bool>,
/// Complétions signalées à la sonde : (outils, kwargs, effort).
completions: Mutex<Vec<Completion>>,
/// Télémétrie rangée : (relevé, empreintes, externe) ; et ce que le Go
/// en tire.
perfs: Mutex<Vec<(crate::chat::perf::PerfRec, Vec<u64>, bool)>>,
noted: Mutex<crate::chat::perf::PerfNoted>,
/// Télémétrie rangée ; et ce que le Go en tire.
perfs: Mutex<Vec<PerfRec>>,
marks: Mutex<PerfMarks>,
}
impl Fake {
@@ -147,17 +146,9 @@ impl TurnHost for Fake {
fn prefix_unstable(&self, _t: &[Tool], _k: Option<&Map<String, Value>>, _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()
async fn perf_record(&self, rec: PerfRec) -> PerfMarks {
self.perfs.lock().unwrap().push(rec);
self.marks.lock().unwrap().clone()
}
}
@@ -206,7 +197,6 @@ fn config(tools: &[&str]) -> TurnConfig {
echo: EchoPolicy::default(),
compact: CompactConfig::default(),
patterns: RetryPatterns::builtin(),
perf_kind: "main",
}
}
@@ -646,32 +636,32 @@ async fn flux_externe_coupe_repris_apres_le_texte_affiche() {
/// 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 {
*host.marks.lock().unwrap() = PerfMarks {
kind: "verify".into(),
lost: Some(40),
lost_after: "foreign".into(),
alert: true,
lost_alert: true,
};
let mut tc = config(&["glob"]);
tc.perf_kind = "verify";
let (r, evs) = run(&host, &tc, &Cancel::never()).await;
let (r, evs) = run(&host, &config(&["glob"]), &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");
for (i, rec) in perfs.iter().enumerate() {
assert_eq!(rec.iter, i as i64);
assert!(rec.complete);
assert_eq!((rec.total, rec.generated), (100, 5));
assert!(!external);
assert!(!prefix.is_empty());
assert_eq!(
(rec.stats.prompt_tokens_total, rec.stats.gen_tokens),
(100, 5)
);
assert!(!rec.external);
assert!(!rec.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);
let (a, b) = (&perfs[0].prefix, &perfs[1].prefix);
assert_eq!(a.len(), 1);
assert_eq!(b.len(), 3);
assert_eq!(b[..1], a[..]);
@@ -699,11 +689,12 @@ async fn telemetrie_des_completions() {
#[tokio::test]
async fn telemetrie_incomplete() {
let host = Fake::new(vec![Script::Cut(vec![content("Bon")])]);
host.marks.lock().unwrap().kind = "main".into();
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!(!perfs[0].complete);
assert!(perfs[0].prefix.is_empty());
assert!(
!evs.iter()
.any(|e| matches!(e, StreamEvent::Stats(s) if !s.kind.is_empty())),
rust/loki-rs/src/chat/verify.rs
+6 -1
View File
@@ -408,7 +408,6 @@ 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(),
@@ -429,6 +428,12 @@ 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 {
super::perf::KIND_VERIFY
} else {
super::perf::KIND_MAIN
},
};
let mut out = PassOut::default();
let mut cb = |ev: StreamEvent| -> bool {
rust/loki-rs/src/main.rs
+31 -6
View File
@@ -123,9 +123,9 @@ fn serve() -> ExitCode {
return ExitCode::FAILURE;
}
};
// LOKI_CHAT=rust : la conversation est tenue par la façade. Le Go le
// sait (il remet alors à la façade le trafic de chat du tunnel) et
// connaît son adresse locale.
// Conversation tenue par la façade (défaut ; LOKI_CHAT=go pour la
// rendre au Go). Le Go le sait (il remet alors à la façade le trafic
// de chat du tunnel) et connaît son adresse locale.
let chat_rust = chat_held_by_facade();
// LOKI_MODELS=rust : les routes des modèles aussi (le Go remet alors
// celles du tunnel et ne vide plus les .part).
@@ -414,9 +414,19 @@ fn serve() -> ExitCode {
})
}
/// `LOKI_CHAT=rust` : conversation tenue par la façade (Go par défaut).
/// Conversation tenue par la façade : c'est le défaut. `LOKI_CHAT=go` (ou
/// 0/off/false/no) la rend au Go — porte de sortie. Le Go la garde de toute
/// façon si un réglage que la façade ne sait pas honorer est posé
/// (`facadeChatBlocker`).
fn chat_held_by_facade() -> bool {
std::env::var("LOKI_CHAT").is_ok_and(|v| v.trim().eq_ignore_ascii_case("rust"))
chat_wanted(std::env::var("LOKI_CHAT").ok().as_deref())
}
fn chat_wanted(v: Option<&str>) -> bool {
!matches!(
v.map(|v| v.trim().to_ascii_lowercase()).as_deref(),
Some("go" | "0" | "off" | "false" | "no")
)
}
/// `LOKI_MODELS=rust` : routes des modèles tenues par la façade.
@@ -512,7 +522,7 @@ async fn start_chat(
s.load().await?;
let mut shared = ChatShared::for_sessions(&s, store.clone(), home.join("chatimg"));
shared.patterns = loki_rs::chat::generate::retry_patterns(&store).await;
eprintln!("[loki-rs] conversation tenue par la façade (LOKI_CHAT=rust)");
eprintln!("[loki-rs] conversation tenue par la façade (LOKI_CHAT=go pour la rendre au Go)");
Ok(ChatService::new(
std::sync::Arc::new(s),
std::sync::Arc::new(shared),
@@ -559,3 +569,18 @@ async fn shutdown_signal() {
let _ = tokio::signal::ctrl_c().await;
}
}
#[cfg(test)]
mod tests {
#[test]
fn conversation_tenue_par_la_facade_par_defaut() {
use super::chat_wanted;
assert!(chat_wanted(None));
assert!(chat_wanted(Some("")));
assert!(chat_wanted(Some("rust")));
assert!(chat_wanted(Some(" RUST ")));
for v in ["go", "GO", "0", "off", "false", "no"] {
assert!(!chat_wanted(Some(v)), "{v}");
}
}
}
rust/loki-rs/tests/common/fakego.rs
+41 -13
View File
@@ -125,7 +125,7 @@ pub struct FakeGo {
/// pour une action donnée.
pub service_os: Mutex<Vec<String>>,
pub service_os_fail: Mutex<Option<(String, String)>>,
/// /internal/v1/perf/record reçus (corps entiers).
/// /internal/v1/turn/perf reçus (corps entiers).
pub perf_records: Mutex<Vec<serde_json::Value>>,
}
@@ -502,14 +502,6 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
}
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(),
@@ -673,15 +665,51 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
"/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={}",
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();
rust/loki-rs/tests/lot6.rs
+55 -9
View File
@@ -692,6 +692,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(),
@@ -711,7 +712,6 @@ 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(),
@@ -767,15 +767,15 @@ async fn tour_complet_par_le_go() {
// 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]["complete"], false);
assert_eq!(perfs[0]["prefix"], serde_json::json!([]));
let r1 = &perfs[1]["rec"];
let r1 = &perfs[1];
assert_eq!(
(
r1["conv"].clone(),
r1["kind"].clone(),
r1["iter"].clone(),
r1["total"].clone()
r1["stats"]["prompt_tokens_total"].clone()
),
(
serde_json::json!("c7"),
@@ -794,7 +794,10 @@ async fn tour_complet_par_le_go() {
_ => None,
})
.unwrap();
assert_eq!((last.lost, last.lost_after.as_str()), (Some(12), "task"));
assert_eq!(
(last.kind.as_str(), last.lost, last.lost_after.as_str()),
("main", Some(3), "tâche")
);
}
#[tokio::test]
@@ -942,6 +945,47 @@ 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) ; 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:?}");
}
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.
@@ -1024,6 +1068,7 @@ async fn hote_et_sonde_de_gabarit() {
kwargs: Default::default(),
},
queued: Box::new(Vec::new),
perf_kind: "main",
}
};
let bash = Tool {
@@ -1848,10 +1893,11 @@ async fn telemetrie_des_resumes() {
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"],
perfs[0],
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 })
"external": false, "prefix": [],
"stats": { "prompt_tokens_total": 3000, "prompt_tokens": 3000, "prompt_ms": 1500.0,
"gen_tokens": 200, "gen_per_second": 40.0 } })
);
assert_eq!(perfs[0]["prefix"], serde_json::json!([]));
assert_eq!(perfs[1]["rec"]["complete"], false);
assert_eq!(perfs[1]["complete"], false);
}
rust/loki-rs/tests/parity.rs
-1
View File
@@ -195,7 +195,6 @@ async fn replay(fx: Fixture) -> Result<(), String> {
echo,
patterns: RetryPatterns::builtin(),
cfg,
perf_kind: "main",
};
let host = ParityHost {
replies: Mutex::new(fx.replies.into()),