diff --git a/rust/loki-rs/src/chat/compact.rs b/rust/loki-rs/src/chat/compact.rs index 6de4f52..6c354fd 100644 --- a/rust/loki-rs/src/chat/compact.rs +++ b/rust/loki-rs/src/chat/compact.rs @@ -504,7 +504,7 @@ pub(crate) fn without_ctx_update(m: &Message) -> Option { } /// `stripContextUpdates` : l'historique sans aucun bloc . -fn strip_context_updates(msgs: Vec) -> Vec { +pub(crate) fn strip_context_updates(msgs: Vec) -> Vec { msgs.into_iter() .map(|m| without_ctx_update(&m).unwrap_or(m)) .collect() diff --git a/rust/loki-rs/src/chat/conv.rs b/rust/loki-rs/src/chat/conv.rs index 5cf5c66..ba0e166 100644 --- a/rust/loki-rs/src/chat/conv.rs +++ b/rust/loki-rs/src/chat/conv.rs @@ -157,6 +157,16 @@ impl Conversation { Ok(g.epoch) } + /// Rend le verrou d'un tour qui n'a pas pu démarrer (moteur tenu par le + /// Go), sans rien coalescer ni réveiller. + pub fn release_turn(&self, epoch: i64) { + let mut g = self.lock(); + if g.epoch == epoch { + g.generating = false; + g.cancel = None; + } + } + /// Fin d'un tour : le journal du tour est coalescé (`compactLogLocked`) /// et le verrou rendu. Un tour dont l'epoch est périmé (reset, bascule) /// ne touche à rien : le verrou a déjà été rendu, peut-être repris. @@ -196,6 +206,13 @@ impl Conversation { f(&mut self.lock().st) } + /// [`Self::with_state`], seulement si l'epoch n'a pas changé (reset ou + /// bascule pendant le tour : rien de ce tour ne doit toucher au nouveau fil). + pub fn with_state_if(&self, epoch: i64, f: impl FnOnce(&mut ConvState) -> R) -> Option { + let mut g = self.lock(); + (g.epoch == epoch).then(|| f(&mut g.st)) + } + /// Instantané à enregistrer, numéroté sous le même verrou que l'état. pub fn snapshot(&self, p: &Persister, project: &str) -> ConvSnap { let g = self.lock(); diff --git a/rust/loki-rs/src/chat/engine.rs b/rust/loki-rs/src/chat/engine.rs index 1c754f0..fec61b5 100644 --- a/rust/loki-rs/src/chat/engine.rs +++ b/rust/loki-rs/src/chat/engine.rs @@ -193,6 +193,21 @@ impl EngineHttp { } } +/// `healthCheck` : `/health` doit répondre tout de suite ou pas du tout +/// (`healthClient`, 3 s) — un moteur qui accepte la connexion sans répondre +/// (très gros modèle en chargement) ne doit pas bloquer l'envoi. +pub const HEALTH_TIMEOUT: Duration = Duration::from_secs(3); + +impl EngineHttp { + /// Le moteur répond 200 sur `url` (`/health`) dans le délai. + pub async fn health(&self, url: &str) -> bool { + match tokio::time::timeout(HEALTH_TIMEOUT, self.client.get(url).send()).await { + Ok(Ok(r)) => r.status().as_u16() == 200, + _ => false, + } + } +} + /// Fournisseur cryptographique de rustls (ring), posé une fois. fn install_crypto() { #[cfg(feature = "tls")] diff --git a/rust/loki-rs/src/chat/generate.rs b/rust/loki-rs/src/chat/generate.rs new file mode 100644 index 0000000..bd67bea --- /dev/null +++ b/rust/loki-rs/src/chat/generate.rs @@ -0,0 +1,821 @@ +//! Un tour de discussion de bout en bout — port de `EnqueueOrStart`, +//! `StartTurn`, `generate`, `drainQueued`, `startQueuedIfAny`, `dropQueued` +//! et `compactAndPublish` (`chat_conversation.go`). +//! +//! Envoi : doublon d'un réessai reconnu (`cid`), tour préparé par le Go +//! (`/internal/v1/turn/start` : pièces jointes, capacités, rôle du mode Code, +//! puce), mis en file si un tour tourne déjà, sinon lancé : moteur prêt +//! (`/health`), verrou du tour en Rust puis bail du moteur du Go (partagé +//! avec tâches, bench, optimiseur), bulle utilisateur journalisée et +//! enregistrée tout de suite. +//! +//! Le tour (`generate`) : compaction de début de tour, vue d'envoi +//! (`/internal/v1/turn/view` + historique), boucle ([`run_turn`]) dont chaque +//! événement devient un delta du journal, historique rangé (outils, réponse, +//! raisonnement renvoyé, rappels en porte-à-faux retirés), contexte estimé +//! quand le moteur ne le compte pas, compaction de fin de tour, puis fin : +//! `turn_done`, journal coalescé, verrou et bail rendus, discussion +//! enregistrée, et la file — abandonnée après un stop, sinon le message +//! suivant part (ou la notification Web Push s'il n'y en a pas). +//! +//! Écarts assumés (restent au Go, à porter au lot 7) : boucle de +//! vérification du mode Code (`codeVerifyLoop`), critères rendus pour la +//! compaction, PROJ_SNAPSHOT et KEEP_TURN_IMAGES (refusés par +//! `/turn/view`), préchauffage, télémétrie, motifs `retry_patterns` +//! personnalisés. + +use std::collections::VecDeque; +use std::future::Future; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use tokio::sync::watch; + +use super::compact::{ + CompactCaps, CompactConfig, CompactOpts, Compactor, TurnView, compact_needed, compact_opts_for, + ctx_pending, estimate_tokens, log_compact, log_ctx_estimate, strip_context_updates, +}; +use super::engine::EngineHttp; +use super::history::{ + drop_stray_nudges, history_from_view, inject_prefix, remind_read_mem_pages, tool_used_delta, +}; +use super::host::{ChatShared, EngineSummarizer, GoTurnHost, StoreRecall, TurnCompact}; +use super::journal::Delta; +use super::lease::{EngineLease, LEASE_TTL, LeaseError, LeaseSignal, engine_stop}; +use super::sessions::{GoSessionHooks, Sessions}; +use super::turn::{Cancel, TurnConfig, run_turn}; +use super::types::{Message, ReasoningEcho, StreamEvent, Tool}; +use super::wire::{echo_policy_for, resolve_chat_endpoint, turn_reasoning, with_echo}; +use crate::engstatus::llm_port; +use crate::store::BK_STATE; + +/// `errModelLoading` : une attente, pas un défaut. +pub const ERR_MODEL_LOADING: &str = + "⏳ Le modèle est encore en train de charger — réessaie dans quelques secondes."; +/// `ErrBusy`. +pub const ERR_BUSY: &str = "génération en cours"; +/// Identifiants d'envoi retenus (`recentCIDs`). +pub const RECENT_CIDS: usize = 64; + +/// Un tour préparé par le Go (`/internal/v1/turn/start`). +#[derive(Debug, Clone, Default, Deserialize)] +pub struct Prepared { + #[serde(default)] + pub files: Value, + /// Capacités demandées. + #[serde(default)] + pub caps: Value, + /// Capacités du tour (rôle du mode Code figé). + #[serde(default)] + pub turn_caps: Value, + /// Le message utilisateur vu par le modèle. + #[serde(default)] + pub content: Value, + #[serde(default)] + pub model: String, + #[serde(default)] + pub code_hint: bool, + /// Texte tapé (bulle du journal). + #[serde(skip)] + pub text: String, + #[serde(skip)] + pub temperature: f64, +} + +impl Prepared { + fn cap(&self, k: &str) -> bool { + self.turn_caps + .get(k) + .and_then(Value::as_bool) + .unwrap_or(false) + } + + /// La bulle utilisateur du journal : texte, modèle qui répond, fichiers. + fn user_delta(&self) -> Delta { + let mut d = Map::new(); + d.insert("user".into(), self.text.clone().into()); + if !self.model.is_empty() { + d.insert("model".into(), self.model.clone().into()); + } + if self.files.as_array().is_some_and(|f| !f.is_empty()) { + d.insert("files".into(), self.files.clone()); + } + d + } + + fn user_message(&self) -> Message { + Message { + role: "user".into(), + content: Some(self.content.clone()), + ..Default::default() + } + } +} + +/// Envoi refusé : statut HTTP de `/api/chat/send` et message. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Refused { + pub status: u16, + pub error: String, +} + +/// Envoi accepté. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct Sent { + /// Mis en file derrière le tour en cours. + pub queued: bool, + /// Réessai d'un envoi déjà accepté. + pub dup: bool, +} + +enum StartError { + /// Le tour de la discussion tient déjà le verrou. + ConvBusy, + /// Le moteur est tenu par le Go (tâche, bench, optimiseur) : 409. + Busy(String), + /// Moteur pas prêt : 503. + NotReady(String), +} + +#[derive(Default)] +struct Queue { + items: VecDeque, + recent: VecDeque, +} + +fn obj(k: &str, v: impl Into) -> Delta { + let mut d = Map::new(); + d.insert(k.into(), v.into()); + d +} + +type BoxFut = Pin + Send>>; + +/// Le chat tenu par la façade : la discussion ouverte, ses tours et sa file. +pub struct ChatService { + pub sessions: Arc>, + pub shared: Arc, + q: Mutex, + lease_ttl: Duration, +} + +impl ChatService { + pub fn new(sessions: Arc>, shared: Arc) -> Arc { + Arc::new(ChatService { + sessions, + shared, + q: Mutex::new(Queue::default()), + lease_ttl: LEASE_TTL, + }) + } + + fn queue(&self) -> std::sync::MutexGuard<'_, Queue> { + self.q.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// `seenCIDLocked` : retient `cid` et dit s'il avait déjà été vu. + fn seen_cid(&self, cid: &str) -> bool { + if cid.is_empty() { + return false; + } + let mut q = self.queue(); + if q.recent.iter().any(|c| c == cid) { + return true; + } + q.recent.push_back(cid.to_string()); + while q.recent.len() > RECENT_CIDS { + q.recent.pop_front(); + } + false + } + + /// Un envoi refusé peut être renvoyé : son cid ne reste pas marqué. + fn forget_cid(&self, cid: &str) { + self.queue().recent.retain(|c| c != cid); + } + + /// `/api/chat/send` (`EnqueueOrStart`) : corps tel que l'UI l'envoie. + pub async fn send(self: &Arc, body: Value) -> Result { + let cid = body + .get("cid") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + if self.seen_cid(&cid) { + return Ok(Sent { + queued: false, + dup: true, + }); + } + let refuse = |me: &Self, status: u16, error: String| { + me.forget_cid(&cid); + Err(Refused { status, error }) + }; + let prep = match self + .shared + .store + .internal_post("/internal/v1/turn/start", &body) + .await + { + Ok((s, v)) if s.as_u16() == 200 => { + serde_json::from_value::(v).unwrap_or_default() + } + Ok((s, v)) => { + let e = v + .get("error") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + return refuse(self, if s.as_u16() == 400 { 400 } else { 503 }, e); + } + Err(e) => return refuse(self, 503, e), + }; + let prep = Prepared { + text: body + .get("message") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(), + temperature: body + .get("temperature") + .and_then(Value::as_f64) + .unwrap_or(0.0), + ..prep + }; + let conv = self.sessions.conv(); + if conv.is_generating() { + self.queue().items.push_back(prep); + return Ok(Sent { + queued: true, + dup: false, + }); + } + match self.start(prep.clone()).await { + Ok(()) => Ok(Sent::default()), + // Un autre tour a démarré entre-temps (deux appareils) : en file. + Err(StartError::ConvBusy) => { + self.queue().items.push_back(prep); + self.start_queued_if_any().await; + Ok(Sent { + queued: true, + dup: false, + }) + } + Err(StartError::Busy(e)) => refuse(self, 409, e), + Err(StartError::NotReady(e)) => refuse(self, 503, e), + } + } + + /// `healthCheck` : preset externe, toujours prêt ; sinon `/health`. + async fn engine_ready(&self) -> bool { + let cfg = self.shared.store.config().await; + if crate::engstatus::is_external(&cfg) { + return true; + } + let url = format!("http://localhost:{}/health", llm_port(&cfg)); + self.shared.http.health(&url).await + } + + /// `StartTurn`. Futur en boîte : la fin d'un tour relance la file, donc + /// `start` → `generate` → `start`, un cycle que le type doit couper. + fn start( + self: &Arc, + prep: Prepared, + ) -> Pin> + Send + '_>> { + Box::pin(self.start_inner(prep)) + } + + async fn start_inner(self: &Arc, prep: Prepared) -> Result<(), StartError> { + if !self.engine_ready().await { + return Err(StartError::NotReady(ERR_MODEL_LOADING.into())); + } + let conv = self.sessions.conv().clone(); + let (tx, cancel) = Cancel::new(); + let tx = Arc::new(tx); + let t = tx.clone(); + let epoch = conv + .begin_turn(Box::new(move || { + let _ = t.send(true); + })) + .map_err(|_| StartError::ConvBusy)?; + let lease = match EngineLease::acquire(self.shared.store.clone(), self.lease_ttl).await { + Ok(l) => l, + Err(e) => { + conv.release_turn(epoch); + return Err(match e { + LeaseError::Busy { error, .. } => StartError::Busy(error), + LeaseError::Other(e) => StartError::NotReady(e), + }); + } + }; + conv.with_state_if(epoch, |st| st.messages.push(prep.user_message())); + conv.append_delta(epoch, prep.user_delta()); + if prep.code_hint && !prep.cap("Code") { + conv.append_delta(epoch, obj("code_hint", true)); + } + // Instantané tout de suite, écriture par l'écrivain : un crash en + // pleine génération ne perd pas le message de l'utilisateur. + self.sessions.persist_async().await; + let me = self.clone(); + let fut: BoxFut = + Box::pin(async move { me.generate(prep, epoch, cancel, tx, lease).await }); + tokio::spawn(fut); + Ok(()) + } + + /// `drainQueued` : messages en file injectés dans le tour en cours, + /// journalisés pour tous les appareils. + fn drain_queued(&self, epoch: i64) -> Vec { + let conv = self.sessions.conv(); + if conv.epoch() != epoch { + return Vec::new(); + } + let items: Vec = self.queue().items.drain(..).collect(); + items + .into_iter() + .map(|p| { + conv.append_delta(epoch, p.user_delta()); + p.user_message() + }) + .collect() + } + + /// `startQueuedIfAny` : le prochain message de la file part, s'il en reste + /// et que rien ne tourne. Rend vrai si un tour a démarré. + async fn start_queued_if_any(self: &Arc) -> bool { + loop { + let conv = self.sessions.conv(); + if conv.is_generating() { + return false; + } + let Some(next) = self.queue().items.pop_front() else { + return false; + }; + let epoch = conv.epoch(); + match self.start(next.clone()).await { + Ok(()) => return true, + // Un tour vient de partir : le message garde sa place, ce tour + // le reprendra (injection ou fin de tour). + Err(StartError::ConvBusy) => { + self.queue().items.push_front(next); + return false; + } + Err(StartError::Busy(e) | StartError::NotReady(e)) => { + conv.append_delta( + epoch, + obj("error", format!("message en attente non envoyé : {e}")), + ); + } + } + } + } + + /// `dropQueued` : bouton stop, l'utilisateur reprend la main. + fn drop_queued(&self, epoch: i64) { + let n = std::mem::take(&mut self.queue().items).len(); + if n > 0 { + self.sessions + .conv() + .append_delta(epoch, obj("queue_dropped", n)); + } + } + + /// `/api/chat/stop` : le tour en cours, ou le détenteur Go du moteur + /// (tâche planifiée, bench, optimiseur). + pub async fn stop(&self) { + let conv = self.sessions.conv(); + if conv.is_generating() { + conv.stop(); + } else if let Err(e) = engine_stop(&self.shared.store).await { + eprintln!("[chat] arrêt du détenteur du moteur impossible : {e}"); + } + } + + /// `compactAndPublish` : bannière, journal, installation du résultat, + /// jauge de contexte. Rend l'historique compacté, `None` s'il n'a pas + /// changé. + #[allow(clippy::too_many_arguments)] + async fn compact_and_publish( + &self, + summarizer: &EngineSummarizer, + ccfg: &CompactConfig, + caps: CompactCaps, + mem_off: bool, + epoch: i64, + phase: &str, + msgs: &[Message], + ctx_used: i64, + opt: &CompactOpts, + ) -> Option> { + let conv = self.sessions.conv(); + conv.append_delta(epoch, obj("compacting", true)); + let recall = StoreRecall(self.shared.store.clone()); + let c = Compactor { + cfg: ccfg, + summarizer, + recall: &recall, + memory: &self.shared.compact_memory, + criteria: "", + }; + let (compacted, changed) = c.compact_messages_noted(msgs, caps, opt).await; + conv.append_delta(epoch, obj("compacting", false)); + log_compact(ccfg, phase, ctx_used, msgs, &compacted, changed); + if !changed { + conv.append_delta(epoch, obj("compact_noop", true)); + return None; + } + let compacted = if caps.agent { + remind_read_mem_pages(compacted, msgs, mem_off) + } else { + compacted + }; + let compacted = strip_context_updates(compacted); + let overhead = (ctx_used - estimate_tokens(msgs, &ccfg.echo)).max(0); + let est = estimate_tokens(&compacted, &ccfg.echo) + overhead; + conv.with_state_if(epoch, |st| { + st.messages = compacted.clone(); + st.proj_snap = None; + st.ctx_used = est; + st.ctx_used_len = compacted.len(); + }); + conv.append_delta(epoch, obj("compacted", true)); + conv.append_delta(epoch, obj("ctx_used", est)); + Some(compacted) + } + + /// `generate` : un tour complet, détaché — seul l'arrêt l'interrompt. + async fn generate( + self: Arc, + prep: Prepared, + epoch: i64, + cancel: Cancel, + cancel_tx: Arc>, + mut lease: EngineLease, + ) { + let turn_start = Instant::now(); + let conv = self.sessions.conv().clone(); + let store = self.shared.store.clone(); + let cfg = store.config().await; + let local_key = store.get_str(BK_STATE, "api_key").await; + let ep = resolve_chat_endpoint(&cfg, &local_key); + let local = !ep.external; + let echo = echo_policy_for(&cfg, &self.shared.echo_refusals, None); + let ccfg = CompactConfig::from_config(&cfg, echo.clone()); + let port = llm_port(&cfg); + let conv_id = conv.id(); + let agent = prep.cap("Agent"); + let mem_off = prep.turn_caps.get("Mem").and_then(Value::as_str) == Some("off"); + let ccaps = CompactCaps { + agent, + code: prep.cap("Code"), + }; + let summarizer = EngineSummarizer { + http: self.shared.http.clone(), + ep: ep.clone(), + cfg: cfg.clone(), + patterns: self.shared.patterns.clone(), + port, + }; + let ctx_now = |st: &super::persist::ConvState, msgs: &[Message]| { + if local { + ctx_pending(st.ctx_used, st.ctx_used_len, msgs, &echo) + } else { + st.ctx_used + } + }; + + // Instantané de la vue modèle et du contexte. + let snap = conv.with_state_if(epoch, |st| { + let msgs = st.messages.clone(); + let used = ctx_now(st, &msgs); + let peak = std::mem::take(&mut st.gen_peak); + (msgs, used, peak) + }); + let Some((mut msgs, ctx_used, peak)) = snap else { + lease.release().await; + return; + }; + + // Compaction de début de tour. + let mut start_est = ctx_used; + if compact_needed(&ccfg, &msgs, ctx_used, peak) + && !self + .shared + .compact_memory + .refused_skip(&ccfg, &conv_id, ctx_used, &msgs) + { + let opt = compact_opts_for( + &ccfg, + &self.shared.compact_memory, + false, + &msgs, + ctx_used, + &conv_id, + |_| TurnView::default(), + ); + if let Some(out) = self + .compact_and_publish( + &summarizer, + &ccfg, + ccaps, + mem_off, + epoch, + "début-tour", + &msgs, + ctx_used, + &opt, + ) + .await + { + msgs = out; + start_est = conv + .with_state_if(epoch, |st| st.ctx_used) + .unwrap_or(start_est); + } + } + + // Vue d'envoi : tête du Go (prompt du preset, consignes, projet), + // l'historique sans bloc , le préambule. + let view = store + .internal_post("/internal/v1/turn/view", &json!({ "caps": prep.turn_caps })) + .await; + let (head, prefix, tools) = match view { + Ok((s, v)) if s.as_u16() == 200 => ( + serde_json::from_value::>(v["head"].clone()).unwrap_or_default(), + v["prefix"].as_str().unwrap_or("").to_string(), + serde_json::from_value::>(v["tools"].clone()).unwrap_or_default(), + ), + other => { + let e = match other { + Ok((_, v)) => v["error"] + .as_str() + .unwrap_or("vue du tour refusée") + .to_string(), + Err(e) => e, + }; + conv.append_delta(epoch, obj("error", e)); + self.finish(epoch, &cancel, lease, turn_start).await; + return; + } + }; + if agent { + // Une discussion qui porte encore des blocs d'une époque où + // PROJ_SNAPSHOT servait en est nettoyée (comme assembleTurn). + conv.with_state_if(epoch, |st| { + if st.proj_snap.is_some() || st.messages.iter().any(|m| m.ctx_upd) { + st.proj_snap = None; + st.messages = strip_context_updates(std::mem::take(&mut st.messages)); + } + }); + } + let hist = strip_context_updates(msgs); + let mut sent = head; + sent.extend(hist.iter().cloned()); + let sent = inject_prefix(sent, &prefix); + + let temperature = if prep.temperature == 0.0 { + 0.7 + } else { + prep.temperature + }; + let reasoning = turn_reasoning(&cfg, &self.shared.effort); + let tc = TurnConfig { + cfg: cfg.clone(), + ep: ep.clone(), + tools: tools.clone(), + temperature, + code: prep.cap("Code"), + reasoning: reasoning.clone(), + echo: echo.clone(), + compact: ccfg.clone(), + patterns: self.shared.patterns.clone(), + }; + let me = self.clone(); + let host = GoTurnHost { + shared: self.shared.clone(), + conv_id: conv_id.clone(), + caps: prep.turn_caps.clone(), + cfg: cfg.clone(), + port, + summarizer: summarizer.clone(), + compact: TurnCompact { + cfg: ccfg.clone(), + caps: ccaps, + criteria: String::new(), + main_conv: conv_id.clone(), + tools, + effort: reasoning.effort.clone(), + kwargs: reasoning.kwargs.clone().unwrap_or_default(), + }, + queued: Box::new(move || me.drain_queued(epoch)), + }; + + // Chaque événement devient un delta du journal. + let mut content = String::new(); + let mut new_base: Option> = None; + let mut final_echo: Option = None; + let mut saw_usage = false; + let mut turn_peak = 0i64; + let mut cb = |ev: StreamEvent| -> bool { + match ev { + StreamEvent::Err(e) => { + conv.append_delta(epoch, obj("error", e)); + } + StreamEvent::ToolUsed(t) => { + // Le texte écrit avant l'appel est déjà dans le message + // assistant qui porte l'appel : pas deux fois. + content.clear(); + conv.append_delta(epoch, tool_used_delta(&t)); + } + StreamEvent::NewHistory(h) => new_base = Some(history_from_view(&h, &hist)), + StreamEvent::Echo(e) => final_echo = Some(e), + StreamEvent::Compacting(b) => { + conv.append_delta(epoch, obj("compacting", b)); + } + StreamEvent::Stats(s) => { + if s.prompt_tokens_total > 0 { + if !saw_usage && local && start_est > 0 { + log_ctx_estimate("début-tour", start_est, s.prompt_tokens_total); + } + saw_usage = true; + conv.with_state_if(epoch, |st| st.ctx_used = s.ctx_after()); + } + if local { + turn_peak = turn_peak.max(s.gen_tokens); + } + conv.append_delta( + epoch, + obj("stats", serde_json::to_value(&s).unwrap_or_default()), + ); + } + StreamEvent::Ask(a) => { + conv.append_delta( + epoch, + obj("ask", serde_json::to_value(&a).unwrap_or_default()), + ); + } + StreamEvent::DropReasoning => { + conv.append_delta(epoch, obj("drop_reasoning", true)); + } + StreamEvent::Reasoning(r) if !r.is_empty() => { + conv.append_delta(epoch, obj("reasoning_content", r)); + } + StreamEvent::Content(c) if !c.is_empty() => { + content.push_str(&c); + conv.append_delta(epoch, obj("content", c)); + } + _ => {} + } + true // génération détachée : jamais interrompue par un abonné + }; + + let res = { + let turn = run_turn(&host, &tc, sent, &cancel, &mut cb); + tokio::pin!(turn); + tokio::select! { + r = &mut turn => r, + sig = lease.interrupted() => { + if sig == LeaseSignal::Lost { + eprintln!("[chat] bail du moteur perdu en plein tour : arrêt"); + } + let _ = cancel_tx.send(true); + turn.await + } + } + }; + drop(host); + + // Vue modèle rangée : outils puis réponse finale — sauf si un reset + // ou une bascule est passé entre-temps. + let after = conv.with_state_if(epoch, |st| { + if let Some(base) = new_base.take() { + st.messages = base; + } + st.messages.extend(res.extra.iter().cloned()); + if !content.trim().is_empty() { + let a = Message { + role: "assistant".into(), + content: Some(Value::String(content.clone())), + ..Default::default() + }; + st.messages.push(with_echo(a, final_echo.as_ref())); + } + st.messages = drop_stray_nudges(std::mem::take(&mut st.messages)); + if saw_usage { + st.ctx_used_len = st.messages.len(); + } + st.gen_peak = st.gen_peak.max(turn_peak); + let msgs = st.messages.clone(); + (ctx_now(st, &msgs), st.gen_peak, msgs) + }); + let Some((mut ctx_used, peak, msgs)) = after else { + lease.release().await; + return; // reset pendant le tour : il a déjà enregistré le fil vide + }; + + // Moteur muet sur l'usage : l'estimation, sinon la jauge reste à zéro. + if !saw_usage { + let est = estimate_tokens(&msgs, &echo); + conv.with_state_if(epoch, |st| { + st.ctx_used = est; + st.ctx_used_len = msgs.len(); + }); + conv.append_delta(epoch, obj("ctx_used", est)); + ctx_used = est; + } + + // Compaction de fin de tour : la jauge retombe tout de suite, le tour + // suivant démarre avec de la marge. Pas après un stop. + if !cancel.is_cancelled() + && compact_needed(&ccfg, &msgs, ctx_used, peak) + && !self + .shared + .compact_memory + .refused_skip(&ccfg, &conv_id, ctx_used, &msgs) + { + let opt = compact_opts_for( + &ccfg, + &self.shared.compact_memory, + false, + &msgs, + ctx_used, + &conv_id, + |_| TurnView::default(), + ); + self.compact_and_publish( + &summarizer, + &ccfg, + ccaps, + mem_off, + epoch, + "fin-tour", + &msgs, + ctx_used, + &opt, + ) + .await; + } + self.finish(epoch, &cancel, lease, turn_start).await; + } + + /// Fin d'un tour (le `defer` de `generate`). + async fn finish( + self: &Arc, + epoch: i64, + cancel: &Cancel, + lease: EngineLease, + turn_start: Instant, + ) { + let conv = self.sessions.conv(); + conv.append_delta(epoch, obj("turn_done", true)); + let current = conv.end_turn(epoch); + lease.release().await; + if !current { + return; // reset pendant le tour : il a déjà enregistré le fil vide + } + self.sessions.persist().await; + if cancel.is_cancelled() { + self.drop_queued(epoch); + return; + } + if !self.start_queued_if_any().await { + let ms = turn_start.elapsed().as_millis() as u64; + if let Err(e) = self + .shared + .store + .internal_json( + hyper::Method::POST, + "/internal/v1/turn/done", + Some(&json!({ "ms": ms })), + ) + .await + { + eprintln!("[chat] fin de tour non signalée : {e}"); + } + } + } +} + +impl ChatShared { + /// L'état partagé des tours, autour de l'écrivain des discussions. + /// `img_dir` : `LOKI_HOME/chatimg`. + pub fn for_sessions( + sessions: &Sessions, + store: crate::store::Store, + img_dir: std::path::PathBuf, + ) -> Self { + ChatShared { + http: EngineHttp::new(), + store, + persister: sessions.persister().clone(), + effort: super::wire::EffortFallbacks::new(), + echo_refusals: super::wire::EchoRefusals::new(), + compact_memory: super::compact::CompactMemory::new(), + images: super::host::ImageCache::new(img_dir), + patterns: super::stream::RetryPatterns::builtin(), + } + } +} diff --git a/rust/loki-rs/src/chat/history.rs b/rust/loki-rs/src/chat/history.rs new file mode 100644 index 0000000..15bdc26 --- /dev/null +++ b/rust/loki-rs/src/chat/history.rs @@ -0,0 +1,536 @@ +//! Début et fin d'un tour, côté historique — port de `historyFromView` +//! (`chat_conversation.go`), `dropStrayNudges` (`llm_injected.go`), du rappel +//! des pages mémoire lues (`chat_mem_pinned.go`), de l'assemblage +//! `InjectSkills` (`llm_client.go`) et de la forme des événements `tool_used` +//! du journal (`generate`). + +use std::collections::HashMap; + +use serde_json::{Map, Value, json}; + +use super::compact::{is_loki_injected, msg_text}; +use super::stream::call_args; +use super::types::{Message, ToolUsedEvent}; +use super::wire::{MEM_REMINDER_PREFIX, is_project_system}; + +/// Noms de pages listés au plus par le rappel (`memReminderMaxNames`). +pub const MEM_REMINDER_MAX_NAMES: usize = 24; + +fn system(text: String) -> Message { + Message { + role: "system".into(), + content: Some(Value::String(text)), + ..Default::default() + } +} + +/// `InjectSkills` avec le préfixe déjà calculé (préambule d'agent + briefing +/// machine, `/internal/v1/turn/view`) : fusionné dans le premier système s'il +/// n'est pas un message projet, sinon posé en tête. Préfixe vide : rien. +pub fn inject_prefix(msgs: Vec, prefix: &str) -> Vec { + if prefix.is_empty() { + return msgs; + } + if let Some(first) = msgs.first() + && first.role == "system" + && !is_project_system(first) + { + let existing = match &first.content { + Some(Value::String(s)) => s.as_str(), + _ => "", + }; + let mut out = msgs.clone(); + out[0] = system(format!("{prefix}\n\n{existing}")); + return out; + } + let mut out = Vec::with_capacity(msgs.len() + 1); + out.push(system(prefix.to_string())); + out.extend(msgs); + out +} + +/// `historyFromView` : l'historique à ranger après une compaction en cours de +/// tour — la vue publiée sans les systèmes que la vue d'envoi avait posés en +/// tête. Un système de tête n'est gardé que s'il figurait en tête de `hist` +/// (le premier, éventuellement fusionné avec le préambule, est rendu tel qu'il +/// était dans l'historique). +pub fn history_from_view(view: &[Message], hist: &[Message]) -> Vec { + let mut own: HashMap = HashMap::new(); + for m in hist.iter().take_while(|m| m.role == "system") { + *own.entry(msg_text(m)).or_default() += 1; + } + let mut out = Vec::new(); + let mut i = 0; + while i < view.len() && view[i].role == "system" { + let k = msg_text(&view[i]); + if let Some(n) = own.get_mut(&k) + && *n > 0 + { + *n -= 1; + out.push(view[i].clone()); + i += 1; + continue; + } + if i == 0 + && let Some(h0) = hist.first() + && h0.role == "system" + { + let h = msg_text(h0); + if !h.is_empty() + && own.get(&h).is_some_and(|n| *n > 0) + && k.ends_with(&format!("\n\n{h}")) + { + if let Some(n) = own.get_mut(&h) { + *n -= 1; + } + out.push(h0.clone()); + } + } + i += 1; + } + out.extend_from_slice(&view[i..]); + out +} + +/// `dropStrayNudges` : retire les rappels de loki en porte-à-faux (en toute +/// fin, ou collés à un autre `user`) — sinon deux `user` d'affilée au tour +/// suivant. +pub fn drop_stray_nudges(mut msgs: Vec) -> Vec { + loop { + let n = msgs.len(); + let drop = (0..n).find(|&i| { + is_loki_injected(&msgs[i]) + && (i == n - 1 + || msgs[i + 1].role == "user" + || (i > 0 && msgs[i - 1].role == "user")) + }); + match drop { + Some(i) => { + msgs.remove(i); + } + None => return msgs, + } + } +} + +/// `parseReminderNames` : noms listés par un rappel, vide si ce n'en est pas. +pub fn parse_reminder_names(text: &str) -> Vec { + if !text.starts_with(MEM_REMINDER_PREFIX) { + return Vec::new(); + } + let line = text.split('\n').next().unwrap_or(""); + let Some(i) = line.rfind(": ") else { + return Vec::new(); + }; + line[i + 2..] + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .collect() +} + +/// `collectReadPageNames` : pages lues (mem_read réussis, puis rappels +/// hérités), dans l'ordre, sans doublon. +pub fn collect_read_page_names(msgs: &[Message]) -> Vec { + let mut read_call: HashMap<&str, String> = HashMap::new(); + for m in msgs { + for tc in &m.tool_calls { + if tc.function.name != "mem_read" || tc.id.is_empty() { + continue; + } + if let Some(Value::String(f)) = call_args(&tc.function.arguments).get("file") + && !f.is_empty() + { + read_call.insert(tc.id.as_str(), f.clone()); + } + } + } + let mut order: Vec = Vec::new(); + let mut add = |name: &str| { + let name = name.trim(); + if !name.is_empty() && !order.iter().any(|n| n == name) { + order.push(name.to_string()); + } + }; + for m in msgs { + match m.role.as_str() { + "system" => { + for n in parse_reminder_names(&msg_text(m)) { + add(&n); + } + } + "tool" => { + let Some(name) = read_call.get(m.tool_call_id.as_str()) else { + continue; + }; + if msg_text(m).trim().starts_with("[erreur]") { + continue; + } + add(name); + } + _ => {} + } + } + order +} + +/// `buildReminderMessage` : le rappel, les plus récentes en dernier ; `None` +/// sans page lue. +pub fn build_reminder_message(names: &[String]) -> Option { + if names.is_empty() { + return None; + } + let names = &names[names.len().saturating_sub(MEM_REMINDER_MAX_NAMES)..]; + Some(system(format!( + "{MEM_REMINDER_PREFIX} Memory pages you read earlier: {}\nTheir full content is no longer inline after context compaction — if any of them is relevant to what you're doing, mem_read it again so you have all the rules and info you need before answering.", + names.join(", ") + ))) +} + +/// `stripReminder`. +pub fn strip_reminder(msgs: Vec) -> Vec { + msgs.into_iter() + .filter(|m| { + !(m.role == "system" + && matches!(&m.content, Some(Value::String(s)) if s.starts_with(MEM_REMINDER_PREFIX))) + }) + .collect() +} + +/// `remindReadMemPages` : rappel reconstruit d'après l'historique d'avant la +/// compaction, en tête de l'historique compacté. Mémoire coupée : rien. +pub fn remind_read_mem_pages( + compacted: Vec, + source: &[Message], + mem_off: bool, +) -> Vec { + if mem_off { + return compacted; + } + let names = collect_read_page_names(source); + let mut out = strip_reminder(compacted); + if let Some(m) = build_reminder_message(&names) { + out.insert(0, m); + } + out +} + +/// L'événement `tool_used` du journal (forme de `generate`) : le corps en +/// cours de frappe, le diff, la capture, l'aperçu et l'id du résultat complet +/// seulement quand ils sont là. +pub fn tool_used_delta(t: &ToolUsedEvent) -> Map { + let mut tu = Map::new(); + tu.insert("name".into(), t.name.clone().into()); + tu.insert("label".into(), t.label.clone().into()); + tu.insert("result".into(), t.result.clone().into()); + tu.insert("done".into(), t.done.into()); + tu.insert("typing".into(), t.typing.into()); + if !t.body.is_empty() { + tu.insert("body".into(), t.body.clone().into()); + tu.insert("body_lines".into(), t.body_lines.into()); + if t.body_tail { + tu.insert("body_tail".into(), true.into()); + } + } + if !t.diff.is_empty() { + tu.insert("diff".into(), json!(t.diff)); + tu.insert("added".into(), t.added.into()); + tu.insert("removed".into(), t.removed.into()); + } + if !t.image.is_empty() { + tu.insert("image".into(), t.image.clone().into()); + } + if !t.shot.is_empty() { + tu.insert("shot".into(), t.shot.clone().into()); + } + if t.result_chars > 0 { + tu.insert("result_chars".into(), t.result_chars.into()); + } + if !t.result_id.is_empty() { + tu.insert("result_id".into(), t.result_id.clone().into()); + } + let mut d = Map::new(); + d.insert("tool_used".into(), Value::Object(tu)); + d +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::chat::compact::{THINK_NUDGE_FIRST, THINK_NUDGE_STUCK}; + use crate::chat::stream::retry_corrective; + use crate::chat::types::{DiffLine, ToolCall, ToolCallFunc}; + use crate::chat::wire::PROJECT_CONTEXT_PREFIX; + + fn msg(role: &str, s: &str) -> Message { + Message { + role: role.into(), + content: Some(json!(s)), + ..Default::default() + } + } + fn um(s: &str) -> Message { + msg("user", s) + } + fn am(s: &str) -> Message { + msg("assistant", s) + } + fn tm(s: &str) -> Message { + msg("tool", s) + } + fn atc(name: &str) -> Message { + Message { + role: "assistant".into(), + tool_calls: vec![ToolCall { + id: "c1".into(), + kind: "function".into(), + function: ToolCallFunc { + name: name.into(), + arguments: "{}".into(), + }, + ..Default::default() + }], + ..Default::default() + } + } + fn read(id: &str, file: &str) -> Message { + Message { + role: "assistant".into(), + tool_calls: vec![ToolCall { + id: id.into(), + function: ToolCallFunc { + name: "mem_read".into(), + arguments: format!("{{\"file\":\"{file}\"}}"), + }, + ..Default::default() + }], + ..Default::default() + } + } + fn res(id: &str, s: &str) -> Message { + Message { + role: "tool".into(), + tool_call_id: id.into(), + content: Some(json!(s)), + ..Default::default() + } + } + + #[test] + fn prefixe_fusionne_ou_pose_en_tete() { + let sys = msg("system", "/sys perso"); + let out = inject_prefix(vec![sys, um("q")], "préambule"); + assert_eq!(out[0], msg("system", "préambule\n\n/sys perso")); + assert_eq!(out.len(), 2); + let proj = msg("system", &format!("{PROJECT_CONTEXT_PREFIX} x")); + let out = inject_prefix(vec![proj.clone(), um("q")], "préambule"); + assert_eq!( + (out.len(), &out[1]), + (3, &proj), + "un message projet n'absorbe pas le préambule" + ); + assert_eq!(inject_prefix(vec![um("q")], ""), vec![um("q")]); + } + + #[test] + fn historique_depuis_la_vue() { + let rem = msg( + "system", + &format!("{MEM_REMINDER_PREFIX} Memory pages you read earlier: a.md"), + ); + let sys = msg("system", "/sys perso"); + let proj = msg("system", &format!("{PROJECT_CONTEXT_PREFIX} x")); + let u = um("q"); + let pre = msg("system", "préambule"); + let cases = [ + ( + "injectés seuls", + vec![pre.clone(), proj.clone(), u.clone()], + vec![u.clone()], + vec![u.clone()], + ), + ( + "rappel gardé", + vec![pre.clone(), proj.clone(), rem.clone(), u.clone()], + vec![rem.clone(), u.clone()], + vec![rem.clone(), u.clone()], + ), + ( + "système fusionné rendu", + vec![msg("system", "préambule\n\n/sys perso"), u.clone()], + vec![sys.clone(), u.clone()], + vec![sys.clone(), u.clone()], + ), + ( + "sans système", + vec![u.clone()], + vec![u.clone()], + vec![u.clone()], + ), + ]; + for (name, view, hist, want) in cases { + assert_eq!(history_from_view(&view, &hist), want, "{name}"); + } + } + + #[test] + fn rappels_en_porte_a_faux() { + let n = um(THINK_NUDGE_FIRST); + type Case = (&'static str, Vec, Option>); + let cases: Vec = vec![ + ( + "rappel répondu", + vec![um("q"), atc("glob"), tm("r"), n.clone(), am("fini")], + None, + ), + ( + "resté sans réponse", + vec![um("q"), atc("glob"), tm("r"), n.clone()], + Some(vec![um("q"), atc("glob"), tm("r")]), + ), + ( + "ajout juste derrière", + vec![ + um("q"), + atc("glob"), + tm("r"), + n.clone(), + um("et aussi"), + am("fini"), + ], + Some(vec![ + um("q"), + atc("glob"), + tm("r"), + um("et aussi"), + am("fini"), + ]), + ), + ( + "collé derrière la demande", + vec![um("q"), n.clone(), am("fini")], + Some(vec![um("q"), am("fini")]), + ), + ( + "correctif sans réponse", + vec![ + um("q"), + am(""), + um(&retry_corrective("")), + ], + Some(vec![um("q"), am("")]), + ), + ( + "deux rappels en fin", + vec![ + um("q"), + atc("glob"), + tm("r"), + n.clone(), + um(THINK_NUDGE_STUCK), + ], + Some(vec![um("q"), atc("glob"), tm("r")]), + ), + ( + "vraie demande en fin", + vec![um("q"), am("a"), um("suite")], + None, + ), + ]; + for (name, input, want) in cases { + let want = want.unwrap_or_else(|| input.clone()); + assert_eq!(drop_stray_nudges(input), want, "{name}"); + } + } + + #[test] + fn rappel_des_pages_lues() { + let msgs = vec![ + um("applique"), + read("c1", "regles-tache.md"), + res("c1", "RÈGLE 1"), + ]; + let names = collect_read_page_names(&msgs); + assert_eq!(names, ["regles-tache.md"]); + let m = build_reminder_message(&names).unwrap(); + let txt = msg_text(&m); + assert!(txt.contains("regles-tache.md") && txt.contains("mem_read")); + assert_eq!(parse_reminder_names(&txt), names); + // Accumulé à travers les compactions, dédoublonné, sans lecture ratée. + let msgs = vec![ + m.clone(), + um("[CONTEXT COMPACTED] résumé…"), + read("c2", "procedure.md"), + res("c2", "procédure"), + read("c3", "procedure.md"), + res("c3", "v2"), + read("e", "absente.md"), + res("e", "[erreur] page introuvable"), + ]; + assert_eq!( + collect_read_page_names(&msgs), + ["regles-tache.md", "procedure.md"] + ); + // Borné aux plus récentes. + let many: Vec = (0..MEM_REMINDER_MAX_NAMES + 10) + .map(|i| format!("p{i}.md")) + .collect(); + let got = parse_reminder_names(&msg_text(&build_reminder_message(&many).unwrap())); + assert_eq!(got.len(), MEM_REMINDER_MAX_NAMES); + assert_eq!(got.last(), many.last()); + // Reconstruit en tête de l'historique compacté ; rien si mémoire coupée. + let compacted = vec![m.clone(), um("résumé")]; + let out = remind_read_mem_pages(compacted.clone(), &msgs, false); + assert_eq!(out.len(), 2); + assert_eq!( + parse_reminder_names(&msg_text(&out[0])), + ["regles-tache.md", "procedure.md"] + ); + assert_eq!( + remind_read_mem_pages(compacted.clone(), &msgs, true), + compacted + ); + assert_eq!(strip_reminder(vec![m, um("x")]), vec![um("x")]); + } + + #[test] + fn evenement_d_outil_du_journal() { + let d = tool_used_delta(&ToolUsedEvent { + name: "bash".into(), + label: "ls".into(), + ..Default::default() + }); + assert_eq!( + Value::Object(d), + json!({"tool_used": {"name": "bash", "label": "ls", "result": "", "done": false, "typing": false}}) + ); + let d = tool_used_delta(&ToolUsedEvent { + name: "edit".into(), + done: true, + body: "x".into(), + body_lines: 1, + body_tail: true, + diff: vec![DiffLine { + op: "+".into(), + text: "x".into(), + }], + added: 1, + result_chars: 3000, + result_id: "c1.ab".into(), + ..Default::default() + }); + let tu = &d["tool_used"]; + assert_eq!(tu["body_tail"], true); + assert_eq!(tu["diff"], json!([{"op": "+", "text": "x"}])); + assert_eq!( + (tu["added"].clone(), tu["removed"].clone()), + (json!(1), json!(0)) + ); + assert_eq!( + (tu["result_chars"].clone(), tu["result_id"].clone()), + (json!(3000), json!("c1.ab")) + ); + assert!(tu.get("image").is_none() && tu.get("shot").is_none()); + } +} diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs index a2919e8..9228ea1 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -161,6 +161,7 @@ impl ImageCache { } /// Le moteur vu de la compaction (`postSummary`). +#[derive(Clone)] pub struct EngineSummarizer { pub http: EngineHttp, pub ep: ChatEndpoint, diff --git a/rust/loki-rs/src/chat/lease.rs b/rust/loki-rs/src/chat/lease.rs index 0920aec..f17fe44 100644 --- a/rust/loki-rs/src/chat/lease.rs +++ b/rust/loki-rs/src/chat/lease.rs @@ -126,8 +126,13 @@ impl EngineLease { /// Se résout au premier stop ou à la perte du bail, et rend lequel. pub async fn interrupted(&mut self) -> LeaseSignal { - match self.signal.wait_for(|s| *s != LeaseSignal::Held).await { - Ok(s) => *s, + let got = self + .signal + .wait_for(|s| *s != LeaseSignal::Held) + .await + .map(|s| *s); + match got { + Ok(s) => s, // Tâche de renouvellement terminée sans signal : rien à attendre. Err(_) => std::future::pending().await, } diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs index 66e54ff..9e65044 100644 --- a/rust/loki-rs/src/chat/mod.rs +++ b/rust/loki-rs/src/chat/mod.rs @@ -15,11 +15,15 @@ //! - [`lease`] : bail du moteur (verrou de génération partagé avec le Go) ; //! - [`nudges`] : budget d'outils, rappels, appels répétés, aperçus ; //! - [`turn`] : la boucle d'un tour (complétions, outils, filets) ; -//! - [`host`] : le tour branché sur le moteur, les outils du Go et la base. +//! - [`host`] : le tour branché sur le moteur, les outils du Go et la base ; +//! - [`history`] : début et fin d'un tour côté historique ; +//! - [`generate`] : un tour de bout en bout (envoi, file, fin de tour). pub mod compact; pub mod conv; pub mod engine; +pub mod generate; +pub mod history; pub mod host; pub mod journal; pub mod lease; diff --git a/rust/loki-rs/src/chat/persist.rs b/rust/loki-rs/src/chat/persist.rs index 81079eb..f7eebed 100644 --- a/rust/loki-rs/src/chat/persist.rs +++ b/rust/loki-rs/src/chat/persist.rs @@ -165,6 +165,12 @@ pub struct ConvState { pub seq: i64, pub ctx_used: i64, pub proj_snap: Option, + /// Messages que `ctx_used` couvre déjà (0 = inconnu) — en mémoire + /// seulement : un rechargement repart du seul `ctx_used`. + pub ctx_used_len: usize, + /// Génération la plus longue du dernier tour (moteur local), pour la garde + /// de marge de la compaction — en mémoire seulement. + pub gen_peak: i64, } /// Forme écrite, empruntée : l'instantané en plein tour écrit un journal @@ -950,8 +956,16 @@ mod tests { }, ..Default::default() }), + ..Default::default() }; let b = st.to_json(); + // Les compteurs en mémoire ne s'écrivent pas. + let mem = ConvState { + ctx_used_len: 3, + gen_peak: 900, + ..st.clone() + }; + assert_eq!(mem.to_json(), b); assert_eq!( String::from_utf8(b.clone()).unwrap(), r#"{"messages":[{"role":"user","content":"bonjour"}],"log":[{"seq":1,"ts":0,"delta":{"user":"bonjour"}}],"seq":1,"ctx_used":42,"proj_snap":{"at":7,"state":{"mem":{"a":"1","b":"2"}},"key":"k","sys_hash":""}}"# diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 38ab5d5..2f6a47e 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -42,6 +42,8 @@ pub struct FakeGo { pub recall: Mutex>, /// Discussion active annoncée et discussions supprimées. pub conv_log: Mutex>, + /// Tours : débuts, vues et fins demandés. + pub turn_log: Mutex>, /// Adresse d'écoute du faux Go (pour un client `Store` direct). pub backend: Mutex>, } @@ -191,6 +193,34 @@ async fn handle(st: Arc, req: Request) -> Response .push(format!("{what} {}", v["id"].as_str().unwrap_or_default())); resp(204, "") } + "/internal/v1/turn/start" => { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); + let msg = v["message"].as_str().unwrap_or_default().to_string(); + st.turn_log.lock().unwrap().push(format!("start {msg}")); + if msg.trim().is_empty() { + return resp(400, r#"{"error":"message vide"}"#); + } + let caps = serde_json::json!({"Agent": true, "Mem": "off", "Code": false, "Role": ""}); + let out = serde_json::json!({ + "files": [], "caps": caps, "turn_caps": caps, + "content": msg, "model": "m1", "code_hint": false, + }); + resp(200, out.to_string()) + } + "/internal/v1/turn/view" => { + st.turn_log.lock().unwrap().push("view".into()); + let out = serde_json::json!({ + "head": [{"role": "system", "content": "Tu es Loki."}], + "prefix": "PRÉAMBULE", + "tools": [{"type": "function", "function": {"name": "bash", "description": "", "parameters": {"type": "object"}}}], + }); + resp(200, out.to_string()) + } + "/internal/v1/turn/done" => { + st.turn_log.lock().unwrap().push("done".into()); + resp(204, "") + } "/internal/v1/engine/stop" => { st.lease_log.lock().unwrap().push("stop".into()); *st.go_busy.lock().unwrap() = None; diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index b60a334..e1d7204 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -443,10 +443,21 @@ use loki_rs::chat::wire::{ChatEndpoint, EchoRefusals, EffortFallbacks, TurnReaso async fn fake_engine( replies: Vec, ) -> (String, Arc>>) { + fake_engine_gated(replies, None).await +} + +type Seen = Arc>>; + +/// Comme [`fake_engine`] ; avec `gate`, chaque complétion attend un jeton. +/// `/health` répond toujours 200. +async fn fake_engine_gated( + replies: Vec, + gate: Option>, +) -> (String, Seen) { use http_body_util::{BodyExt, Full}; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); - let seen = Arc::new(StdMutex::new(Vec::new())); + let seen: Seen = Arc::new(StdMutex::new(Vec::new())); let replies = Arc::new(StdMutex::new(std::collections::VecDeque::from(replies))); let s = seen.clone(); tokio::spawn(async move { @@ -454,12 +465,20 @@ async fn fake_engine( let Ok((stream, _)) = listener.accept().await else { return; }; - let (s, replies) = (s.clone(), replies.clone()); + let (s, replies, gate) = (s.clone(), replies.clone(), gate.clone()); tokio::spawn(async move { let svc = hyper::service::service_fn( move |req: hyper::Request| { - let (s, replies) = (s.clone(), replies.clone()); + let (s, replies, gate) = (s.clone(), replies.clone(), gate.clone()); async move { + let ok = |b: String| { + Ok::<_, std::convert::Infallible>(hyper::Response::new(Full::new( + bytes::Bytes::from(b), + ))) + }; + if req.uri().path() == "/health" { + return ok("ok".into()); + } let auth = req .headers() .get("authorization") @@ -469,10 +488,11 @@ async fn fake_engine( s.lock() .unwrap() .push((auth, serde_json::from_slice(&body).unwrap())); + if let Some(g) = &gate { + g.acquire().await.unwrap().forget(); + } let out = replies.lock().unwrap().pop_front().unwrap_or_default(); - Ok::<_, std::convert::Infallible>(hyper::Response::new(Full::new( - bytes::Bytes::from(out), - ))) + ok(out) } }, ); @@ -647,3 +667,208 @@ async fn crochets_go_des_discussions() { ] ); } + +// ---- un tour de bout en bout, par le service ---- + +use loki_rs::chat::generate::{ChatService, ERR_MODEL_LOADING, Refused, Sent}; +use loki_rs::chat::sessions::GoSessionHooks; + +fn port_of(url: &str) -> String { + url.trim_start_matches("http://127.0.0.1:") + .split('/') + .next() + .unwrap() + .to_string() +} + +async fn service( + engine_url: Option<&str>, +) -> ( + Arc, + Arc, + tokio::sync::oneshot::Sender<()>, + tempfile::TempDir, +) { + let (st, go, stop, t) = store().await; + go.set( + "config", + "PORT", + &engine_url.map(port_of).unwrap_or_else(|| "1".into()), + ); + go.set("state", "api_key", "cle-du-moteur"); + let sessions = Sessions::new(st.clone(), Conversation::new(), GoSessionHooks(st.clone())); + sessions.load().await.unwrap(); + let shared = ChatShared::for_sessions(&sessions, st, t.path().join("chatimg")); + ( + ChatService::new(Arc::new(sessions), Arc::new(shared)), + go, + stop, + t, + ) +} + +async fn until_done(svc: &ChatService, turns: usize) { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let c = svc.sessions.conv(); + let done = c.with_state(|st| { + st.log + .iter() + .filter(|e| e.delta.contains_key("turn_done")) + .count() + }); + if done >= turns && !c.is_generating() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("tour jamais terminé"); +} + +fn deltas(svc: &ChatService, key: &str) -> Vec { + svc.sessions.conv().with_state(|st| { + st.log + .iter() + .filter_map(|e| e.delta.get(key).cloned()) + .collect() + }) +} + +fn tool_then_answer(answer: &str) -> Vec { + vec![ + sse(&[ + serde_json::json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","type":"function","function":{"name":"bash","arguments":"{\"command\":\"ls\"}"}}]}}]}), + serde_json::json!({"choices":[{"delta":{},"finish_reason":"tool_calls"}]}), + ]), + sse(&[ + serde_json::json!({"choices":[{"delta":{"content": answer}}]}), + serde_json::json!({"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":80,"completion_tokens":3}}), + ]), + ] +} + +#[tokio::test] +async fn tour_de_bout_en_bout() { + let (url, seen) = fake_engine(tool_then_answer("fini")).await; + let (svc, go, _stop, _t) = service(Some(&url)).await; + let sent = svc + .send(serde_json::json!({"message": "liste", "cid": "a1"})) + .await + .unwrap(); + assert_eq!(sent, Sent::default()); + until_done(&svc, 1).await; + // La fin est signalée en tout dernier (après l'enregistrement) : on l'attend. + until(|| go.turn_log.lock().unwrap().contains(&"done".to_string())).await; + + // Vue modèle rangée : la demande, l'appel, son résultat, la réponse. + let msgs = svc.sessions.conv().with_state(|st| st.messages.clone()); + let roles: Vec<&str> = msgs.iter().map(|m| m.role.as_str()).collect(); + assert_eq!(roles, ["user", "assistant", "tool", "assistant"]); + assert_eq!(msgs[3].content, Some(serde_json::json!("fini"))); + // Journal : bulle avec le modèle, outil, réponse, fin. + assert_eq!(deltas(&svc, "user"), [serde_json::json!("liste")]); + assert_eq!(deltas(&svc, "model"), [serde_json::json!("m1")]); + assert!( + deltas(&svc, "tool_used") + .iter() + .any(|t| t["done"] == true && t["result"] == "exec:bash") + ); + assert_eq!(deltas(&svc, "content"), [serde_json::json!("fini")]); + assert!(!deltas(&svc, "stats").is_empty()); + // Requête : préambule fusionné au prompt du preset, clé du moteur, outils. + let seen = seen.lock().unwrap().clone(); + assert_eq!(seen[0].0, "Bearer cle-du-moteur"); + assert_eq!( + seen[0].1["messages"][0]["content"], + "PRÉAMBULE\n\nTu es Loki." + ); + assert_eq!(seen[0].1["messages"][1]["content"], "liste"); + assert_eq!(seen[0].1["tools"][0]["function"]["name"], "bash"); + // Bail rendu, fin signalée, discussion enregistrée. + assert_eq!(*go.lease.lock().unwrap(), None); + let id = svc.sessions.conv().id(); + 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. + let again = svc + .send(serde_json::json!({"message": "liste", "cid": "a1"})) + .await + .unwrap(); + assert!(again.dup); +} + +#[tokio::test] +async fn message_en_file_injecte_dans_le_tour() { + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + let (url, seen) = fake_engine_gated(tool_then_answer("les deux"), Some(gate.clone())).await; + let (svc, _go, _stop, _t) = service(Some(&url)).await; + svc.send(serde_json::json!({"message": "premier", "cid": "p1"})) + .await + .unwrap(); + until(|| seen.lock().unwrap().len() == 1).await; + let q = svc + .send(serde_json::json!({"message": "deuxième", "cid": "p2"})) + .await + .unwrap(); + assert!(q.queued, "un tour tourne : en file"); + gate.add_permits(10); + until_done(&svc, 1).await; + assert_eq!( + deltas(&svc, "user"), + [serde_json::json!("premier"), serde_json::json!("deuxième")] + ); + let second = seen.lock().unwrap()[1].1.clone(); + let last = second["messages"] + .as_array() + .unwrap() + .last() + .unwrap() + .clone(); + assert_eq!(last["content"], "deuxième", "injecté entre deux étapes"); + let users = svc + .sessions + .conv() + .with_state(|st| st.messages.iter().filter(|m| m.role == "user").count()); + assert_eq!(users, 2); +} + +#[tokio::test] +async fn moteur_tenu_par_une_tache_planifiee() { + let (url, _seen) = fake_engine(vec![]).await; + let (svc, go, _stop, _t) = service(Some(&url)).await; + *go.go_busy.lock().unwrap() = Some("la tâche planifiée « veille » occupe le modèle".into()); + let body = serde_json::json!({"message": "salut", "cid": "b1"}); + let want = Refused { + status: 409, + error: "la tâche planifiée « veille » occupe le modèle".into(), + }; + assert_eq!(svc.send(body.clone()).await, Err(want.clone())); + // Refusé : le même envoi peut être renvoyé (pas pris pour un doublon). + assert_eq!(svc.send(body).await, Err(want)); + assert!(!svc.sessions.conv().is_generating()); + assert!(deltas(&svc, "user").is_empty()); + // Le stop du chat arrête le détenteur Go. + svc.stop().await; + assert!(go.lease_log.lock().unwrap().contains(&"stop".to_string())); +} + +#[tokio::test] +async fn moteur_pas_pret_et_message_vide() { + let (svc, _go, _stop, _t) = service(None).await; + let r = svc + .send(serde_json::json!({"message": "salut", "cid": "c1"})) + .await; + assert_eq!( + r, + Err(Refused { + status: 503, + error: ERR_MODEL_LOADING.into() + }) + ); + let r = svc + .send(serde_json::json!({"message": " ", "cid": "c2"})) + .await; + assert_eq!(r.unwrap_err().status, 400); +}