mirror of
https://github.com/R0m1k3/Loki.git
synced 2026-10-22 14:38:12 +02:00
Lot 31 : slots du moteur pour le tour tenu par la façade
SIDE_SLOT et SLOT_PERSIST ne gardent plus la conversation au Go. La machinerie reste au Go (configuration, requêtes en vol, tampon du slot) ; la façade la consulte avant chaque requête au moteur local. Go : - /internal/v1/turn/slot (facade_slots.go) rend engineSlotFor et, pour une étape de tour, recharge d'abord l'état gardé de la discussion (slotPersistRestore). La sauvegarde à la bascule voit les étapes de la façade : /internal/v1/oai/served marque le slot (engineMarkMain). - facadeChatBlocker : SIDE_SLOT et SLOT_PERSIST retirés. Rust : - TurnHost::engine_slot : l'id_slot posé dans chaque requête (0 pour le fil de la discussion, 1 pour la vérification), le ticket « en vol » ouvert sur le second slot. - Cache KV plein en plein calcul (sideSlotPoolError) : rejoué deux fois tel quel, sans compaction, puis l'erreur. - Résumés de compaction comptés en vol comme postSummary : transcription sur le second slot, continuation sur le slot 0. - La passe de vérification est un travail annexe (engineSideJob) : le Go efface le slot à la fin s'il a servi (écart du lot 7 levé). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv
This commit is contained in:
14 files changed
+679
-58
No files matched your search
@@ -26,7 +26,9 @@
|
||||
//! (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/turn/perf`) ; les motifs `retry_patterns` personnalisés
|
||||
//! sont lus au démarrage ([`retry_patterns`]).
|
||||
//! sont lus au démarrage ([`retry_patterns`]). Chaque requête au moteur local
|
||||
//! part sur le slot que rend le Go (SIDE_SLOT, rechargement SLOT_PERSIST,
|
||||
//! lot 31).
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::future::Future;
|
||||
@@ -510,7 +512,7 @@ impl ChatService {
|
||||
cfg: cfg.clone(),
|
||||
patterns: self.shared.patterns.clone(),
|
||||
port,
|
||||
perf: Some((store.clone(), self.sessions.conv().id())),
|
||||
go: Some((store.clone(), self.sessions.conv().id())),
|
||||
};
|
||||
TurnEnv {
|
||||
cfg,
|
||||
@@ -820,6 +822,7 @@ impl ChatService {
|
||||
},
|
||||
queued: Box::new(move || me.drain_queued(epoch)),
|
||||
perf_kind: super::perf::KIND_MAIN,
|
||||
side_slot: Default::default(),
|
||||
};
|
||||
|
||||
// Chaque événement devient un delta du journal.
|
||||
|
||||
+103
-31
@@ -11,7 +11,11 @@
|
||||
//! ([`EngineSummarizer`]), blocs archivés pour recall par le Go
|
||||
//! ([`StoreRecall`]) ;
|
||||
//! - images jointes (`loki-img:…`) lues sur disque et ouvertes par le Go
|
||||
//! (déchiffrées), gardées en cache comme `chatImageDataURL`.
|
||||
//! (déchiffrées), gardées en cache comme `chatImageDataURL` ;
|
||||
//! - slots du moteur (lot 31) : avant chaque requête au moteur local, le Go
|
||||
//! rend le slot où l'envoyer (SIDE_SLOT) et recharge au besoin l'état
|
||||
//! gardé de la discussion (SLOT_PERSIST) ; la requête est comptée en vol,
|
||||
//! sur ce slot, jusqu'à la fin de sa lecture.
|
||||
//!
|
||||
//! Écarts assumés : le résumé en continuation (COMPACT_CONTINUATION) demande
|
||||
//! l'état du slot du moteur, tenu par le Go — `slot_holds` rend faux et la
|
||||
@@ -20,7 +24,9 @@
|
||||
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_util::StreamExt;
|
||||
use serde_json::{Map, Value, json};
|
||||
@@ -170,13 +176,28 @@ pub struct EngineSummarizer {
|
||||
pub cfg: HashMap<String, String>,
|
||||
pub patterns: Arc<RetryPatterns>,
|
||||
pub port: u16,
|
||||
/// Télémétrie des résumés (kind=compact), rattachée à cette discussion :
|
||||
/// `None` sans base (tests).
|
||||
pub perf: Option<(Store, String)>,
|
||||
/// Le Go et la discussion des résumés : télémétrie (kind=compact),
|
||||
/// requêtes en vol et slot du moteur. `None` sans base (tests).
|
||||
pub go: Option<(Store, String)>,
|
||||
}
|
||||
|
||||
impl EngineSummarizer {
|
||||
async fn post(&self, payload: Value) -> Result<SummaryChoice, SummaryError> {
|
||||
/// `postSummary`. `main` : le résumé prolonge le prompt du slot de la
|
||||
/// discussion (continuation) ; sinon c'est un prompt neuf, sur le second
|
||||
/// slot quand il est en service (SIDE_SLOT).
|
||||
async fn post(&self, mut payload: Value, main: bool) -> Result<SummaryChoice, SummaryError> {
|
||||
let mut _gate = None;
|
||||
if !self.ep.external
|
||||
&& let Some((store, _)) = &self.go
|
||||
{
|
||||
let slot = engine_slot_for(store, "", main).await;
|
||||
if let Some(n) = slot
|
||||
&& let Some(o) = payload.as_object_mut()
|
||||
{
|
||||
o.insert("id_slot".into(), Value::from(n));
|
||||
}
|
||||
_gate = engine_begin(store, "", slot == Some(1)).await;
|
||||
}
|
||||
let body = serde_json::to_vec(&payload).unwrap_or_default();
|
||||
let resp = self
|
||||
.http
|
||||
@@ -211,7 +232,7 @@ impl EngineSummarizer {
|
||||
Err(SummaryError::Other(e)) => e == super::compact::ERR_EMPTY_SUMMARY,
|
||||
Err(_) => false,
|
||||
};
|
||||
if decoded && let Some((store, conv)) = &self.perf {
|
||||
if decoded && let Some((store, conv)) = &self.go {
|
||||
let (complete, stats) = super::perf::wire_stats(&raw);
|
||||
let rec = PerfRec {
|
||||
complete,
|
||||
@@ -247,7 +268,7 @@ impl Summarizer for EngineSummarizer {
|
||||
{
|
||||
o.remove("chat_template_kwargs");
|
||||
}
|
||||
self.post(p).await
|
||||
self.post(p, false).await
|
||||
}
|
||||
|
||||
async fn summarize_continuation(
|
||||
@@ -274,7 +295,7 @@ impl Summarizer for EngineSummarizer {
|
||||
o.remove("stream_options");
|
||||
o.insert("max_tokens".into(), json!(req.max_tokens));
|
||||
}
|
||||
self.post(p).await
|
||||
self.post(p, true).await
|
||||
}
|
||||
|
||||
fn slot_holds(&self, _conv: &str) -> bool {
|
||||
@@ -340,11 +361,68 @@ pub struct GoTurnHost {
|
||||
/// 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,
|
||||
/// La requête qui part va sur le second slot (posé par `engine_slot`,
|
||||
/// lu par `send` pour le ticket « en vol »).
|
||||
pub side_slot: AtomicBool,
|
||||
}
|
||||
|
||||
/// Délai d'une demande de slot : le rechargement SLOT_PERSIST relit jusqu'à
|
||||
/// 8 Gio (`slotPersistRestoreTimeout`, deux minutes).
|
||||
const SLOT_ASK_TIMEOUT: Duration = Duration::from_secs(180);
|
||||
|
||||
/// `engineSlotFor` demandé au Go (`/internal/v1/turn/slot`). `main` : fil de
|
||||
/// la discussion ; `conv` (étape d'un tour seulement) : le Go recharge
|
||||
/// d'abord l'état SLOT_PERSIST de cette discussion. Go injoignable : aucun
|
||||
/// `id_slot`, la requête d'avant la clé.
|
||||
pub async fn engine_slot_for(store: &Store, conv: &str, main: bool) -> Option<u8> {
|
||||
match store
|
||||
.internal_post_within(
|
||||
"/internal/v1/turn/slot",
|
||||
&json!({ "conv": conv, "main": main }),
|
||||
SLOT_ASK_TIMEOUT,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok((code, v)) if code.is_success() => v["slot"].as_i64().and_then(|n| u8::try_from(n).ok()),
|
||||
Ok((code, v)) => {
|
||||
eprintln!("[loki-rs] moteur : slot non demandé ({code} {v})");
|
||||
None
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[loki-rs] moteur : slot non demandé ({e})");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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> {
|
||||
match store
|
||||
.internal_json(
|
||||
hyper::Method::POST,
|
||||
"/internal/v1/oai/begin",
|
||||
Some(&json!({ "main_conv": main_conv, "side": side })),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(v) => v["ticket"].as_u64().map(|ticket| GateTicket {
|
||||
store: store.clone(),
|
||||
ticket,
|
||||
}),
|
||||
Err(e) => {
|
||||
eprintln!("[loki-rs] moteur : requête non comptée en vol ({e})");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Rend le ticket « en vol » au Go quand la réponse du moteur est lue ou
|
||||
/// abandonnée (`engineRequestBeginSide` … `end`).
|
||||
struct GateTicket {
|
||||
pub struct GateTicket {
|
||||
store: Store,
|
||||
ticket: u64,
|
||||
}
|
||||
@@ -367,34 +445,14 @@ impl Drop for GateTicket {
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
let side = self.side_slot.load(Ordering::SeqCst);
|
||||
engine_begin(&self.shared.store, main_conv, side).await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -646,6 +704,20 @@ impl TurnHost for GoTurnHost {
|
||||
self.shared.echo_refusals.refuse(model);
|
||||
}
|
||||
|
||||
async fn engine_slot(&self, ep: &ChatEndpoint) -> Option<u8> {
|
||||
if ep.external {
|
||||
self.side_slot.store(false, Ordering::SeqCst);
|
||||
return None;
|
||||
}
|
||||
// Étape d'un tour de la discussion : le slot 0, après rechargement de
|
||||
// l'état gardé (SLOT_PERSIST). Vérification : le second slot.
|
||||
let main = self.perf_kind == super::perf::KIND_MAIN;
|
||||
let conv = if main { self.conv_id.as_str() } else { "" };
|
||||
let slot = engine_slot_for(&self.shared.store, conv, main).await;
|
||||
self.side_slot.store(slot == Some(1), Ordering::SeqCst);
|
||||
slot
|
||||
}
|
||||
|
||||
fn cache_hint(&self) -> Option<CacheHintFn> {
|
||||
None
|
||||
}
|
||||
|
||||
@@ -27,12 +27,16 @@
|
||||
//! [`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.
|
||||
//!
|
||||
//! Slots du moteur (lot 31) : chaque requête part sur le slot que rend
|
||||
//! [`TurnHost::engine_slot`] (SIDE_SLOT : 0 pour le fil de la discussion, 1
|
||||
//! pour une vérification ; aucun sans second slot en service), et un cache KV
|
||||
//! à court de place en plein calcul est rejoué tel quel. Le rechargement
|
||||
//! SLOT_PERSIST et le comptage des requêtes en vol passent par le Go (l'hôte,
|
||||
//! autour de [`TurnHost::engine_slot`] et de [`TurnHost::send`]).
|
||||
//!
|
||||
//! 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. 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`]).
|
||||
//! KEEP_TURN_IMAGES, PROJ_SNAPSHOT, préchauffage. Avec l'une de ces clés, le
|
||||
//! Go garde la conversation (`facadeChatBlocker`, décidé au démarrage).
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::future::Future;
|
||||
@@ -189,6 +193,12 @@ pub trait TurnHost: Send + Sync {
|
||||
fn preflight(&self) -> Option<PreflightFn> {
|
||||
None
|
||||
}
|
||||
/// Slot du moteur où envoyer la requête qui part (`engineSlotFor`) ;
|
||||
/// `None` : aucun `id_slot`. Appelé avant chaque requête, moteur local ou
|
||||
/// non : l'hôte y recharge aussi l'état SLOT_PERSIST de la discussion.
|
||||
fn engine_slot(&self, _ep: &ChatEndpoint) -> impl Future<Output = Option<u8>> + Send {
|
||||
async { 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).
|
||||
@@ -324,6 +334,20 @@ fn log_stream_retry(attempt: u32, cause: &str) {
|
||||
);
|
||||
}
|
||||
|
||||
/// `sideSlotPoolError` : « Context size has been exceeded. » sans le nombre
|
||||
/// de jetons de la requête n'est pas un prompt trop long, c'est le cache KV
|
||||
/// qui a manqué de place pendant le calcul — avec deux slots, la faute du
|
||||
/// voisin, pas de la conversation.
|
||||
pub fn side_slot_pool_error(msg: &str) -> bool {
|
||||
static REQ_EXCEEDS: std::sync::LazyLock<regex::Regex> = std::sync::LazyLock::new(|| {
|
||||
regex::Regex::new(r"(?i)exceeds the available context size|request \([0-9]+ tokens\)")
|
||||
.expect("motif fixe")
|
||||
});
|
||||
msg.to_lowercase()
|
||||
.contains("context size has been exceeded")
|
||||
&& !REQ_EXCEEDS.is_match(msg)
|
||||
}
|
||||
|
||||
fn text_msg(role: &str, text: impl Into<String>) -> Message {
|
||||
Message {
|
||||
role: role.into(),
|
||||
@@ -400,6 +424,8 @@ struct Loop {
|
||||
budget_nudges: i64,
|
||||
/// `perfIter` : complétions de ce tour, complètes ou non.
|
||||
perf_iter: i64,
|
||||
/// Rejeux « cache KV plein » (SIDE_SLOT).
|
||||
pool_retries: u32,
|
||||
}
|
||||
|
||||
/// `runChatLoop` : le tour, de la première requête à la réponse finale.
|
||||
@@ -467,9 +493,10 @@ 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 slot = host.engine_slot(ep).await;
|
||||
let (body, echo_sent, prefix) = {
|
||||
let (sent, echo_sent) = wire_messages(&messages, s.tool_choice_none, &pol);
|
||||
let payload = build_chat_payload(
|
||||
let mut payload = build_chat_payload(
|
||||
ep,
|
||||
&sent,
|
||||
tc.temperature,
|
||||
@@ -484,6 +511,12 @@ pub async fn run_turn<H: TurnHost>(
|
||||
&tc.cfg,
|
||||
&|r| host.image_data_url(r),
|
||||
);
|
||||
// SIDE_SLOT : le slot rendu par l'hôte ; aucun, la requête d'avant.
|
||||
if let Some(n) = slot
|
||||
&& let Some(o) = payload.as_object_mut()
|
||||
{
|
||||
o.insert("id_slot".into(), Value::from(n));
|
||||
}
|
||||
(
|
||||
serde_json::to_vec(&payload).unwrap_or_default(),
|
||||
echo_sent,
|
||||
@@ -532,6 +565,26 @@ pub async fn run_turn<H: TurnHost>(
|
||||
effort = fixed;
|
||||
continue;
|
||||
}
|
||||
// SIDE_SLOT : un cache KV à court de place en plein calcul n'est
|
||||
// pas un prompt trop long. Rejouée telle quelle, sans compaction ni
|
||||
// réduction — deux fois au plus, le temps que l'autre slot finisse,
|
||||
// puis l'erreur telle quelle. Sans second slot en service, rien.
|
||||
if !ep.external && slot.is_some() && side_slot_pool_error(&msg) {
|
||||
if s.pool_retries < 2 {
|
||||
s.pool_retries += 1;
|
||||
log_ctx(
|
||||
"cache KV plein pendant le calcul (deux slots) : requête rejouée telle quelle, sans compaction",
|
||||
);
|
||||
if cancel.backoff(s.pool_retries - 1).await {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return fail(
|
||||
cb,
|
||||
extra,
|
||||
format!("llama-server a renvoyé {status} : {msg}"),
|
||||
);
|
||||
}
|
||||
if echo_sent > 0 && echo_template_error(status, &msg) {
|
||||
s.echo_off_turn = true;
|
||||
s.echo_tpl_retry = true;
|
||||
|
||||
@@ -39,6 +39,9 @@ struct Fake {
|
||||
/// Télémétrie rangée ; et ce que le Go en tire.
|
||||
perfs: Mutex<Vec<PerfRec>>,
|
||||
marks: Mutex<PerfMarks>,
|
||||
/// Slot rendu par l'hôte (SIDE_SLOT) ; demandes reçues.
|
||||
slot: Mutex<Option<u8>>,
|
||||
slot_asks: Mutex<usize>,
|
||||
}
|
||||
|
||||
impl Fake {
|
||||
@@ -150,6 +153,10 @@ impl TurnHost for Fake {
|
||||
self.perfs.lock().unwrap().push(rec);
|
||||
self.marks.lock().unwrap().clone()
|
||||
}
|
||||
async fn engine_slot(&self, _ep: &ChatEndpoint) -> Option<u8> {
|
||||
*self.slot_asks.lock().unwrap() += 1;
|
||||
*self.slot.lock().unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
fn content(s: &str) -> String {
|
||||
@@ -375,6 +382,95 @@ async fn erreur_500_tool_choice_none_puis_outils_retires() {
|
||||
assert!(host.completions.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
/// SIDE_SLOT : chaque requête porte le slot que rend l'hôte, demandé avant
|
||||
/// chaque envoi ; sans slot, aucun `id_slot`.
|
||||
#[tokio::test]
|
||||
async fn slot_du_moteur_dans_chaque_requete() {
|
||||
let host = Fake::new(vec![
|
||||
tool_step("c1", "glob", r#"{"pattern":"*.go"}"#),
|
||||
answer("Trouvé"),
|
||||
]);
|
||||
*host.slot.lock().unwrap() = Some(1);
|
||||
let (r, _) = run(&host, &config(&["glob"]), &Cancel::never()).await;
|
||||
assert_eq!(r.err, None);
|
||||
assert_eq!(
|
||||
(
|
||||
host.body(0)["id_slot"].clone(),
|
||||
host.body(1)["id_slot"].clone()
|
||||
),
|
||||
(json!(1), json!(1))
|
||||
);
|
||||
assert_eq!(*host.slot_asks.lock().unwrap(), 2);
|
||||
let host = Fake::new(vec![answer("ok")]);
|
||||
let (r, _) = run(&host, &config(&[]), &Cancel::never()).await;
|
||||
assert_eq!(r.err, None);
|
||||
assert!(host.body(0).get("id_slot").is_none());
|
||||
}
|
||||
|
||||
/// Cache KV plein en plein calcul avec deux slots : rejouée telle quelle,
|
||||
/// deux fois au plus, sans compaction ; au-delà, l'erreur. Sans second slot,
|
||||
/// c'est un débordement comme un autre.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn cache_kv_plein_rejoue_sans_compaction() {
|
||||
let plein = r#"{"error":{"code":500,"message":"Context size has been exceeded.","type":"server_error"}}"#;
|
||||
let host = Fake::new(vec![
|
||||
Script::Status(500, plein.into()),
|
||||
Script::Status(500, plein.into()),
|
||||
answer("ok"),
|
||||
]);
|
||||
*host.slot.lock().unwrap() = Some(0);
|
||||
*host.compacted.lock().unwrap() = Some(vec![um("résumé")]);
|
||||
let (r, evs) = run(&host, &config(&[]), &Cancel::never()).await;
|
||||
assert_eq!(r.err, None);
|
||||
assert_eq!(text(&evs), "ok");
|
||||
assert_eq!(host.sent(), 3);
|
||||
assert!(host.phases.lock().unwrap().is_empty(), "pas de compaction");
|
||||
assert_eq!(host.body(2)["messages"], host.body(0)["messages"]);
|
||||
// Trois fois de suite : l'erreur telle quelle, sans compaction.
|
||||
let host = Fake::new((0..3).map(|_| Script::Status(500, plein.into())).collect());
|
||||
*host.slot.lock().unwrap() = Some(0);
|
||||
*host.compacted.lock().unwrap() = Some(vec![um("résumé")]);
|
||||
let (r, _) = run(&host, &config(&[]), &Cancel::never()).await;
|
||||
// Le corps du refus tel quel, comme le Go.
|
||||
assert_eq!(r.err, Some(format!("llama-server a renvoyé 500 : {plein}")));
|
||||
assert_eq!(host.sent(), 3);
|
||||
assert!(host.phases.lock().unwrap().is_empty());
|
||||
// Sans second slot : compaction réactive.
|
||||
let host = Fake::new(vec![Script::Status(500, plein.into()), answer("ok")]);
|
||||
*host.compacted.lock().unwrap() = Some(vec![um("résumé")]);
|
||||
let (r, _) = run(&host, &config(&[]), &Cancel::never()).await;
|
||||
assert_eq!(r.err, None);
|
||||
assert_eq!(*host.phases.lock().unwrap(), [CompactPhase::Reactive]);
|
||||
// Moteur externe : jamais (le slot vient d'un moteur local).
|
||||
let host = Fake::new(vec![Script::Status(500, plein.into()), answer("ok")]);
|
||||
*host.slot.lock().unwrap() = Some(0);
|
||||
*host.compacted.lock().unwrap() = Some(vec![um("résumé")]);
|
||||
let mut tc = config(&[]);
|
||||
tc.ep.external = true;
|
||||
let (r, _) = run(&host, &tc, &Cancel::never()).await;
|
||||
assert_eq!(r.err, None);
|
||||
assert_eq!(*host.phases.lock().unwrap(), [CompactPhase::Reactive]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cache_kv_plein_ou_prompt_trop_long() {
|
||||
for (msg, want) in [
|
||||
(
|
||||
r#"{"error":{"code":500,"message":"Context size has been exceeded.","type":"server_error"}}"#,
|
||||
true,
|
||||
),
|
||||
(
|
||||
r#"{"error":{"message":"request (40000 tokens) exceeds the available context size (32768 tokens), try increasing it","type":"exceed_context_size_error"}}"#,
|
||||
false,
|
||||
),
|
||||
("the request exceeds the available context size", false),
|
||||
("CONTEXT SIZE HAS BEEN EXCEEDED, Request (12 TOKENS)", false),
|
||||
("failed to parse tool call", false),
|
||||
] {
|
||||
assert_eq!(side_slot_pool_error(msg), want, "{msg}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn debordement_compaction_reactive() {
|
||||
let host = Fake::new(vec![
|
||||
|
||||
@@ -12,9 +12,10 @@
|
||||
//! en base (`chat/crit:<id>`, chiffrés comme le reste) ; le contexte isolé du
|
||||
//! vérificateur vient de `/internal/v1/turn/verifier`.
|
||||
//!
|
||||
//! Écart assumé : l'isolation du slot du moteur après la passe de
|
||||
//! vérification (`engineSideJob`, effacement du cache) reste au Go, qui ne
|
||||
//! voit pas passer les requêtes de la façade.
|
||||
//! La passe de vérification est un travail annexe (`engineSideJob`, lot 31) :
|
||||
//! le Go voit passer ses requêtes (tickets « en vol », moteur qui les accepte)
|
||||
//! et, à la fin, efface le slot local s'il a servi — la discussion se
|
||||
//! recharge alors depuis le cache RAM du moteur au lieu de se recalculer.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -54,6 +55,29 @@ pub struct Criterion {
|
||||
pub note: String,
|
||||
}
|
||||
|
||||
/// `engineSideJob` : début d'un travail annexe auprès du Go ; `None` s'il ne
|
||||
/// répond pas.
|
||||
async fn side_job_begin(store: &Store) -> Option<i64> {
|
||||
store
|
||||
.internal_post("/internal/v1/engine/sidejob", &json!({}))
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|(_, v)| v.get("id").and_then(Value::as_i64))
|
||||
}
|
||||
|
||||
/// Fin du travail annexe. Les requêtes servies ont déjà été comptées une à
|
||||
/// une (`/internal/v1/oai/served`) : rien à ajouter.
|
||||
async fn side_job_end(store: &Store, id: Option<i64>) {
|
||||
if let Some(id) = id {
|
||||
let _ = store
|
||||
.internal_post(
|
||||
"/internal/v1/engine/sidejob",
|
||||
&json!({ "end": id, "served": false }),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// `critList` : les critères de la discussion (vide si absents ou illisibles,
|
||||
/// mémoire verrouillée comprise).
|
||||
pub async fn crit_list(store: &Store, conv_id: &str) -> Vec<Criterion> {
|
||||
@@ -277,8 +301,12 @@ impl ChatService {
|
||||
..Default::default()
|
||||
},
|
||||
];
|
||||
// Travail annexe : le Go efface le slot à la fin s'il a servi
|
||||
// (les requêtes de la passe le lui disent en passant).
|
||||
let side = side_job_begin(&self.shared.store).await;
|
||||
self.run_pass(pe, epoch, msgs, tools, v["caps"].clone(), true, &[], run)
|
||||
.await;
|
||||
side_job_end(&self.shared.store, side).await;
|
||||
}
|
||||
other => {
|
||||
let e = match other {
|
||||
@@ -434,6 +462,7 @@ impl ChatService {
|
||||
} else {
|
||||
super::perf::KIND_MAIN
|
||||
},
|
||||
side_slot: Default::default(),
|
||||
};
|
||||
let mut out = PassOut::default();
|
||||
let mut cb = |ev: StreamEvent| -> bool {
|
||||
|
||||
@@ -127,6 +127,11 @@ pub struct FakeGo {
|
||||
pub service_os_fail: Mutex<Option<(String, String)>>,
|
||||
/// /internal/v1/turn/perf reçus (corps entiers).
|
||||
pub perf_records: Mutex<Vec<serde_json::Value>>,
|
||||
/// /internal/v1/turn/slot : second slot en service (le fil de la
|
||||
/// discussion sur le 0, le reste sur le 1 ; sinon aucun) ; demandes
|
||||
/// reçues (« conv main »).
|
||||
pub turn_side_live: Mutex<bool>,
|
||||
pub turn_slots: Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
impl FakeGo {
|
||||
@@ -666,10 +671,11 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
|
||||
let body = req.into_body().collect().await.unwrap().to_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 { "" };
|
||||
st.oai_log
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(format!("begin main_conv={conv}"));
|
||||
.push(format!("begin main_conv={conv}{side}"));
|
||||
} else {
|
||||
st.oai_log.lock().unwrap().push(format!(
|
||||
"begin read={} side={} post_v1={}",
|
||||
@@ -681,6 +687,21 @@ async fn handle(st: Arc<FakeGo>, req: Request<Incoming>) -> Response<Full<Bytes>
|
||||
}
|
||||
resp(200, r#"{"ticket":7}"#)
|
||||
}
|
||||
"/internal/v1/turn/slot" => {
|
||||
let body = req.into_body().collect().await.unwrap().to_bytes();
|
||||
let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
|
||||
let main = v["main"] == true;
|
||||
st.turn_slots
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(format!("{} {main}", v["conv"].as_str().unwrap_or("")));
|
||||
let slot = match (*st.turn_side_live.lock().unwrap(), main) {
|
||||
(false, _) => -1,
|
||||
(true, true) => 0,
|
||||
(true, false) => 1,
|
||||
};
|
||||
resp(200, serde_json::json!({ "slot": slot }).to_string())
|
||||
}
|
||||
"/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();
|
||||
|
||||
@@ -680,7 +680,7 @@ async fn tour_complet_par_le_go() {
|
||||
cfg: cfg.clone(),
|
||||
patterns: shared.patterns.clone(),
|
||||
port: 8080,
|
||||
perf: None,
|
||||
go: None,
|
||||
},
|
||||
compact: TurnCompact {
|
||||
cfg: CompactConfig::default(),
|
||||
@@ -693,6 +693,7 @@ async fn tour_complet_par_le_go() {
|
||||
},
|
||||
queued: Box::new(Vec::new),
|
||||
perf_kind: "main",
|
||||
side_slot: Default::default(),
|
||||
};
|
||||
let bash = Tool {
|
||||
kind: "function".into(),
|
||||
@@ -1056,7 +1057,7 @@ async fn hote_et_sonde_de_gabarit() {
|
||||
cfg,
|
||||
patterns: shared.patterns.clone(),
|
||||
port: 1,
|
||||
perf: None,
|
||||
go: None,
|
||||
},
|
||||
compact: TurnCompact {
|
||||
cfg: CompactConfig::default(),
|
||||
@@ -1069,6 +1070,7 @@ async fn hote_et_sonde_de_gabarit() {
|
||||
},
|
||||
queued: Box::new(Vec::new),
|
||||
perf_kind: "main",
|
||||
side_slot: Default::default(),
|
||||
}
|
||||
};
|
||||
let bash = Tool {
|
||||
@@ -1179,6 +1181,7 @@ async fn contrat_verifie_par_la_facade() {
|
||||
.await;
|
||||
let (svc, go, _stop, _t) = service(Some(&url)).await;
|
||||
*go.code.lock().unwrap() = true;
|
||||
*go.turn_side_live.lock().unwrap() = true;
|
||||
let id = svc.sessions.conv().id();
|
||||
go.set(
|
||||
"chat",
|
||||
@@ -1193,6 +1196,39 @@ async fn contrat_verifie_par_la_facade() {
|
||||
|
||||
let seen = seen.lock().unwrap().clone();
|
||||
assert_eq!(seen.len(), 3);
|
||||
// SIDE_SLOT (lot 31) : le tour sur le slot 0 (rechargement SLOT_PERSIST
|
||||
// demandé pour la discussion), la vérification sur le 1, comptée en vol
|
||||
// sur le second slot ; la passe est un travail annexe pour le Go.
|
||||
let slots: Vec<_> = seen.iter().map(|(_, b)| b["id_slot"].clone()).collect();
|
||||
assert_eq!(
|
||||
slots,
|
||||
[
|
||||
serde_json::json!(0),
|
||||
serde_json::json!(1),
|
||||
serde_json::json!(1)
|
||||
]
|
||||
);
|
||||
assert_eq!(
|
||||
*go.turn_slots.lock().unwrap(),
|
||||
[format!("{id} true"), " false".into(), " false".into()]
|
||||
);
|
||||
let begins: Vec<String> = go
|
||||
.oai_log
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|l| l.starts_with("begin"))
|
||||
.cloned()
|
||||
.collect();
|
||||
assert_eq!(
|
||||
begins,
|
||||
[
|
||||
format!("begin main_conv={id}"),
|
||||
"begin main_conv= side".into(),
|
||||
"begin main_conv= side".into()
|
||||
]
|
||||
);
|
||||
assert_eq!(*go.side_jobs.lock().unwrap(), ["begin 1", "end 1 false"]);
|
||||
let v = &seen[1].1;
|
||||
assert_eq!(v["messages"][0]["content"], "VÉRIFICATEUR");
|
||||
assert_eq!(
|
||||
@@ -1855,7 +1891,7 @@ async fn parite_du_tour_complet() {
|
||||
async fn telemetrie_des_resumes() {
|
||||
use loki_rs::chat::compact::{Summarizer, TranscriptRequest};
|
||||
let (st, go, _stop, _t) = store().await;
|
||||
let (url, _seen) = fake_engine(vec![
|
||||
let (url, seen) = fake_engine(vec![
|
||||
serde_json::json!({
|
||||
"choices": [{"message": {"content": "résumé"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 3000, "completion_tokens": 200},
|
||||
@@ -1864,6 +1900,10 @@ async fn telemetrie_des_resumes() {
|
||||
.to_string(),
|
||||
r#"{"choices":[]}"#.into(),
|
||||
"pas du json".into(),
|
||||
serde_json::json!({
|
||||
"choices": [{"message": {"content": "suite"}, "finish_reason": "stop"}]
|
||||
})
|
||||
.to_string(),
|
||||
])
|
||||
.await;
|
||||
let s = EngineSummarizer {
|
||||
@@ -1877,8 +1917,9 @@ async fn telemetrie_des_resumes() {
|
||||
cfg: Default::default(),
|
||||
patterns: RetryPatterns::builtin(),
|
||||
port: 8080,
|
||||
perf: Some((st, "c9".into())),
|
||||
go: Some((st, "c9".into())),
|
||||
};
|
||||
*go.turn_side_live.lock().unwrap() = true;
|
||||
let req = || TranscriptRequest {
|
||||
system: "s".into(),
|
||||
transcript: "t".into(),
|
||||
@@ -1890,8 +1931,58 @@ async fn telemetrie_des_resumes() {
|
||||
);
|
||||
assert!(s.summarize_transcript(req()).await.is_err(), "réponse vide");
|
||||
assert!(s.summarize_transcript(req()).await.is_err(), "illisible");
|
||||
// Lot 31 : la transcription, prompt neuf, sur le second slot, comptée en
|
||||
// vol sur lui et rendue à la fin ; jamais de rechargement SLOT_PERSIST.
|
||||
let sent = seen.lock().unwrap().clone();
|
||||
assert!(sent.iter().all(|(_, b)| b["id_slot"] == 1), "{sent:?}");
|
||||
assert_eq!(
|
||||
*go.turn_slots.lock().unwrap(),
|
||||
[" false", " false", " false"]
|
||||
);
|
||||
until(|| {
|
||||
go.oai_log
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|l| l.starts_with("end"))
|
||||
.count()
|
||||
== 3
|
||||
})
|
||||
.await;
|
||||
assert_eq!(
|
||||
go.oai_log
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|l| *l == "begin main_conv= side")
|
||||
.count(),
|
||||
3
|
||||
);
|
||||
// En continuation, il prolonge le prompt de la discussion : le slot 0.
|
||||
use loki_rs::chat::compact::ContinuationRequest;
|
||||
let cont = s
|
||||
.summarize_continuation(ContinuationRequest {
|
||||
messages: vec![],
|
||||
tools: vec![],
|
||||
effort: String::new(),
|
||||
kwargs: Default::default(),
|
||||
max_tokens: 500,
|
||||
conv: "c9".into(),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(cont.content, "suite");
|
||||
assert_eq!(seen.lock().unwrap()[3].1["id_slot"], 0);
|
||||
assert_eq!(go.turn_slots.lock().unwrap()[3], " true");
|
||||
until(|| {
|
||||
go.oai_log
|
||||
.lock()
|
||||
.unwrap()
|
||||
.contains(&"begin main_conv=".to_string())
|
||||
})
|
||||
.await;
|
||||
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.len(), 3, "la réponse illisible n'est pas rangée");
|
||||
assert_eq!(
|
||||
perfs[0],
|
||||
serde_json::json!({ "kind": "compact", "conv": "c9", "iter": 0, "complete": true,
|
||||
|
||||
Reference in new issue
Block a user