Lot 33 : préchauffage (PREWARM) tenu par la façade

PREWARM ne garde plus la conversation au Go. La façade tient
l'historique : elle assemble et envoie la requête du prochain tour ; le
Go garde le compteur des requêtes en vol et ce qui l'annule.

Go (facade_prewarm.go) :
- /internal/v1/prewarm/check : moteur libre d'un vrai travail, loin de
  minuit, en bonne santé, slot visé ;
- /internal/v1/prewarm/begin : rechargement SLOT_PERSIST puis inscription
  comme LA requête en vol ; son annulation est relayée à la façade ;
- /internal/v1/prewarm/end : place rendue, cache repris noté ;
- /internal/v1/oai/begin : prewarm_keep pour l'étape qui le prolonge ;
- fin de tâche planifiée : le déclencheur part à la façade.

Rust (chat/prewarm.rs) :
- assemblage par les mêmes fonctions que le tour, sentinelle « . », un
  seul jeton, non streamé ; abstention (capacités inconnues, tour en
  cours, discussion vide, compaction attendue, assemblage inattendu) ;
- déclencheurs : fin de tour, fin de tâche, changement de discussion ou
  de projet (full : après 3 s) ;
- le tour garde le préchauffage qu'il prolonge (empreintes des messages
  et des options de rendu) et coupe tout de suite les autres ;
- télémétrie kind=prewarm.

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 02:52:44 +00:00
1 parent f79b74dac3
commit ff5c6d4571
21 files changed
+1457 -35

No files matched your search

