diff --git a/rust/loki-rs/src/chat/compact.rs b/rust/loki-rs/src/chat/compact.rs index f3653f6..6de4f52 100644 --- a/rust/loki-rs/src/chat/compact.rs +++ b/rust/loki-rs/src/chat/compact.rs @@ -644,7 +644,12 @@ pub struct RecallEntry { pub trait RecallArchive { /// Range `content` VERBATIM et rend son id (« r7 »). Une erreur (base /// absente…) laisse le bloc non archivé : repli transparent. - fn archive(&self, label: &str, role: &str, content: &str) -> Result; + fn archive( + &self, + label: &str, + role: &str, + content: &str, + ) -> impl Future> + Send; } /// Archive absente : rien n'est jamais archivé. @@ -652,7 +657,7 @@ pub trait RecallArchive { pub struct NoRecall; impl RecallArchive for NoRecall { - fn archive(&self, _label: &str, _role: &str, _content: &str) -> Result { + async fn archive(&self, _label: &str, _role: &str, _content: &str) -> Result { Err("archive recall indisponible".into()) } } @@ -1587,7 +1592,7 @@ impl Compactor<'_, S, R> { let content = msg_text(m); let tool = names.get(&m.tool_call_id).map_or("", String::as_str); let label = recall_label(&m.role, tool, &content); - let Ok(id) = self.recall.archive(&label, &m.role, &content) else { + let Ok(id) = self.recall.archive(&label, &m.role, &content).await else { continue; }; let e = RecallEntry { id, label }; @@ -2011,7 +2016,7 @@ mod tests { } impl RecallArchive for MemRecall { - fn archive(&self, label: &str, role: &str, content: &str) -> Result { + async fn archive(&self, label: &str, role: &str, content: &str) -> Result { let mut b = self.blocks.lock().unwrap(); b.push((label.into(), role.into(), content.into())); Ok(format!("r{}", b.len())) diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs new file mode 100644 index 0000000..a2919e8 --- /dev/null +++ b/rust/loki-rs/src/chat/host.rs @@ -0,0 +1,580 @@ +//! Le tour branché sur le vrai monde : le [`TurnHost`] de la façade. +//! +//! - envoi au moteur par [`EngineHttp`] (clé de CET endpoint), pannes +//! classées et formulées comme `friendlyLLMError` ; +//! - outils exécutés par le Go (`/internal/v1/tool/exec`), avec ce qu'il tire +//! du résultat (capture de la bulle, aperçu du navigateur, image relayée) ; +//! - résultats complets (« voir plus ») confiés à l'écrivain des discussions ; +//! - compaction par [`Compactor`] : résumé demandé au moteur +//! ([`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`. +//! +//! É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 +//! compaction résume la transcription. Ni conseil de cache ni garde +//! d'écriture du mode Code (état du Go) : `cache_hint` et `preflight` absents. + +use std::collections::{HashMap, VecDeque}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +use futures_util::StreamExt; +use serde_json::{Map, Value, json}; + +use super::compact::{ + CompactCaps, CompactConfig, CompactMemory, CompactOpts, Compactor, ContinuationRequest, + RecallArchive, Summarizer, SummaryChoice, SummaryError, TranscriptRequest, compact_opts_mid, + parse_summary_response, +}; +use super::engine::{EngineHttp, classify, friendly_error}; +use super::persist::{Persister, StoreSink, ToolResJob}; +use super::stream::{CacheHintFn, PreflightFn, RetryPatterns}; +use super::turn::{CompactPhase, Reply, SendError, ToolOutcome, TurnHost}; +use super::types::{AskEvent, DiffLine, Message, Tool}; +use super::wire::{ + ChatEndpoint, ChatPayloadOpts, EchoRefusals, EffortFallbacks, IMG_REF_SCHEME, + build_chat_payload, content_parts, part_image_url, wire_messages, +}; +use crate::store::Store; + +/// Taille au-delà de laquelle le cache d'images oublie les plus anciennes +/// (`imgCacheMax`). +pub const IMG_CACHE_MAX: usize = 64 << 20; +/// Corps d'un refus de résumé gardé au plus (`postSummary`). +const SUMMARY_ERR_MAX: usize = 500; + +/// Base64 standard (avec remplissage), comme `base64.StdEncoding`. +pub fn b64(data: &[u8]) -> String { + const T: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = String::with_capacity(data.len().div_ceil(3) * 4); + for c in data.chunks(3) { + let n = (c[0] as u32) << 16 + | (*c.get(1).unwrap_or(&0) as u32) << 8 + | *c.get(2).unwrap_or(&0) as u32; + out.push(T[(n >> 18) as usize & 63] as char); + out.push(T[(n >> 12) as usize & 63] as char); + out.push(if c.len() > 1 { + T[(n >> 6) as usize & 63] as char + } else { + '=' + }); + out.push(if c.len() > 2 { + T[n as usize & 63] as char + } else { + '=' + }); + } + out +} + +/// Type MIME d'une image rangée, d'après son extension (`imgExtByMime`). +fn image_mime(name: &str) -> &'static str { + for (ext, mime) in [ + (".png", "image/png"), + (".gif", "image/gif"), + (".webp", "image/webp"), + ] { + if name.ends_with(ext) { + return mime; + } + } + "image/jpeg" +} + +/// Nom d'une image rangée, si la référence est sûre (ni chemin ni `..`). +fn image_name(r: &str) -> Option<&str> { + let name = r.strip_prefix(IMG_REF_SCHEME)?; + (!name.is_empty() && !name.contains(['/', '\\']) && !name.contains("..")).then_some(name) +} + +#[derive(Default)] +struct ImgCacheInner { + map: HashMap, + order: VecDeque, + bytes: usize, +} + +/// Images jointes en data-URL, bornées à [`IMG_CACHE_MAX`] (la plus ancienne +/// part en premier, jamais la dernière). +pub struct ImageCache { + dir: PathBuf, + inner: Mutex, +} + +impl ImageCache { + /// `dir` : `LOKI_HOME/chatimg`. + pub fn new(dir: PathBuf) -> Self { + ImageCache { + dir, + inner: Mutex::new(ImgCacheInner::default()), + } + } + + pub fn get(&self, r: &str) -> Option { + let name = image_name(r)?; + self.inner.lock().ok()?.map.get(name).cloned() + } + + fn put(&self, name: &str, url: String) { + let Ok(mut g) = self.inner.lock() else { return }; + if g.map.contains_key(name) { + return; + } + g.bytes += url.len(); + g.map.insert(name.to_string(), url); + g.order.push_back(name.to_string()); + while g.bytes > IMG_CACHE_MAX && g.order.len() > 1 { + if let Some(old) = g.order.pop_front() + && let Some(u) = g.map.remove(&old) + { + g.bytes -= u.len(); + } + } + } + + /// Lit et ouvre (par le Go) les images référencées pas encore en cache. + /// Une image illisible reste absente : la requête dira « indisponible ». + pub async fn prefetch(&self, store: &Store, msgs: &[Message]) { + for m in msgs { + for p in content_parts(m.content.as_ref()) { + let Some(name) = image_name(part_image_url(p)) else { + continue; + }; + if self.inner.lock().is_ok_and(|g| g.map.contains_key(name)) { + continue; + } + let Ok(raw) = tokio::fs::read(self.dir.join(name)).await else { + continue; + }; + let Ok(b) = store.open(raw).await else { + continue; + }; + self.put( + name, + format!("data:{};base64,{}", image_mime(name), b64(&b)), + ); + } + } + } +} + +/// Le moteur vu de la compaction (`postSummary`). +pub struct EngineSummarizer { + pub http: EngineHttp, + pub ep: ChatEndpoint, + pub cfg: HashMap, + pub patterns: Arc, + pub port: u16, +} + +impl EngineSummarizer { + async fn post(&self, payload: Value) -> Result { + let body = serde_json::to_vec(&payload).unwrap_or_default(); + let resp = self + .http + .post(&self.ep.url, &self.ep.key, body) + .await + .map_err(|e| { + SummaryError::Other(friendly_error(classify(&e), &e.to_string(), self.port, "")) + })?; + let status = resp.status().as_u16(); + let raw = resp + .bytes() + .await + .map_err(|e| SummaryError::Other(e.to_string()))?; + if status != 200 { + let b = &raw[..raw.len().min(SUMMARY_ERR_MAX)]; + return Err(SummaryError::Http { + who: if self.ep.external { + "API externe" + } else { + "llama-server" + } + .into(), + status, + body: String::from_utf8_lossy(b).trim().to_string(), + }); + } + parse_summary_response(&raw) + } +} + +impl Summarizer for EngineSummarizer { + async fn summarize_transcript( + &self, + req: TranscriptRequest, + ) -> Result { + let mut p = json!({ + "model": self.ep.model, + "messages": [ + {"role": "system", "content": req.system}, + {"role": "user", "content": req.transcript}, + ], + "stream": false, + "temperature": 0.2, + "max_tokens": req.max_tokens, + "chat_template_kwargs": {"enable_thinking": false}, + }); + // Propre à llama.cpp : une API distante stricte refuse l'argument. + if self.ep.external + && let Some(o) = p.as_object_mut() + { + o.remove("chat_template_kwargs"); + } + self.post(p).await + } + + async fn summarize_continuation( + &self, + req: ContinuationRequest, + ) -> Result { + let mut p = build_chat_payload( + &self.ep, + &req.messages, + 0.2, + &ChatPayloadOpts { + tools: &req.tools, + disable_tools: false, + tool_choice_none: true, + effort: &req.effort, + kwargs: Some(&req.kwargs), + no_sampling: true, + }, + &self.cfg, + &|_| None, + ); + if let Some(o) = p.as_object_mut() { + o.insert("stream".into(), Value::Bool(false)); + o.remove("stream_options"); + o.insert("max_tokens".into(), json!(req.max_tokens)); + } + self.post(p).await + } + + fn slot_holds(&self, _conv: &str) -> bool { + false + } + + fn textual_tool_call(&self, content: &str) -> bool { + self.patterns.textual_tool_call(content) + } +} + +/// Archive recall tenue par le Go. +pub struct StoreRecall(pub Store); + +impl RecallArchive for StoreRecall { + async fn archive(&self, label: &str, role: &str, content: &str) -> Result { + self.0.recall_archive(label, role, content).await + } +} + +/// Ce qui survit d'un tour à l'autre (un exemplaire par processus). +pub struct ChatShared { + pub http: EngineHttp, + pub store: Store, + pub persister: Persister, + pub effort: EffortFallbacks, + pub echo_refusals: EchoRefusals, + pub compact_memory: CompactMemory, + pub images: ImageCache, + pub patterns: Arc, +} + +/// Ce que la compaction sait du tour. +pub struct TurnCompact { + pub cfg: CompactConfig, + pub caps: CompactCaps, + /// Mode Code : critères rendus de la discussion. + pub criteria: String, + /// Discussion du tour (vide : tâche ou sous-agent). + pub main_conv: String, + /// Forme de la requête du tour, pour les options de la compaction en + /// cours de tour. + pub tools: Vec, + pub effort: String, + pub kwargs: Map, +} + +/// Le [`TurnHost`] de la façade. +pub struct GoTurnHost { + pub shared: Arc, + /// Discussion dont relèvent les résultats complets. + pub conv_id: String, + /// Capacités du tour, telles que l'API d'outils les attend (champs Go). + pub caps: Value, + pub cfg: HashMap, + pub port: u16, + pub summarizer: EngineSummarizer, + pub compact: TurnCompact, + /// Messages mis en file par l'utilisateur pendant le tour. + pub queued: Box Vec + Send + Sync>, +} + +/// `toolResID` : « .<24 hex> ». +fn tool_res_id(conv: &str) -> String { + let mut b = [0u8; 12]; + let _ = getrandom::fill(&mut b); + let conv = match conv.trim() { + "" => "nosession", + c => c, + }; + format!("{conv}.{}", hex::encode(b)) +} + +/// Réponse de `/internal/v1/tool/exec` → [`ToolOutcome`]. +pub fn tool_outcome(v: &Value, label: &str) -> ToolOutcome { + let s = |k: &str| v.get(k).and_then(Value::as_str).unwrap_or("").to_string(); + let n = |k: &str| v.get(k).and_then(Value::as_i64).unwrap_or(0); + let mut relay: Option = v + .get("relay") + .and_then(|r| serde_json::from_value(r.clone()).ok()); + if relay.is_none() + && let Some(img) = v.get("vision_img").filter(|i| i.is_object()) + { + relay = Some(see_image_message(label, img.clone())); + } + ToolOutcome { + result: s("result"), + diff: v + .get("diff") + .and_then(|d| serde_json::from_value::>(d.clone()).ok()) + .unwrap_or_default(), + added: n("added"), + removed: n("removed"), + ask: v + .get("ask") + .and_then(|a| serde_json::from_value::(a.clone()).ok()), + image: s("image"), + live_shot: s("shot"), + relay, + } +} + +/// `seeImageMessage` : l'image relayée au modèle. +pub fn see_image_message(label: &str, img: Value) -> Message { + let text = if label.is_empty() { + "Image :".to_string() + } else { + format!("Image demandée ({label}) :") + }; + Message { + role: "user".into(), + content: Some(json!([{"type": "text", "text": text}, img])), + ..Default::default() + } +} + +impl TurnHost for GoTurnHost { + async fn send(&self, ep: &ChatEndpoint, body: Vec) -> Result { + match self.shared.http.post(&ep.url, &ep.key, body).await { + Ok(resp) => { + let status = resp.status().as_u16(); + let body = resp.bytes_stream().map(|r| r.map_err(|e| e.to_string())); + Ok(Reply { + status, + body: Box::pin(body), + }) + } + Err(e) => { + let kind = classify(&e); + Err(SendError { + kind, + message: friendly_error(kind, &e.to_string(), self.port, ""), + }) + } + } + } + + async fn exec_tool(&self, name: &str, args: &Map, label: &str) -> ToolOutcome { + match self + .shared + .store + .tool_exec(name, &Value::Object(args.clone()), &self.caps) + .await + { + Ok(v) => tool_outcome(&v, label), + Err(e) => ToolOutcome { + result: format!("[erreur] exécution de l'outil impossible : {e}"), + ..Default::default() + }, + } + } + + fn save_tool_result(&self, result: &str) -> String { + let id = tool_res_id(&self.conv_id); + self.shared.persister.enqueue_tool_res(ToolResJob { + key: id.clone(), + plain: Arc::from(result), + }); + id + } + + async fn compact(&self, phase: CompactPhase, msgs: &[Message]) -> Option> { + let recall = StoreRecall(self.shared.store.clone()); + let c = Compactor { + cfg: &self.compact.cfg, + summarizer: &self.summarizer, + recall: &recall, + memory: &self.shared.compact_memory, + criteria: &self.compact.criteria, + }; + let (out, changed) = match phase { + CompactPhase::Reactive | CompactPhase::FullWindow => { + c.compact_messages(msgs, self.compact.caps).await + } + CompactPhase::Mid { used, rescue } => { + let conv = self.compact.main_conv.as_str(); + let opt: CompactOpts = compact_opts_mid( + &self.compact.cfg, + &self.shared.compact_memory, + self.summarizer.slot_holds(conv), + self.summarizer.ep.external, + conv, + msgs, + used, + self.compact.tools.clone(), + &self.compact.effort, + self.compact.kwargs.clone(), + rescue, + |m| { + wire_messages(m, false, &self.compact.cfg.echo) + .0 + .into_owned() + }, + ); + c.compact_messages_noted(msgs, self.compact.caps, &opt) + .await + } + }; + changed.then_some(out) + } + + fn compact_refused(&self, used: i64, msgs: &[Message]) -> bool { + self.shared.compact_memory.refused_skip( + &self.compact.cfg, + &self.compact.main_conv, + used, + msgs, + ) + } + + fn drain_queued(&self) -> Vec { + (self.queued)() + } + + async fn prepare_images(&self, msgs: &[Message]) { + self.shared.images.prefetch(&self.shared.store, msgs).await; + } + + fn image_data_url(&self, r: &str) -> Option { + self.shared.images.get(r) + } + + fn effort_remember(&self, want: &str, got: &str) { + self.shared.effort.remember(&self.cfg, want, got); + } + + fn echo_refuse(&self, model: &str) { + self.shared.echo_refusals.refuse(model); + } + + fn cache_hint(&self) -> Option { + None + } + + fn preflight(&self) -> Option { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn base64_standard() { + assert_eq!(b64(b""), ""); + assert_eq!(b64(b"f"), "Zg=="); + assert_eq!(b64(b"fo"), "Zm8="); + assert_eq!(b64(b"foo"), "Zm9v"); + assert_eq!(b64(b"foobar"), "Zm9vYmFy"); + assert_eq!(b64(&[0xff, 0xfe, 0x00]), "//4A"); + } + + #[test] + fn references_d_images() { + assert_eq!(image_name("loki-img:ab.png"), Some("ab.png")); + for bad in [ + "loki-img:", + "loki-img:../x", + "loki-img:a/b", + "loki-img:a\\b", + "data:x", + "ab.png", + ] { + assert_eq!(image_name(bad), None, "{bad}"); + } + assert_eq!(image_mime("a.png"), "image/png"); + assert_eq!(image_mime("a.webp"), "image/webp"); + assert_eq!(image_mime("a.jpg"), "image/jpeg"); + assert_eq!(image_mime("a.img"), "image/jpeg"); + } + + #[test] + fn cache_d_images_borne() { + let c = ImageCache::new(PathBuf::from("/nulle-part")); + let big = "x".repeat(IMG_CACHE_MAX / 2 + 1); + c.put("a.png", big.clone()); + c.put("b.png", big.clone()); + assert_eq!(c.get("loki-img:a.png"), None, "la plus ancienne part"); + assert_eq!(c.get("loki-img:b.png").map(|u| u.len()), Some(big.len())); + // Une seule image, même trop grosse, reste. + let c = ImageCache::new(PathBuf::from("/nulle-part")); + c.put("g.png", "y".repeat(IMG_CACHE_MAX + 1)); + assert!(c.get("loki-img:g.png").is_some()); + } + + #[test] + fn reponse_de_l_api_d_outils() { + let v = json!({ + "label": "ls", "result": "a\nb", + "diff": [{"op": "+", "text": "x"}], "added": 1, "removed": 2, + "ask": {"question": "On continue ?", "options": ["oui"]}, + "image": "captures/a.png", "shot": "s1", + "relay": {"role": "user", "content": [{"type": "text", "text": "Voici la capture demandée."}]} + }); + let o = tool_outcome(&v, "ls"); + assert_eq!(o.result, "a\nb"); + assert_eq!((o.added, o.removed, o.diff.len()), (1, 2, 1)); + assert_eq!(o.ask.unwrap().question, "On continue ?"); + assert_eq!( + (o.image.as_str(), o.live_shot.as_str()), + ("captures/a.png", "s1") + ); + assert_eq!(o.relay.unwrap().role, "user"); + // vision_img seul : le message relayé est bâti ici (seeImageMessage). + let img = json!({"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}}); + let o = tool_outcome(&json!({"result": "ok", "vision_img": img}), "photo.png"); + assert_eq!( + o.relay.unwrap().content, + Some(json!([{"type": "text", "text": "Image demandée (photo.png) :"}, img])) + ); + assert_eq!( + see_image_message("", json!({})).content.unwrap()[0]["text"], + "Image :" + ); + assert_eq!(tool_outcome(&json!({}), "").result, ""); + } + + #[test] + fn identifiant_d_un_resultat_complet() { + let id = tool_res_id("c42"); + let (conv, rand) = id.split_once('.').unwrap(); + assert_eq!((conv, rand.len()), ("c42", 24)); + assert!(tool_res_id(" ").starts_with("nosession.")); + assert_ne!(tool_res_id("c"), tool_res_id("c")); + } +} diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs index 11bd048..66e54ff 100644 --- a/rust/loki-rs/src/chat/mod.rs +++ b/rust/loki-rs/src/chat/mod.rs @@ -14,11 +14,13 @@ //! - [`sessions`] : plusieurs discussions (index, bascules, chargement) ; //! - [`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). +//! - [`turn`] : la boucle d'un tour (complétions, outils, filets) ; +//! - [`host`] : le tour branché sur le moteur, les outils du Go et la base. pub mod compact; pub mod conv; pub mod engine; +pub mod host; pub mod journal; pub mod lease; pub mod nudges; diff --git a/rust/loki-rs/src/chat/turn.rs b/rust/loki-rs/src/chat/turn.rs index 1857e8e..9445c6a 100644 --- a/rust/loki-rs/src/chat/turn.rs +++ b/rust/loki-rs/src/chat/turn.rs @@ -144,8 +144,18 @@ pub trait TurnHost: Send + Sync { phase: CompactPhase, msgs: &[Message], ) -> impl Future>> + Send; + /// Compaction en cours de tour déjà refusée pour cet état + /// (`compactRefusedSkip`) : on ne la retente pas. + fn compact_refused(&self, _used: i64, _msgs: &[Message]) -> bool { + false + } /// Messages mis en file par l'utilisateur, à injecter maintenant. fn drain_queued(&self) -> Vec; + /// Prépare ce que [`Self::image_data_url`] devra rendre pour ces messages + /// (lecture et déchiffrement des images rangées), avant chaque requête. + fn prepare_images(&self, _msgs: &[Message]) -> impl Future + Send { + async {} + } /// Image rangée (`loki-img:…`) → data-URL. fn image_data_url(&self, r: &str) -> Option; /// Intensité refusée par le gabarit : traduction retenue. @@ -377,6 +387,7 @@ pub async fn run_turn( } let calls_on = !tc.tools.is_empty() && !s.disable_tools && !s.tool_choice_none; let pol = turn_echo_policy(ep.external || s.echo_off_turn, || tc.echo.clone()); + host.prepare_images(&messages).await; let (body, echo_sent) = { let (sent, echo_sent) = wire_messages(&messages, s.tool_choice_none, &pol); let payload = build_chat_payload( @@ -811,7 +822,9 @@ pub async fn run_turn( used = base + estimate_tokens(&messages[step_start..], &ccfg.echo); s.last_est = used; } - if compact_needed(ccfg, &messages, used, s.peak_gen) { + if compact_needed(ccfg, &messages, used, s.peak_gen) + && !host.compact_refused(used, &messages) + { cb(StreamEvent::Compacting(true)); let phase = CompactPhase::Mid { used, diff --git a/rust/loki-rs/src/store.rs b/rust/loki-rs/src/store.rs index d9c466a..a896ddd 100644 --- a/rust/loki-rs/src/store.rs +++ b/rust/loki-rs/src/store.rs @@ -169,6 +169,28 @@ impl Store { .await } + /// Archive un bloc verbatim pour l'outil recall (`archiveRecallBlock`, + /// numéroté par le Go) et rend son id (« r7 »). + pub async fn recall_archive( + &self, + label: &str, + role: &str, + content: &str, + ) -> Result { + let v = self + .internal_json( + Method::POST, + "/internal/v1/recall/archive", + Some(&serde_json::json!({ "label": label, "role": role, "content": content })), + ) + .await?; + v.get("id") + .and_then(|i| i.as_str()) + .filter(|i| !i.is_empty()) + .map(str::to_string) + .ok_or_else(|| format!("archive recall : réponse sans id {v}")) + } + /// Valeur brute ; `Ok(None)` = absente, `Err` = pas pu lire (à distinguer /// pour les secrets : comme `getBytesErr`). pub async fn get(&self, bucket: &str, key: &str) -> Result>, String> { diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 9876e97..01c001d 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -37,6 +37,9 @@ pub struct FakeGo { pub lease_stop: Mutex, pub go_busy: Mutex>, pub lease_log: Mutex>, + /// Résultat imposé à tout outil (sinon « exec: »), et archive recall. + pub tool_result: Mutex>, + pub recall: Mutex>, /// Adresse d'écoute du faux Go (pour un client `Store` direct). pub backend: Mutex>, } @@ -166,6 +169,16 @@ async fn handle(st: Arc, req: Request) -> Response let stop = *st.lease_stop.lock().unwrap(); resp(200, serde_json::json!({"stop": stop}).to_string()) } + "/internal/v1/recall/archive" => { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); + let mut r = st.recall.lock().unwrap(); + r.push(v["content"].as_str().unwrap_or_default().to_string()); + resp( + 200, + serde_json::json!({"id": format!("r{}", r.len())}).to_string(), + ) + } "/internal/v1/engine/stop" => { st.lease_log.lock().unwrap().push("stop".into()); *st.go_busy.lock().unwrap() = None; @@ -174,9 +187,15 @@ async fn handle(st: Arc, req: Request) -> Response "/internal/v1/tool/exec" => { let body = req.into_body().collect().await.unwrap().to_bytes(); let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let result = st + .tool_result + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| format!("exec:{}", v["name"].as_str().unwrap())); let out = serde_json::json!({ "label": v["args"]["command"], - "result": format!("exec:{}", v["name"].as_str().unwrap()), + "result": result, "caps_agent": v["caps"]["Agent"], }); resp(200, out.to_string()) diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index f1b9dca..7f73513 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -415,3 +415,198 @@ async fn supprimer_l_ouverte_puis_la_derniere() { assert_eq!(idx[0].id, c); assert_eq!(s.delete("inconnue").await, Err(SessionError::NotFound)); } + +// ---- un tour complet, branché ---- + +use loki_rs::chat::compact::{CompactCaps, CompactConfig, CompactMemory}; +use loki_rs::chat::engine::EngineHttp; +use loki_rs::chat::host::{ChatShared, EngineSummarizer, GoTurnHost, ImageCache, TurnCompact}; +use loki_rs::chat::stream::RetryPatterns; +use loki_rs::chat::turn::{Cancel, TurnConfig, run_turn}; +use loki_rs::chat::types::{Message, StreamEvent, Tool, ToolFunction}; +use loki_rs::chat::wire::{ChatEndpoint, EchoRefusals, EffortFallbacks, TurnReasoning}; + +/// Faux moteur : rend les réponses SSE dans l'ordre, garde les requêtes. +async fn fake_engine( + replies: Vec, +) -> (String, Arc>>) { + 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 replies = Arc::new(StdMutex::new(std::collections::VecDeque::from(replies))); + let s = seen.clone(); + tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let (s, replies) = (s.clone(), replies.clone()); + tokio::spawn(async move { + let svc = hyper::service::service_fn( + move |req: hyper::Request| { + let (s, replies) = (s.clone(), replies.clone()); + async move { + let auth = req + .headers() + .get("authorization") + .map(|v| v.to_str().unwrap().to_string()) + .unwrap_or_default(); + let body = req.into_body().collect().await.unwrap().to_bytes(); + s.lock() + .unwrap() + .push((auth, serde_json::from_slice(&body).unwrap())); + let out = replies.lock().unwrap().pop_front().unwrap_or_default(); + Ok::<_, std::convert::Infallible>(hyper::Response::new(Full::new( + bytes::Bytes::from(out), + ))) + } + }, + ); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(hyper_util::rt::TokioIo::new(stream), svc) + .await; + }); + } + }); + (format!("http://{addr}/v1/chat/completions"), seen) +} + +fn sse(chunks: &[serde_json::Value]) -> String { + chunks + .iter() + .map(|c| format!("data: {c}\n\n")) + .collect::() + + "data: [DONE]\n\n" +} + +#[tokio::test] +async fn tour_complet_par_le_go() { + let (st, go, _stop, t) = store().await; + let long = "ligne\n".repeat(400); // > aperçu : rangé pour « voir plus » + *go.tool_result.lock().unwrap() = Some(long.clone()); + let imgdir = t.path().join("chatimg"); + std::fs::create_dir_all(&imgdir).unwrap(); + std::fs::write(imgdir.join("p.png"), b"SCELLE:PNG").unwrap(); + let (url, seen) = fake_engine(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":"Voilà."}}]}), + serde_json::json!({"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":50,"completion_tokens":2}}), + ]), + ]) + .await; + let ep = ChatEndpoint { + url, + model: "loki".into(), + key: "cle-locale".into(), + external: false, + }; + let shared = Arc::new(ChatShared { + http: EngineHttp::new(), + store: st.clone(), + persister: Persister::new(StoreSink::new(st.clone())), + effort: EffortFallbacks::new(), + echo_refusals: EchoRefusals::new(), + compact_memory: CompactMemory::new(), + images: ImageCache::new(imgdir), + patterns: RetryPatterns::builtin(), + }); + let cfg = std::collections::HashMap::new(); + let host = GoTurnHost { + shared: shared.clone(), + conv_id: "c7".into(), + caps: serde_json::json!({"Agent": true}), + cfg: cfg.clone(), + port: 8080, + summarizer: EngineSummarizer { + http: shared.http.clone(), + ep: ep.clone(), + cfg: cfg.clone(), + patterns: shared.patterns.clone(), + port: 8080, + }, + compact: TurnCompact { + cfg: CompactConfig::default(), + caps: CompactCaps::default(), + criteria: String::new(), + main_conv: "c7".into(), + tools: vec![], + effort: String::new(), + kwargs: Default::default(), + }, + queued: Box::new(Vec::new), + }; + let bash = Tool { + kind: "function".into(), + function: ToolFunction { + name: "bash".into(), + description: String::new(), + parameters: serde_json::json!({"type":"object"}), + }, + }; + let tc = TurnConfig { + cfg, + ep, + tools: vec![bash], + temperature: 0.7, + code: false, + reasoning: TurnReasoning::default(), + echo: Default::default(), + compact: CompactConfig::default(), + patterns: shared.patterns.clone(), + }; + let question = Message { + role: "user".into(), + content: Some(serde_json::json!([ + {"type": "text", "text": "Regarde"}, + {"type": "image_url", "image_url": {"url": "loki-img:p.png"}} + ])), + ..Default::default() + }; + let mut evs = Vec::new(); + let mut cb = |e: StreamEvent| { + evs.push(e); + true + }; + let r = run_turn(&host, &tc, vec![question], &Cancel::never(), &mut cb).await; + assert_eq!(r.err, None); + assert_eq!(r.extra.len(), 2); + + let seen = seen.lock().unwrap().clone(); + assert_eq!(seen.len(), 2); + assert_eq!(seen[0].0, "Bearer cle-locale"); + // Image jointe : lue sur disque, ouverte par le Go, envoyée en data-URL. + assert_eq!( + seen[0].1["messages"][0]["content"][1]["image_url"]["url"], + "data:image/png;base64,UE5H" + ); + // Le modèle reçoit le résultat (plafonné), l'UI un aperçu et un id. + assert_eq!(seen[1].1["messages"][2]["role"], "tool"); + let done = evs + .iter() + .find_map(|e| match e { + StreamEvent::ToolUsed(t) if t.done => Some(t.clone()), + _ => None, + }) + .unwrap(); + assert_eq!(done.result_chars, long.chars().count() as i64); + assert!(done.result.len() < long.len()); + assert!(done.result_id.starts_with("c7.")); + shared.persister.flush().await; + assert_eq!( + opened(&go, BK_TOOL_RES, &done.result_id).unwrap(), + long.as_bytes() + ); + let answer: String = evs + .iter() + .filter_map(|e| match e { + StreamEvent::Content(s) => Some(s.as_str()), + _ => None, + }) + .collect(); + assert_eq!(answer, "Voilà."); +}