Conversation tenue par Rust : télémétrie et requêtes en vol comme le Go

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 <noreply@anthropic.com>
2026-10-16 19:59:56 +02:00 · 2026-10-15 21:04:05 +02:00
1 parent 98196e1730
commit 5edcad50a5
@@ -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":
@@ -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)
+}
@@ -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,
+	})
+}
@@ -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)
+	}
+}
@@ -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.
@@ -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<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 == "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> ».
@@ -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<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),
@@ -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<PreflightFn> {
        None
    }
@@ -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<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>,
+    /// Requête de forme normale du moteur local : de quoi lancer la sonde du
+    /// gabarit (`tplProbeKick`).
+    pub probe: Option<PerfProbe>,
+}
+
+#[derive(Debug, Clone, PartialEq, Default)]
+pub struct PerfProbe {
+    pub tools: Vec<Tool>,
+    pub kwargs: Option<Map<String, Value>>,
+    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<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).
@@ -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<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) = {
+        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<H: TurnHost>(
                &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<H: TurnHost>(
            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 {
@@ -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 {
@@ -358,15 +358,48 @@ 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={} 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();
@@ -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.