+14 -4
View File
@@ -22,8 +22,9 @@
//! corrections du builder) suit le tour, avant la compaction de fin
//! ([`super::verify`]) ; toute compaction rend l'état des critères au résumé.
//!
//! Écarts assumés (restent au Go) : PROJ_SNAPSHOT (refusé par
//! `/turn/view`), préchauffage. Les images gardées (KEEP_TURN_IMAGES, lot 32)
//! Écart assumé (reste au Go) : PROJ_SNAPSHOT (refusé par `/turn/view`). Le
//! préchauffage (PREWARM, lot 33) part en fin de tour ([`super::prewarm`]).
//! Les images gardées (KEEP_TURN_IMAGES, lot 32)
//! partent en début et en fin de tour quand il le faut
//! ([`ChatService::kept_images_turn_start`]). La télémétrie des complétions
//! (tours, vérifications, résumés) rejoint l'anneau du Go
@@ -684,7 +685,7 @@ impl ChatService {
/// Changement de projet (`/api/projects`, action switch), remis par le
/// Go : refusé pendant un tour.
pub async fn project_switch(&self, slug: &str) -> Result<(), Refused> {
pub async fn project_switch(self: &Arc<Self>, slug: &str) -> Result<(), Refused> {
if self.sessions.conv().is_generating() {
return Err(Refused {
status: 409,
@@ -697,7 +698,9 @@ impl ChatService {
.map_err(|e| Refused {
status: 400,
error: e.to_string(),
})
})?;
self.prewarm_after_switch().await;
Ok(())
}
/// Suppression d'un projet (`deleteProject`) : ses discussions d'abord,
@@ -752,6 +755,9 @@ impl ChatService {
);
let local = !ep.external;
let conv_id = conv.id();
// Capacités demandées pour ce tour, retenues en mémoire (jamais
// persistées) pour préchauffer le suivant (PREWARM).
self.shared.prewarm.note_caps(&conv_id, prep.caps.clone());
let agent = prep.cap("Agent");
let mem_off = prep.turn_caps.get("Mem").and_then(Value::as_str) == Some("off");
let ccaps = CompactCaps {
@@ -1105,6 +1111,9 @@ impl ChatService {
{
eprintln!("[chat] fin de tour non signalée : {e}");
}
// Plus rien ne suit : le prochain tour peut être préparé pendant
// que l'utilisateur lit (PREWARM, off par défaut).
self.prewarm_kick(super::prewarm::TURN_END).await;
}
}
}
@@ -1160,6 +1169,7 @@ impl ChatShared {
images: super::host::ImageCache::new(img_dir),
patterns: super::stream::RetryPatterns::builtin(),
tpl: std::sync::Arc::new(super::tplprobe::TplProbe::new()),
prewarm: Default::default(),
}
}
}
+37 -10
View File
@@ -196,7 +196,7 @@ impl EngineSummarizer {
{
o.insert("id_slot".into(), Value::from(n));
}
_gate = engine_begin(store, "", slot == Some(1)).await;
_gate = engine_begin(store, "", slot == Some(1), false).await;
}
let body = serde_json::to_vec(&payload).unwrap_or_default();
let resp = self
@@ -328,6 +328,8 @@ pub struct ChatShared {
pub patterns: Arc<RetryPatterns>,
/// Sonde de gabarit (`chat_tplprobe.go`) : derniers verdicts et cache.
pub tpl: Arc<super::tplprobe::TplProbe>,
/// Préchauffage (PREWARM) : celui en vol, capacités des discussions.
pub prewarm: super::prewarm::Prewarm,
}
/// Ce que la compaction sait du tour.
@@ -397,15 +399,21 @@ pub async fn engine_slot_for(store: &Store, conv: &str, main: bool) -> Option<u8
/// Ouvre un ticket « en vol » auprès du Go (`engineRequestBeginSide`) : le
/// slot de la discussion n'est pas effacé pendant la requête, et un
/// préchauffage est annulé. `main_conv` : étape d'un tour de cette
/// discussion ; `side` : requête sur le second slot. `None` si le Go ne
/// répond pas — la requête part quand même.
pub async fn engine_begin(store: &Store, main_conv: &str, side: bool) -> Option<GateTicket> {
/// préchauffage est annulé — sauf `keep_prewarm`, une étape de tour qui le
/// prolonge. `main_conv` : étape d'un tour de cette discussion ; `side` :
/// requête sur le second slot. `None` si le Go ne répond pas — la requête
/// part quand même.
pub async fn engine_begin(
store: &Store,
main_conv: &str,
side: bool,
keep_prewarm: bool,
) -> Option<GateTicket> {
match store
.internal_json(
hyper::Method::POST,
"/internal/v1/oai/begin",
Some(&json!({ "main_conv": main_conv, "side": side })),
Some(&json!({ "main_conv": main_conv, "side": side, "prewarm_keep": keep_prewarm })),
)
.await
{
@@ -461,14 +469,14 @@ impl Drop for GateTicket {
}
impl GoTurnHost {
async fn gate_begin(&self) -> Option<GateTicket> {
async fn gate_begin(&self, keep_prewarm: bool) -> Option<GateTicket> {
let main_conv = if self.perf_kind == super::perf::KIND_MAIN {
self.conv_id.as_str()
} else {
""
};
let side = self.side_slot.load(Ordering::SeqCst);
engine_begin(&self.shared.store, main_conv, side).await
engine_begin(&self.shared.store, main_conv, side, keep_prewarm).await
}
}
@@ -527,13 +535,23 @@ 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> {
async fn send(
&self,
ep: &ChatEndpoint,
body: Vec<u8>,
keep_prewarm: bool,
) -> 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.
// Un préchauffage qu'elle ne prolonge pas est coupé tout de suite (le
// Go l'annulerait aussi, un aller-retour plus tard).
let gate = if ep.external {
None
} else {
self.gate_begin().await
if !keep_prewarm {
self.shared.prewarm.stop();
}
self.gate_begin(keep_prewarm).await
};
match self.shared.http.post(&ep.url, &ep.key, body).await {
Ok(resp) => {
@@ -720,6 +738,15 @@ impl TurnHost for GoTurnHost {
self.shared.echo_refusals.refuse(model);
}
fn prewarm_keeps(&self, prefix: &[u64], payload: &Value) -> bool {
self.perf_kind == super::perf::KIND_MAIN
&& self.shared.prewarm.live()
&& self
.shared
.prewarm
.keeps(prefix, super::prewarm::opts_hash(payload))
}
async fn keep_images(&self) -> Option<i64> {
// Fil principal d'une discussion seulement : la vérification ne
// persiste pas son historique.
+1
View File
@@ -33,6 +33,7 @@ pub mod lease;
pub mod nudges;
pub mod perf;
pub mod persist;
pub mod prewarm;
pub mod sessions;
pub mod stream;
pub mod tplprobe;
+1
View File
@@ -19,6 +19,7 @@ use crate::store::Store;
pub const KIND_MAIN: &str = "main";
pub const KIND_VERIFY: &str = "verify";
pub const KIND_COMPACT: &str = "compact";
pub const KIND_PREWARM: &str = "prewarm";
/// `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
+513
View File
@@ -0,0 +1,513 @@
//! PREWARM tenu par la façade (lot 33) — port de `chat_prewarm.go`.
//!
//! Préparer le prochain tour pendant que l'utilisateur lit. Certains
//! recalculs sont inévitables (prompt réécrit par une compaction, dernier
//! message assistant rendu autrement, discussion reprise après une tâche qui
//! a pris le slot) ; sans la clé, ils tombent sur le premier jeton du message
//! suivant.
//!
//! Avec la clé, la façade envoie au moteur local, dès qu'il est libre, la
//! requête du prochain tour telle qu'elle partira — assemblée par les MÊMES
//! fonctions que le tour — suivie d'un message utilisateur sentinelle « . »,
//! avec `max_tokens` 1. Le moteur calcule tout le préfixe ; le vrai tour ne
//! calcule plus que son propre message. Rien n'est persisté.
//!
//! Le Go garde ce qui décide si le préchauffage peut partir et ce qui
//! l'annule (`facade_prewarm.go`) : il l'inscrit comme LA requête en vol, et
//! toute requête de Loki qui part ensuite l'annule — sauf un tour de la
//! façade dont la requête le PROLONGE exactement (empreintes des messages,
//! des outils et des arguments du gabarit), qui le suit dans le même slot.
//! Aucune erreur ne remonte.
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use serde_json::{Value, json};
use tokio::sync::watch;
use super::compact::{compact_needed, ctx_pending, strip_context_updates};
use super::generate::ChatService;
use super::history::inject_prefix;
use super::perf::{self, wire_stats};
use super::turn::{PerfRec, perf_prefix};
use super::types::Message;
use super::wire::{
ChatPayloadOpts, build_chat_payload, turn_echo_policy, turn_reasoning, wire_messages,
};
use crate::engstatus::is_external;
/// Le message utilisateur qui tient la place du vrai.
pub const SENTINEL: &str = ".";
/// Déclencheurs (pour le journal).
pub const TURN_END: &str = "fin-tour";
pub const AFTER_TASK: &str = "après-tâche";
pub const SWITCH: &str = "discussion";
/// PREWARM=full : préparer la discussion ouverte si l'utilisateur y reste.
pub const SWITCH_DELAY: Duration = Duration::from_secs(3);
/// Borne d'un préchauffage (un MoE aux experts en RAM met des minutes sur 20k
/// jetons). Toute vraie requête l'annule bien avant.
pub const TIMEOUT: Duration = Duration::from_secs(15 * 60);
/// Attente de la fin d'un préchauffage remplacé.
const STOP_WAIT: Duration = Duration::from_secs(5);
/// Capacités gardées, une par discussion.
const CAPS_MAX: usize = 64;
/// `prewarmMode` : « on », « full » ou "" (off). Jamais pour un preset externe.
pub fn prewarm_mode(cfg: &HashMap<String, String>) -> &'static str {
if is_external(cfg) {
return "";
}
match cfg
.get("PREWARM")
.map(|v| v.trim().to_lowercase())
.as_deref()
{
Some("on" | "1" | "true" | "yes" | "oui") => "on",
Some("full") => "full",
_ => "",
}
}
/// `prewarmOptsKeys` : les champs du corps, hors messages, qui changent le
/// prompt rendu par le gabarit (l'échantillonnage n'y touche pas).
const OPTS_KEYS: [&str; 5] = [
"tools",
"parallel_tool_calls",
"tool_choice",
"chat_template_kwargs",
"reasoning_effort",
];
/// `prewarmOptsHash` : FNV-1a 64 de ces champs. Comparée seulement aux
/// empreintes calculées ici, d'un corps à l'autre.
pub fn opts_hash(payload: &Value) -> u64 {
const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const PRIME: u64 = 0x0000_0100_0000_01b3;
let mut h = OFFSET;
let mut eat = |b: &[u8]| {
for &c in b {
h = (h ^ u64::from(c)).wrapping_mul(PRIME);
}
};
for k in OPTS_KEYS {
eat(k.as_bytes());
eat(&[0]);
if let Some(v) = payload.get(k) {
eat(&serde_json::to_vec(v).unwrap_or_default());
}
eat(&[0]);
}
h
}
/// `withTurnRole(caps, ".")` : en mode Code, le rôle d'un message ordinaire.
pub fn sentinel_caps(caps: &Value) -> Value {
let mut c = caps.clone();
let code = c.get("Code").and_then(Value::as_bool).unwrap_or(false);
let role_empty = c
.get("Role")
.and_then(Value::as_str)
.unwrap_or("")
.is_empty();
if code
&& role_empty
&& let Some(o) = c.as_object_mut()
{
o.insert("Role".into(), Value::from("builder"));
}
c
}
/// Un préchauffage inscrit au Go : `prefix` couvre ses `n` premiers messages
/// (sans la sentinelle), `opts` les champs qui façonnent le rendu.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Run {
pub id: u64,
pub n: usize,
pub prefix: u64,
pub opts: u64,
}
impl Run {
/// `matches` : la requête dont `prefix` est l'empreinte cumulée des
/// messages prolonge-t-elle exactement ce préchauffage ?
pub fn matches(&self, prefix: &[u64], opts: u64) -> bool {
self.n > 0
&& prefix.len() > self.n
&& prefix[self.n - 1] == self.prefix
&& opts == self.opts
}
}
struct Current {
/// Rempli une fois inscrit au Go.
run: Option<Run>,
cancel: watch::Sender<bool>,
done: watch::Receiver<bool>,
}
/// L'état partagé du préchauffage (un exemplaire par processus).
#[derive(Default)]
pub struct Prewarm {
cur: Mutex<Option<Current>>,
/// Capacités du dernier tour de chaque discussion, en mémoire seulement.
caps: Mutex<HashMap<String, Value>>,
/// Raisons d'échec déjà journalisées.
logged: Mutex<HashSet<String>>,
}
impl Prewarm {
fn cur(&self) -> std::sync::MutexGuard<'_, Option<Current>> {
self.cur.lock().unwrap_or_else(|e| e.into_inner())
}
/// Un préchauffage est-il en vol (ou en train de partir) ?
pub fn live(&self) -> bool {
self.cur().is_some()
}
/// La requête d'un tour le prolonge-t-elle exactement ?
pub fn keeps(&self, prefix: &[u64], opts: u64) -> bool {
self.cur()
.as_ref()
.and_then(|c| c.run)
.is_some_and(|r| r.matches(prefix, opts))
}
/// `prewarmStop` : annule le préchauffage en vol et rend de quoi attendre
/// sa fin (`None` : rien en vol).
pub fn stop(&self) -> Option<watch::Receiver<bool>> {
let c = self.cur().take()?;
let _ = c.cancel.send(true);
Some(c.done)
}
/// Annulation relayée par le Go (une autre requête est partie). Un
/// préchauffage pas encore inscrit est annulé aussi : il n'y en a qu'un.
pub fn cancel_id(&self, id: u64) -> bool {
let g = self.cur();
match g.as_ref() {
Some(c) if c.run.is_none_or(|r| r.id == id) => {
let _ = c.cancel.send(true);
true
}
_ => false,
}
}
/// `prewarmNoteCaps`.
pub fn note_caps(&self, conv: &str, caps: Value) {
if conv.is_empty() {
return;
}
let mut m = self.caps.lock().unwrap_or_else(|e| e.into_inner());
if !m.contains_key(conv)
&& m.len() >= CAPS_MAX
&& let Some(k) = m.keys().next().cloned()
{
m.remove(&k);
}
m.insert(conv.to_string(), caps);
}
pub fn caps_for(&self, conv: &str) -> Option<Value> {
self.caps
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(conv)
.cloned()
}
fn log_once(&self, key: &str, msg: &str) {
if self
.logged
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(key.to_string())
{
eprintln!("[prewarm] {msg}");
}
}
/// Inscrit le préchauffage qui part ; `None` si un autre est déjà là.
fn register(&self) -> Option<(watch::Receiver<bool>, watch::Sender<bool>)> {
let mut g = self.cur();
if g.is_some() {
return None;
}
let (cancel, cancelled) = watch::channel(false);
let (done_tx, done) = watch::channel(false);
*g = Some(Current {
run: None,
cancel,
done,
});
Some((cancelled, done_tx))
}
fn set_run(&self, run: Run) {
if let Some(c) = self.cur().as_mut() {
c.run = Some(run);
}
}
/// Fin du préchauffage `run` (ou de celui qui partait) : la place est
/// rendue s'il est toujours le courant.
fn release(&self, id: Option<u64>) {
let mut g = self.cur();
if g.as_ref().is_some_and(|c| c.run.map(|r| r.id) == id) {
*g = None;
}
}
}
/// Ce que l'assemblage rend : le corps, et l'empreinte du préfixe préchauffé.
#[derive(Debug, Clone, PartialEq)]
pub struct Body {
pub payload: Value,
pub n: usize,
pub prefix: u64,
pub opts: u64,
}
impl ChatService {
/// `prewarmKick` : un déclencheur. Sans la clé, rien — ni tâche, ni
/// requête.
pub async fn prewarm_kick(self: &Arc<Self>, why: &'static str) {
let cfg = self.shared.store.config().await;
if prewarm_mode(&cfg).is_empty() {
return;
}
let me = self.clone();
tokio::spawn(async move { me.prewarm(why).await });
}
/// `prewarmAfterSwitch` : changement de discussion. Le préchauffage de
/// l'ancienne est annulé ; avec PREWARM=full, la nouvelle est préparée si
/// l'utilisateur y reste 3 s.
pub async fn prewarm_after_switch(self: &Arc<Self>) {
self.shared.prewarm.stop(); // sa fin n'est pas attendue
let cfg = self.shared.store.config().await;
if prewarm_mode(&cfg) != "full" {
return;
}
let id = self.sessions.conv().id();
let me = self.clone();
tokio::spawn(async move {
tokio::time::sleep(SWITCH_DELAY).await;
if me.sessions.conv().id() == id {
me.prewarm_kick(SWITCH).await;
}
});
}
/// `prewarmBodyFor` : le corps du préchauffage de la discussion ouverte,
/// ou la raison de s'abstenir. Lecture seule.
pub async fn prewarm_body(&self) -> Result<Body, &'static str> {
let env = self.turn_env().await;
let conv_id = self.sessions.conv().id();
let caps = self
.shared
.prewarm
.caps_for(&conv_id)
.ok_or("capacités du dernier tour inconnues")?;
let turn_caps = sentinel_caps(&caps);
let conv = self.sessions.conv();
if conv.is_generating() {
return Err("moteur occupé");
}
let (mut msgs, ctx_now, peak) = conv.with_state(|st| {
let msgs = st.messages.clone();
let used = ctx_pending(st.ctx_used, st.ctx_used_len, &msgs, &env.echo);
(msgs, used, st.gen_peak)
});
// Sans message utilisateur, le contexte projet irait en tête de la
// sentinelle : rien à préparer.
if !msgs.iter().any(|m| m.role == "user") {
return Err("discussion vide");
}
msgs.push(Message {
role: "user".into(),
content: Some(Value::from(SENTINEL)),
..Default::default()
});
// Le prochain tour compactera d'abord : son prompt ne sera pas celui-ci.
if compact_needed(&env.ccfg, &msgs, ctx_now, peak) {
return Err("compaction attendue au prochain tour");
}
let (head, prefix, tools) = self
.turn_view(&turn_caps)
.await
.map_err(|_| "vue du tour refusée")?;
let mut sent = head;
sent.extend(strip_context_updates(msgs));
let sent = inject_prefix(sent, &prefix);
let reasoning = turn_reasoning(&env.cfg, &self.shared.effort);
let pol = turn_echo_policy(false, || env.echo.clone());
self.shared.images.prefetch(&self.shared.store, &sent).await;
let (wire, _) = wire_messages(&sent, false, &pol);
let n = wire.len();
let last_ok = wire.last().is_some_and(|m| {
m.role == "user" && m.content.as_ref().and_then(Value::as_str) == Some(SENTINEL)
});
if n < 2 || !last_ok {
return Err("assemblage inattendu");
}
let mut payload = build_chat_payload(
&env.ep,
&wire,
0.7,
&ChatPayloadOpts {
tools: &tools,
disable_tools: false,
tool_choice_none: false,
effort: &reasoning.effort,
kwargs: reasoning.kwargs.as_ref(),
no_sampling: false,
},
&env.cfg,
&|r| self.shared.images.get(r),
);
// Seuls les champs de réponse changent : un seul jeton, non streamé.
if let Some(o) = payload.as_object_mut() {
o.insert("stream".into(), Value::Bool(false));
o.remove("stream_options");
o.insert("max_tokens".into(), json!(1));
}
let opts = opts_hash(&payload);
Ok(Body {
payload,
n: n - 1,
prefix: perf_prefix(&wire)[n - 2],
opts,
})
}
/// `prewarm` : un préchauffage complet, ou rien. Jamais d'erreur rendue.
pub async fn prewarm(self: &Arc<Self>, why: &str) {
let cfg = self.shared.store.config().await;
if prewarm_mode(&cfg).is_empty() {
return;
}
let env = self.turn_env().await;
if env.ep.external {
return;
}
// Un seul en vol : le nouveau déclencheur remplace l'ancien.
if let Some(mut done) = self.shared.prewarm.stop() {
let ended = tokio::time::timeout(STOP_WAIT, done.wait_for(|d| *d)).await;
if !matches!(ended, Ok(Ok(_))) {
return;
}
}
let store = &self.shared.store;
let slot = match store
.internal_post("/internal/v1/prewarm/check", &json!({}))
.await
{
Ok((code, v)) if code.is_success() && v["ok"] == true => {
v["slot"].as_i64().unwrap_or(-1)
}
_ => return,
};
let Ok(mut body) = self.prewarm_body().await else {
return;
};
if slot >= 0
&& let Some(o) = body.payload.as_object_mut()
{
o.insert("id_slot".into(), json!(slot));
}
let Ok(raw_body) = serde_json::to_vec(&body.payload) else {
return;
};
let Some((mut cancelled, done)) = self.shared.prewarm.register() else {
return;
};
let conv_id = self.sessions.conv().id();
let begun = store
.internal_post(
"/internal/v1/prewarm/begin",
&json!({ "conv": conv_id, "n": body.n, "prefix": body.prefix, "opts": body.opts }),
)
.await;
let id = match begun {
Ok((code, v)) if code.is_success() && v["ok"] == true => v["id"].as_u64(),
_ => None,
};
let Some(id) = id else {
// Une requête de Loki est partie entre-temps.
self.shared.prewarm.release(None);
let _ = done.send(true);
return;
};
self.shared.prewarm.set_run(Run {
id,
n: body.n,
prefix: body.prefix,
opts: body.opts,
});
let (mut total, mut cached) = (0i64, None::<i64>);
if !*cancelled.borrow() {
let send = async {
let resp = self
.shared
.http
.post(&env.ep.url, &env.ep.key, raw_body)
.await;
match resp {
Err(e) => Err(format!("moteur injoignable ({why}) : {e}")),
Ok(r) => {
let status = r.status();
let raw = r.bytes().await.map_err(|e| e.to_string())?;
let raw = &raw[..raw.len().min(1 << 20)];
Ok((status, raw.to_vec()))
}
}
};
let out: Option<Result<(reqwest::StatusCode, Vec<u8>), String>> = tokio::select! {
o = tokio::time::timeout(TIMEOUT, send) => o.ok(),
_ = cancelled.wait_for(|c| *c) => None,
};
match out {
Some(Err(e)) => self.shared.prewarm.log_once("net", &e),
Some(Ok((status, raw))) if status.as_u16() != 200 => {
let txt = String::from_utf8_lossy(&raw[..raw.len().min(200)])
.trim()
.to_string();
self.shared.prewarm.log_once(
status.as_str(),
&format!("refusé par le moteur ({why}) : {status} {txt}"),
);
}
Some(Ok((_, raw))) => {
// Télémétrie seulement (kind=prewarm) : la réponse est jetée.
let (complete, stats) = wire_stats(&raw);
total = stats.prompt_tokens_total;
cached = stats.cache_tokens;
let rec = PerfRec {
complete,
stats,
..Default::default()
};
perf::record(store, perf::KIND_PREWARM, &conv_id, &rec).await;
}
None => {}
}
}
let _ = store
.internal_post(
"/internal/v1/prewarm/end",
&json!({ "id": id, "total": total, "cached": cached }),
)
.await;
self.shared.prewarm.release(Some(id));
let _ = done.send(true);
}
}
#[cfg(test)]
mod tests;
+167
View File
@@ -0,0 +1,167 @@
//! Tests du préchauffage (cas de `chat_prewarm_test.go`).
use super::*;
use serde_json::json;
fn cfg(kv: &[(&str, &str)]) -> HashMap<String, String> {
kv.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
fn um(s: &str) -> Message {
Message {
role: "user".into(),
content: Some(json!(s)),
..Default::default()
}
}
#[test]
fn mode() {
for (v, want) in [
("on", "on"),
(" OUI ", "on"),
("1", "on"),
("full", "full"),
("FULL", "full"),
("off", ""),
("", ""),
] {
assert_eq!(prewarm_mode(&cfg(&[("PREWARM", v)])), want, "{v:?}");
}
assert_eq!(prewarm_mode(&cfg(&[])), "");
// Jamais pour un preset externe.
assert_eq!(
prewarm_mode(&cfg(&[
("PREWARM", "full"),
(crate::engstatus::EXT_KEY_FLAG, "1")
])),
""
);
}
/// Seuls les champs qui changent le prompt rendu comptent ; l'échantillonnage
/// et les champs de réponse, non.
#[test]
fn empreinte_des_options() {
let base = json!({
"model": "m", "messages": [], "stream": true, "temperature": 0.7,
"tools": [{"type": "function", "function": {"name": "bash"}}],
"tool_choice": "auto", "chat_template_kwargs": {"enable_thinking": true}
});
let h = opts_hash(&base);
let mut same = base.clone();
same["temperature"] = json!(0.2);
same["stream"] = json!(false);
same["max_tokens"] = json!(1);
same["id_slot"] = json!(0);
same["messages"] = json!([{"role": "user", "content": "x"}]);
assert_eq!(opts_hash(&same), h);
for (k, v) in [
("tools", json!([])),
("tool_choice", json!("none")),
("chat_template_kwargs", json!({"enable_thinking": false})),
("reasoning_effort", json!("low")),
("parallel_tool_calls", json!(true)),
] {
let mut other = base.clone();
other[k] = v;
assert_ne!(opts_hash(&other), h, "{k}");
}
// Un champ absent n'est pas un champ vide.
let mut absent = base.clone();
absent.as_object_mut().unwrap().remove("tool_choice");
let mut empty = base.clone();
empty["tool_choice"] = json!("");
assert_ne!(opts_hash(&absent), opts_hash(&empty));
}
#[test]
fn role_de_la_sentinelle() {
assert_eq!(
sentinel_caps(&json!({"Code": true, "Role": ""}))["Role"],
"builder"
);
assert_eq!(sentinel_caps(&json!({"Code": true}))["Role"], "builder");
assert_eq!(
sentinel_caps(&json!({"Code": true, "Role": "verifier"}))["Role"],
"verifier"
);
assert!(sentinel_caps(&json!({"Agent": true})).get("Role").is_none());
}
/// Une requête prolonge le préchauffage quand ses n premiers messages ont la
/// même empreinte, qu'elle en a au moins un de plus, et mêmes options.
#[test]
fn prolongement() {
let wire = vec![um("a"), um("b"), um(SENTINEL)];
let pre = perf_prefix(&wire);
let run = Run {
id: 1,
n: 2,
prefix: pre[1],
opts: 7,
};
let real = perf_prefix(&[um("a"), um("b"), um("vraie question")]);
assert!(run.matches(&real, 7));
assert!(!run.matches(&real, 8), "autres options");
assert!(!run.matches(&real[..2], 7), "rien de plus que le préfixe");
let other = perf_prefix(&[um("a"), um("c"), um("vraie question")]);
assert!(!run.matches(&other, 7), "préfixe différent");
let longer = perf_prefix(&[um("a"), um("b"), um("x"), um("y")]);
assert!(run.matches(&longer, 7));
assert!(!Run { n: 0, ..run }.matches(&real, 7));
}
/// Un seul en vol ; gardé par la requête qui le prolonge ; annulé (par le
/// Go ou par une autre requête) ; la place rendue à sa fin seulement.
#[test]
fn etat_partage() {
let p = Prewarm::default();
assert!(!p.live());
let (cancelled, done) = p.register().expect("libre");
assert!(p.live());
assert!(p.register().is_none(), "deux en vol");
// Pas encore inscrit au Go : rien à garder, mais annulable.
assert!(!p.keeps(&[1, 2, 3], 7));
p.set_run(Run {
id: 4,
n: 2,
prefix: 2,
opts: 7,
});
assert!(p.keeps(&[1, 2, 3], 7));
assert!(!p.keeps(&[1, 2, 3], 8));
assert!(!p.cancel_id(9), "annulation d'un autre préchauffage");
assert!(!*cancelled.borrow());
assert!(p.cancel_id(4));
assert!(*cancelled.borrow());
assert!(p.live(), "la place reste prise jusqu'à la fin");
p.release(Some(9));
assert!(p.live(), "fin d'un autre préchauffage");
p.release(Some(4));
assert!(!p.live());
drop(done);
// stop : annule et rend de quoi attendre la fin.
let (cancelled, done) = p.register().unwrap();
let mut wait = p.stop().expect("en vol");
assert!(*cancelled.borrow() && !p.live());
let _ = done.send(true);
assert!(*wait.borrow_and_update());
assert!(p.stop().is_none());
}
#[test]
fn capacites_par_discussion() {
let p = Prewarm::default();
p.note_caps("", json!({"Agent": true}));
assert!(p.caps_for("").is_none());
p.note_caps("c1", json!({"Agent": true}));
p.note_caps("c1", json!({"Agent": false}));
assert_eq!(p.caps_for("c1"), Some(json!({"Agent": false})));
for i in 0..CAPS_MAX + 10 {
p.note_caps(&format!("x{i}"), json!({}));
}
assert_eq!(p.caps.lock().unwrap().len(), CAPS_MAX);
}
+21 -7
View File
@@ -38,9 +38,12 @@
//! est gardée à l'essai jusqu'à la mesure de son coût par le moteur
//! ([`super::keepimg`]) ; les images gardées partent avant tout résumé.
//!
//! Non portés (clés opt-in, ou machinerie du moteur local restée au Go) :
//! PROJ_SNAPSHOT, préchauffage. Avec l'une de ces clés, le Go garde la
//! conversation (`facadeChatBlocker`, décidé au démarrage).
//! Préchauffage (PREWARM, lot 33) : une requête qui prolonge exactement le
//! préchauffage en vol le garde ([`TurnHost::prewarm_keeps`]) ; toute autre
//! l'annule ([`super::prewarm`]).
//!
//! Non porté : PROJ_SNAPSHOT. Avec cette clé, le Go garde la conversation
//! (`facadeChatBlocker`, décidé au démarrage).
use std::collections::HashMap;
use std::future::Future;
@@ -138,12 +141,20 @@ pub enum CompactPhase {
/// Les effets d'un tour, fournis par l'appelant.
pub trait TurnHost: Send + Sync {
/// POST du corps vers l'endpoint (sa clé à lui).
/// POST du corps vers l'endpoint (sa clé à lui). `keep_prewarm` : la
/// requête prolonge exactement le préchauffage en vol, qui n'est pas
/// annulé ([`Self::prewarm_keeps`]).
fn send(
&self,
ep: &ChatEndpoint,
body: Vec<u8>,
keep_prewarm: bool,
) -> impl Future<Output = Result<Reply, SendError>> + Send;
/// PREWARM : la requête qui part (empreintes cumulées de ses messages,
/// corps) prolonge-t-elle exactement le préchauffage en vol ?
fn prewarm_keeps(&self, _prefix: &[u64], _payload: &Value) -> bool {
false
}
/// Exécute un appel d'outil (`execTool`).
fn exec_tool(
&self,
@@ -541,7 +552,7 @@ async fn run_turn_loop<H: TurnHost>(
let pol = turn_echo_policy(ep.external || s.echo_off_turn, || tc.echo.clone());
host.prepare_images(&messages).await;
let slot = host.engine_slot(ep).await;
let (body, echo_sent, prefix) = {
let (body, echo_sent, prefix, keep_prewarm) = {
let (sent, echo_sent) = wire_messages(&messages, s.tool_choice_none, &pol);
let mut payload = build_chat_payload(
ep,
@@ -564,10 +575,13 @@ async fn run_turn_loop<H: TurnHost>(
{
o.insert("id_slot".into(), Value::from(n));
}
let prefix = perf_prefix(&sent);
let keep = !ep.external && host.prewarm_keeps(&prefix, &payload);
(
serde_json::to_vec(&payload).unwrap_or_default(),
echo_sent,
perf_prefix(&sent),
prefix,
keep,
)
};
let t_req = Instant::now();
@@ -575,7 +589,7 @@ async fn run_turn_loop<H: TurnHost>(
if let Some(k) = kt.as_mut() {
k.sending(messages.len());
}
let reply = match cancel.or_stop(host.send(ep, body)).await {
let reply = match cancel.or_stop(host.send(ep, body, keep_prewarm)).await {
None => return TurnResult { extra, err: None },
Some(Ok(r)) => r,
Some(Err(e)) => {
+40 -1
View File
@@ -48,6 +48,10 @@ struct Fake {
relay: Mutex<Option<Message>>,
keep_stores: Mutex<usize>,
keep_refuse: Mutex<bool>,
/// PREWARM : la requête prolonge-t-elle le préchauffage ? ce que `send`
/// a reçu.
prewarm_match: Mutex<bool>,
sent_keeps: Mutex<Vec<bool>>,
}
impl Fake {
@@ -77,7 +81,13 @@ fn stream_of(lines: Vec<String>, cut: bool) -> BodyStream {
}
impl TurnHost for Fake {
async fn send(&self, _ep: &ChatEndpoint, body: Vec<u8>) -> Result<Reply, SendError> {
async fn send(
&self,
_ep: &ChatEndpoint,
body: Vec<u8>,
keep: bool,
) -> Result<Reply, SendError> {
self.sent_keeps.lock().unwrap().push(keep);
self.bodies
.lock()
.unwrap()
@@ -163,6 +173,12 @@ impl TurnHost for Fake {
async fn keep_images(&self) -> Option<i64> {
*self.keep.lock().unwrap()
}
fn prewarm_keeps(&self, prefix: &[u64], payload: &Value) -> bool {
// Ce que l'hôte reçoit : les empreintes des messages envoyés et le
// corps de la requête.
assert_eq!(prefix.len(), payload["messages"].as_array().unwrap().len());
*self.prewarm_match.lock().unwrap()
}
async fn keep_image(&self, m: &Message) -> Option<(Message, i64)> {
*self.keep_stores.lock().unwrap() += 1;
if *self.keep_refuse.lock().unwrap() {
@@ -1055,3 +1071,26 @@ async fn images_retirees_fenetre_pleine() {
assert!(evs.iter().any(|e| matches!(e, StreamEvent::DropReasoning)));
assert_eq!(text(&evs), "fini");
}
/// PREWARM : la boucle demande à l'hôte si la requête prolonge le
/// préchauffage en vol et le lui transmet à l'envoi ; jamais vers une API
/// externe.
#[tokio::test]
async fn prechauffage_garde_par_la_requete_qui_le_prolonge() {
let host = Fake::new(vec![answer("ok")]);
*host.prewarm_match.lock().unwrap() = true;
let (r, _) = run(&host, &config(&[]), &Cancel::never()).await;
assert_eq!(r.err, None);
assert_eq!(*host.sent_keeps.lock().unwrap(), [true]);
let host = Fake::new(vec![answer("ok")]);
let (r, _) = run(&host, &config(&[]), &Cancel::never()).await;
assert_eq!(r.err, None);
assert_eq!(*host.sent_keeps.lock().unwrap(), [false]);
let host = Fake::new(vec![answer("ok")]);
*host.prewarm_match.lock().unwrap() = true;
let mut tc = config(&[]);
tc.ep.external = true;
let (r, _) = run(&host, &tc, &Cancel::never()).await;
assert_eq!(r.err, None);
assert_eq!(*host.sent_keeps.lock().unwrap(), [false]);
}
+32 -1
View File
@@ -66,6 +66,8 @@ pub fn is_internal(path: &str) -> bool {
| "/internal/facade/v1/project/switch"
| "/internal/facade/v1/project/drop"
| "/internal/facade/v1/chat/tpl"
| "/internal/facade/v1/chat/prewarm"
| "/internal/facade/v1/chat/prewarm/cancel"
)
}
@@ -107,6 +109,27 @@ pub async fn serve_internal(chat: &Arc<ChatService>, req: Request<Incoming>) ->
None => send_json(StatusCode::OK, &json!({ "ok": false })),
}
}
// PREWARM (lot 33) : déclencheur venu du Go (fin d'une tâche
// planifiée), annulation relayée (une autre requête est partie).
"/internal/facade/v1/chat/prewarm" => {
let v = read_json(req.into_body()).await.unwrap_or_default();
let why = match str_of(&v, "why") {
crate::chat::prewarm::AFTER_TASK => crate::chat::prewarm::AFTER_TASK,
crate::chat::prewarm::SWITCH => crate::chat::prewarm::SWITCH,
_ => crate::chat::prewarm::TURN_END,
};
chat.prewarm_kick(why).await;
send_json(StatusCode::OK, &json!({ "ok": true }))
}
"/internal/facade/v1/chat/prewarm/cancel" => {
let v = read_json(req.into_body()).await.unwrap_or_default();
let id = v.get("id").and_then(Value::as_u64).unwrap_or(0);
let cancelled = chat.shared.prewarm.cancel_id(id);
send_json(
StatusCode::OK,
&json!({ "ok": true, "cancelled": cancelled }),
)
}
"/internal/facade/v1/project/switch" | "/internal/facade/v1/project/drop" => {
let v = read_json(req.into_body()).await.unwrap_or_default();
let slug = str_of(&v, "slug");
@@ -238,8 +261,16 @@ pub async fn serve(
// Corps illisible : champs vides, comme le `_ = Decode` du Go.
"/api/conversations/switch" => {
let v = read_json(body).await.unwrap_or_default();
let before = chat.sessions.conv().id();
match chat.sessions.switch(str_of(&v, "id")).await {
Ok(()) => ok(),
Ok(()) => {
// PREWARM : le préchauffage de l'ancienne est annulé ; avec
// « full », la nouvelle est préparée si on y reste.
if chat.sessions.conv().id() != before {
chat.prewarm_after_switch().await;
}
ok()
}
Err(e) => session_error(e),
}
}
+33 -1
View File
@@ -137,6 +137,10 @@ pub struct FakeGo {
/// /internal/v1/turn/keepimg : vision active ; rangements reçus.
pub keep_vision: Mutex<bool>,
pub keep_stores: Mutex<Vec<serde_json::Value>>,
/// /internal/v1/prewarm/* : refus du contrôle (raison), appels reçus
/// (« check », « begin {corps} », « end {corps} »).
pub prewarm_refuse: Mutex<Option<String>>,
pub prewarm_log: Mutex<Vec<String>>,
}
impl FakeGo {
@@ -680,10 +684,15 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
if let Some(conv) = v["main_conv"].as_str() {
let side = if v["side"] == true { " side" } else { "" };
let keep = if v["prewarm_keep"] == true {
" prewarm_keep"
} else {
""
};
st.oai_log
.lock()
.unwrap()
.push(format!("begin main_conv={conv}{side}"));
.push(format!("begin main_conv={conv}{side}{keep}"));
} else {
st.oai_log.lock().unwrap().push(format!(
"begin read={} side={} post_v1={}",
@@ -695,6 +704,29 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
}
resp(200, r#"{"ticket":7}"#)
}
"/internal/v1/prewarm/check" => {
st.prewarm_log.lock().unwrap().push("check".into());
let v = match st.prewarm_refuse.lock().unwrap().clone() {
Some(why) => serde_json::json!({ "ok": false, "why": why }),
None => serde_json::json!({ "ok": true, "slot": -1 }),
};
resp(200, v.to_string())
}
"/internal/v1/prewarm/begin" | "/internal/v1/prewarm/end" => {
let body = req.into_body().collect().await.unwrap().to_bytes();
let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
let what = if path.ends_with("begin") {
"begin"
} else {
"end"
};
st.prewarm_log.lock().unwrap().push(format!("{what} {v}"));
if what == "begin" {
resp(200, r#"{"ok":true,"id":5}"#)
} else {
resp(204, "")
}
}
"/internal/v1/turn/keepimg" => {
let v = serde_json::json!({
"vision": *st.keep_vision.lock().unwrap(),
+133
View File
@@ -666,6 +666,7 @@ async fn tour_complet_par_le_go() {
images: ImageCache::new(imgdir),
patterns: RetryPatterns::builtin(),
tpl: Arc::new(loki_rs::chat::tplprobe::TplProbe::new()),
prewarm: Default::default(),
});
let cfg = std::collections::HashMap::new();
let host = GoTurnHost {
@@ -2082,3 +2083,135 @@ async fn images_gardees_par_la_facade() {
"image encore envoyée : {sent}"
);
}
/// Réponse non streamée d'un préchauffage (un seul jeton, jeté).
fn prewarm_reply() -> String {
serde_json::json!({
"choices": [{"message": {"content": "x"}, "finish_reason": "length"}],
"usage": {"prompt_tokens": 60, "completion_tokens": 1},
"timings": {"prompt_n": 18, "cache_n": 42}
})
.to_string()
}
/// Lot 33 : PREWARM tenu par la façade. Fin de tour → la requête du prochain
/// tour, sentinelle « . » et un seul jeton, inscrite au Go et rendue avec ce
/// que le moteur a repris. Le tour suivant, qui la prolonge, ne l'annule pas.
#[tokio::test]
async fn prechauffage_par_la_facade() {
let gate = Arc::new(tokio::sync::Semaphore::new(1));
let (url, seen) = fake_engine_gated(
vec![answer("vu"), prewarm_reply(), answer("ok"), prewarm_reply()],
Some(gate.clone()),
)
.await;
let (svc, go, _stop, _t) = service(Some(&url)).await;
go.set("config", "PREWARM", "on");
svc.send(serde_json::json!({"message": "bonjour", "cid": "p1"}))
.await
.unwrap();
until_done(&svc, 1).await;
let id = svc.sessions.conv().id();
// Le préchauffage attend le moteur : il est en vol.
until(|| seen.lock().unwrap().len() == 2).await;
assert_eq!(go.prewarm_log.lock().unwrap()[..1], ["check".to_string()]);
let pre = seen.lock().unwrap()[1].1.clone();
assert_eq!(pre["stream"], false);
assert_eq!(pre["max_tokens"], 1);
let msgs = pre["messages"].as_array().unwrap().clone();
assert_eq!(msgs.last().unwrap()["content"], ".");
assert_eq!(msgs.last().unwrap()["role"], "user");
let first = seen.lock().unwrap()[0].1.clone();
assert_eq!(msgs[..2], first["messages"].as_array().unwrap()[..2]);
assert_eq!(msgs[2]["content"], "vu");
assert_eq!(pre["tools"], first["tools"]);
let begin = go.prewarm_log.lock().unwrap()[1].clone();
assert!(
begin.starts_with("begin ")
&& begin.contains(&format!("\"conv\":\"{id}\""))
&& begin.contains("\"n\":3"),
"{begin}"
);
// Pas de ticket « en vol » de la façade : le Go l'a inscrit lui-même.
let begins = || {
go.oai_log
.lock()
.unwrap()
.iter()
.filter(|l| l.starts_with("begin"))
.cloned()
.collect::<Vec<_>>()
};
assert_eq!(begins().len(), 1);
// Le tour suivant prolonge le préfixe préchauffé : gardé.
svc.send(serde_json::json!({"message": "suite", "cid": "p2"}))
.await
.unwrap();
until(|| begins().len() == 2).await;
assert_eq!(begins()[1], format!("begin main_conv={id} prewarm_keep"));
gate.add_permits(10);
until_done(&svc, 2).await;
until(|| {
go.prewarm_log
.lock()
.unwrap()
.iter()
.filter(|l| l.starts_with("end"))
.count()
== 2
})
.await;
let log = go.prewarm_log.lock().unwrap().clone();
let first_end = log.iter().find(|l| l.starts_with("end")).unwrap();
let v: serde_json::Value = serde_json::from_str(first_end.trim_start_matches("end ")).unwrap();
assert_eq!(v, serde_json::json!({"id": 5, "total": 60, "cached": 42}));
let perfs = go.perf_records.lock().unwrap().clone();
assert!(
perfs
.iter()
.any(|p| p["kind"] == "prewarm" && p["stats"]["prompt_tokens_total"] == 60)
);
}
/// Le Go refuse (moteur occupé) : rien ne part ; annulé par une autre
/// requête : la connexion est coupée et la place rendue.
#[tokio::test]
async fn prechauffage_refuse_ou_annule() {
let (url, seen) = fake_engine(vec![answer("vu")]).await;
let (svc, go, _stop, _t) = service(Some(&url)).await;
go.set("config", "PREWARM", "on");
*go.prewarm_refuse.lock().unwrap() = Some("moteur occupé".into());
svc.send(serde_json::json!({"message": "bonjour", "cid": "r1"}))
.await
.unwrap();
until_done(&svc, 1).await;
until(|| !go.prewarm_log.lock().unwrap().is_empty()).await;
tokio::time::sleep(Duration::from_millis(100)).await;
assert_eq!(*go.prewarm_log.lock().unwrap(), ["check"]);
assert_eq!(seen.lock().unwrap().len(), 1);
// Annulation relayée par le Go pendant que le moteur calcule.
let gate = Arc::new(tokio::sync::Semaphore::new(0));
let (url, seen) = fake_engine_gated(vec![prewarm_reply()], Some(gate.clone())).await;
go.set("config", "PORT", &port_of(&url));
*go.prewarm_refuse.lock().unwrap() = None;
go.prewarm_log.lock().unwrap().clear();
svc.prewarm_kick(loki_rs::chat::prewarm::AFTER_TASK).await;
until(|| seen.lock().unwrap().len() == 1).await;
assert!(svc.shared.prewarm.live());
assert!(svc.shared.prewarm.cancel_id(5));
until(|| !svc.shared.prewarm.live()).await;
let log = go.prewarm_log.lock().unwrap().clone();
assert_eq!(
log.last().unwrap(),
r#"end {"cached":null,"id":5,"total":0}"#
);
assert!(
go.perf_records
.lock()
.unwrap()
.iter()
.all(|p| p["kind"] != "prewarm")
);
}
+6 -1
View File
@@ -63,7 +63,12 @@ struct ParityHost {
}
impl TurnHost for ParityHost {
async fn send(&self, _ep: &ChatEndpoint, body: Vec<u8>) -> Result<Reply, SendError> {
async fn send(
&self,
_ep: &ChatEndpoint,
body: Vec<u8>,
_keep: bool,
) -> Result<Reply, SendError> {
self.requests
.lock()
.unwrap()