diff --git a/rust/loki-rs/src/api.rs b/rust/loki-rs/src/api.rs index 7836ebc..f774f4b 100644 --- a/rust/loki-rs/src/api.rs +++ b/rust/loki-rs/src/api.rs @@ -24,6 +24,8 @@ pub struct Ctx { pub host_env: HostEnv, /// `$LOKI_HOME` et nom du service moteur (fichiers PID / journal). pub engine: engine::Paths, + /// Conversation tenue par la façade (`LOKI_CHAT=rust`) ; `None` : au Go. + pub chat: Option>, } impl Ctx { diff --git a/rust/loki-rs/src/chat/generate.rs b/rust/loki-rs/src/chat/generate.rs index bd67bea..79f6d40 100644 --- a/rust/loki-rs/src/chat/generate.rs +++ b/rust/loki-rs/src/chat/generate.rs @@ -154,6 +154,16 @@ fn obj(k: &str, v: impl Into) -> Delta { type BoxFut = Pin + Send>>; +/// Ce qu'un tour lit une fois au départ. +struct TurnEnv { + cfg: std::collections::HashMap, + ep: super::wire::ChatEndpoint, + echo: super::wire::EchoPolicy, + ccfg: CompactConfig, + port: u16, + summarizer: EngineSummarizer, +} + /// Le chat tenu par la façade : la discussion ouverte, ses tours et sa file. pub struct ChatService { pub sessions: Arc>, @@ -446,6 +456,142 @@ impl ChatService { Some(compacted) } + /// Configuration, point de chat et résumeur d'un tour (lus une fois). + async fn turn_env(&self) -> TurnEnv { + let store = &self.shared.store; + 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 echo = echo_policy_for(&cfg, &self.shared.echo_refusals, None); + let ccfg = CompactConfig::from_config(&cfg, echo.clone()); + let port = llm_port(&cfg); + let summarizer = EngineSummarizer { + http: self.shared.http.clone(), + ep: ep.clone(), + cfg: cfg.clone(), + patterns: self.shared.patterns.clone(), + port, + }; + TurnEnv { + cfg, + ep, + echo, + ccfg, + port, + summarizer, + } + } + + /// `/api/chat/reset` : fil vide pour tous les appareils, tour arrêté et + /// verrou rendu, file vidée, fichiers de la discussion effacés. + pub async fn reset(&self) { + let conv = self.sessions.conv(); + conv.reset(); + self.queue().items.clear(); + let body = json!({ "id": conv.id(), "files_only": true }); + if let Err(e) = self + .shared + .store + .internal_json( + hyper::Method::POST, + "/internal/v1/conv/dropped", + Some(&body), + ) + .await + { + eprintln!("[chat] fichiers de la discussion non effacés : {e}"); + } + self.sessions.persist().await; + } + + /// `/api/chat/compact` (`CompactNow`) : compaction manuelle, en tâche de + /// fond ; la progression passe par le journal. + pub async fn compact_now(self: &Arc) -> Result<(), Refused> { + if !self.engine_ready().await { + return Err(Refused { + status: 503, + error: ERR_MODEL_LOADING.into(), + }); + } + let conv = self.sessions.conv().clone(); + let epoch = conv.begin_turn(Box::new(|| {})).map_err(|_| Refused { + status: 409, + error: ERR_BUSY.into(), + })?; + 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, .. } => Refused { status: 409, error }, + LeaseError::Other(error) => Refused { status: 503, error }, + }); + } + }; + let me = self.clone(); + tokio::spawn(async move { + let env = me.turn_env().await; + let local = !env.ep.external; + let snap = conv.with_state_if(epoch, |st| { + let msgs = st.messages.clone(); + let used = if local { + ctx_pending(st.ctx_used, st.ctx_used_len, &msgs, &env.echo) + } else { + st.ctx_used + }; + (msgs, used) + }); + if let Some((msgs, used)) = snap { + let changed = me + .compact_and_publish( + &env.summarizer, + &env.ccfg, + CompactCaps::default(), + true, + epoch, + "manuel", + &msgs, + used, + &CompactOpts::default(), + ) + .await + .is_some(); + if changed { + me.sessions.persist().await; + } + } + conv.release_turn(epoch); + lease.release().await; + }); + Ok(()) + } + + /// `/api/chat/tool-result` (`loadToolResult`) : le résultat complet d'un + /// outil — en mémoire tant qu'il n'est pas écrit, sinon en base. + pub async fn tool_result(&self, id: &str) -> Option { + if id.contains(['/', '\\']) { + return None; + } + if let Some(s) = self.shared.persister.tool_res_pending(id) { + return Some(s.to_string()); + } + let b = self + .shared + .store + .get_open(super::persist::BK_TOOL_RES, id) + .await + .ok() + .flatten()?; + Some(String::from_utf8_lossy(&b).into_owned()) + } + + /// Un événement d'affichage né hors d'un tour (bascule du mode Code, + /// critères édités), sur le fil courant. + pub fn push_delta(&self, d: Delta) { + let conv = self.sessions.conv(); + conv.append_delta(conv.epoch(), d); + } + /// `generate` : un tour complet, détaché — seul l'arrêt l'interrompt. async fn generate( self: Arc, @@ -458,13 +604,15 @@ impl ChatService { 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 TurnEnv { + cfg, + ep, + echo, + ccfg, + port, + summarizer, + } = self.turn_env().await; 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"); @@ -472,13 +620,6 @@ impl ChatService { 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) @@ -611,6 +752,12 @@ impl ChatService { kwargs: reasoning.kwargs.clone().unwrap_or_default(), }, queued: Box::new(move || me.drain_queued(epoch)), + notice: { + let conv = conv.clone(); + Box::new(move |d| { + conv.append_delta(epoch, d); + }) + }, }; // Chaque événement devient un delta du journal. diff --git a/rust/loki-rs/src/chat/host.rs b/rust/loki-rs/src/chat/host.rs index 9228ea1..a8266c1 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -312,6 +312,9 @@ pub struct GoTurnHost { pub compact: TurnCompact, /// Messages mis en file par l'utilisateur pendant le tour. pub queued: Box Vec + Send + Sync>, + /// Événement d'affichage né d'un outil hors du flux du modèle (critères + /// changés par l'outil `criteria`), à journaliser. + pub notice: Box) + Send + Sync>, } /// `toolResID` : « .<24 hex> ». @@ -396,7 +399,19 @@ impl TurnHost for GoTurnHost { .tool_exec(name, &Value::Object(args.clone()), &self.caps) .await { - Ok(v) => tool_outcome(&v, label), + Ok(v) => { + if let Some(list) = v.get("criteria").filter(|l| l.is_array()) { + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0); + let mut d = Map::new(); + d.insert("criteria".into(), list.clone()); + d.insert("ts".into(), ts.into()); + (self.notice)(d); + } + tool_outcome(&v, label) + } Err(e) => ToolOutcome { result: format!("[erreur] exécution de l'outil impossible : {e}"), ..Default::default() diff --git a/rust/loki-rs/src/chatapi.rs b/rust/loki-rs/src/chatapi.rs new file mode 100644 index 0000000..d0a696e --- /dev/null +++ b/rust/loki-rs/src/chatapi.rs @@ -0,0 +1,312 @@ +//! Routes du chat servies par Rust quand la conversation est tenue par la +//! façade (`LOKI_CHAT=rust`) — port des handlers de `web_chat.go`. +//! +//! - `/api/chat` : flux d'abonnement SSE (rejeu puis direct, battement de +//! cœur toutes les 4 s, trames regroupées) ; +//! - `/api/chat/send|stop|reset|state|compact|tool-result` ; +//! - `/api/conversations{,/new,/switch,/rename,/delete}`. +//! +//! `/api/chat/mode` et `/api/chat/criteria` restent au Go (état du mode Code +//! en base), mais l'événement qu'ils poussent aux abonnés doit atteindre LE +//! journal tenu ici : la façade relaie la requête puis journalise la réponse +//! ([`after_forward`]). Pièces jointes, fichiers, images, export : au Go, +//! qui suit la discussion active. + +use std::sync::Arc; +use std::time::Duration; + +use bytes::Bytes; +use http_body_util::{BodyExt, Full, Limited, StreamBody}; +use hyper::body::{Frame, Incoming}; +use hyper::header::{self, HeaderValue}; +use hyper::{Method, Request, Response, StatusCode}; +use serde_json::{Map, Value, json}; +use tokio::sync::mpsc; + +use crate::api::{Ctx, query_param}; +use crate::auth::json as send_json; +use crate::chat::generate::ChatService; +use crate::chat::journal::{SSE_HEARTBEAT, SseBatcher, sse_data_frame}; +use crate::chat::sessions::SessionError; +use crate::proxy::Body; + +/// Corps de requête lu au plus (un message collé peut être long). +const MAX_BODY: usize = 32 << 20; +/// Battement de cœur du flux (`sseHeartbeat`). +const HEARTBEAT: Duration = Duration::from_secs(4); + +/// Cette route est-elle servie par la conversation tenue ici ? +pub fn handles(path: &str) -> bool { + matches!( + path, + "/api/chat" + | "/api/chat/send" + | "/api/chat/stop" + | "/api/chat/reset" + | "/api/chat/state" + | "/api/chat/compact" + | "/api/chat/tool-result" + | "/api/conversations" + | "/api/conversations/new" + | "/api/conversations/switch" + | "/api/conversations/rename" + | "/api/conversations/delete" + ) +} + +/// Route relayée au Go dont la réponse est journalisée ici. +pub fn relays_delta(path: &str) -> bool { + matches!(path, "/api/chat/mode" | "/api/chat/criteria") +} + +/// Corps JSON de la requête ; vide → `{}` ; illisible → `Err(message)`. +async fn read_json(body: Incoming) -> Result { + let bytes = Limited::new(body, MAX_BODY) + .collect() + .await + .map_err(|e| e.to_string())? + .to_bytes(); + if bytes.iter().all(u8::is_ascii_whitespace) { + return Err("EOF".into()); + } + serde_json::from_slice(&bytes).map_err(|e| e.to_string()) +} + +fn ok() -> Response { + send_json(StatusCode::OK, &json!({ "ok": true })) +} + +fn refused(status: u16, error: &str) -> Response { + send_json( + StatusCode::from_u16(status).unwrap_or(StatusCode::SERVICE_UNAVAILABLE), + &json!({ "ok": false, "error": error }), + ) +} + +fn session_error(e: SessionError) -> Response { + send_json( + StatusCode::BAD_REQUEST, + &json!({ "ok": false, "error": e.to_string() }), + ) +} + +fn str_of<'a>(v: &'a Value, k: &str) -> &'a str { + v.get(k).and_then(Value::as_str).unwrap_or("") +} + +pub async fn serve(ctx: &Ctx, chat: &Arc, req: Request) -> Response { + let (parts, body) = req.into_parts(); + if let Err(refus) = ctx.guard(&parts.headers).await { + return *refus; + } + let path = parts.uri.path().to_string(); + match path.as_str() { + "/api/chat" => match read_json(body).await { + Ok(v) => subscribe(chat, &v), + Err(e) => plain(StatusCode::BAD_REQUEST, &e), + }, + "/api/chat/send" => { + let v = match read_json(body).await { + Ok(v) => v, + Err(e) => return refused(400, &e), + }; + match chat.send(v).await { + Ok(s) if s.dup => send_json(StatusCode::OK, &json!({ "ok": true, "dup": true })), + Ok(s) => send_json(StatusCode::OK, &json!({ "ok": true, "queued": s.queued })), + Err(r) => refused(r.status, &r.error), + } + } + "/api/chat/stop" => { + chat.stop().await; + ok() + } + "/api/chat/reset" => { + chat.reset().await; + ok() + } + "/api/chat/state" => send_json( + StatusCode::OK, + &Value::Object(chat.sessions.conv().state_json()), + ), + "/api/chat/compact" => match chat.compact_now().await { + Ok(()) => ok(), + Err(r) => refused(r.status, &r.error), + }, + "/api/chat/tool-result" => { + let id = query_param(&parts.uri, "id").unwrap_or_default(); + let id = id.trim(); + if id.is_empty() { + return send_json(StatusCode::BAD_REQUEST, &json!({ "error": "id manquant" })); + } + match chat.tool_result(id).await { + Some(s) => send_json(StatusCode::OK, &json!({ "result": s })), + None => send_json( + StatusCode::NOT_FOUND, + &json!({ "error": "résultat non disponible" }), + ), + } + } + "/api/conversations" => match chat.sessions.list().await { + Ok((list, active)) => send_json( + StatusCode::OK, + &json!({ "conversations": list, "active": active, "busy": chat.sessions.conv().is_generating() }), + ), + Err(e) => send_json(StatusCode::SERVICE_UNAVAILABLE, &json!({ "error": e })), + }, + "/api/conversations/new" => match chat.sessions.new_conv().await { + Ok(id) => send_json(StatusCode::OK, &json!({ "ok": true, "id": id })), + Err(e) => session_error(e), + }, + // Corps illisible : champs vides, comme le `_ = Decode` du Go. + "/api/conversations/switch" => { + let v = read_json(body).await.unwrap_or_default(); + match chat.sessions.switch(str_of(&v, "id")).await { + Ok(()) => ok(), + Err(e) => session_error(e), + } + } + "/api/conversations/rename" => { + let v = read_json(body).await.unwrap_or_default(); + match chat + .sessions + .rename(str_of(&v, "id"), str_of(&v, "title")) + .await + { + Ok(()) => ok(), + Err(e) => session_error(e), + } + } + "/api/conversations/delete" => { + let v = read_json(body).await.unwrap_or_default(); + match chat.sessions.delete(str_of(&v, "id")).await { + Ok(()) => ok(), + Err(e) => session_error(e), + } + } + _ => plain(StatusCode::NOT_FOUND, "404 page not found"), + } +} + +/// Après le relais d'une bascule de mode ou d'une édition de critères au Go : +/// l'état rendu est journalisé ici, pour que tous les appareils suivent. +pub async fn after_forward( + chat: &ChatService, + method: &Method, + path: &str, + resp: Response, +) -> Response { + if *method != Method::POST || resp.status() != StatusCode::OK { + return resp; + } + let (parts, body) = resp.into_parts(); + let bytes = match body.collect().await { + Ok(b) => b.to_bytes(), + Err(_) => return plain(StatusCode::BAD_GATEWAY, "réponse du serveur Go interrompue"), + }; + if let Ok(v) = serde_json::from_slice::(&bytes) { + let mut d = Map::new(); + match path { + "/api/chat/mode" if v.get("mode").is_some() => { + d.insert("mode".into(), v["mode"].clone()); + } + "/api/chat/criteria" if v.get("criteria").is_some() => { + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|t| t.as_millis() as i64) + .unwrap_or(0); + d.insert("criteria".into(), v["criteria"].clone()); + d.insert("ts".into(), ts.into()); + } + _ => {} + } + if !d.is_empty() { + chat.push_delta(d); + } + } + Response::from_parts(parts, Full::new(bytes).map_err(|n| match n {}).boxed()) +} + +/// `handleChat` : le flux d'abonnement. `from` : dernier Seq vu ; `tail` : +/// échanges à rejouer (absent : ancien client, tout) ; `conv_id` : discussion +/// affichée. +fn subscribe(chat: &Arc, v: &Value) -> Response { + let from = v.get("from").and_then(Value::as_i64).unwrap_or(0); + let tail = v.get("tail").and_then(Value::as_i64).unwrap_or(-1); + let mut sub = chat + .sessions + .conv() + .subscribe(from, tail, str_of(v, "conv_id")); + let (tx, rx) = mpsc::channel::(8); + tokio::spawn(async move { + let mut beat = tokio::time::interval(HEARTBEAT); + beat.tick().await; + loop { + tokio::select! { + evs = sub.next() => { + let mut batch = SseBatcher::new(); + for e in &evs { + if let Some(chunk) = batch.queue(&sse_data_frame(e)) + && tx.send(Bytes::from(chunk)).await.is_err() + { + return; + } + } + if let Some(chunk) = batch.flush() + && tx.send(Bytes::from(chunk)).await.is_err() + { + return; + } + } + _ = beat.tick() => { + if tx.send(Bytes::from_static(SSE_HEARTBEAT)).await.is_err() { + return; + } + } + _ = tx.closed() => return, + } + } + }); + let stream = futures_util::stream::unfold(rx, |mut rx| async move { + rx.recv() + .await + .map(|b| (Ok::<_, hyper::Error>(Frame::data(b)), rx)) + }); + let mut r = Response::new(BodyExt::boxed(StreamBody::new(stream))); + let h = r.headers_mut(); + h.insert( + header::CONTENT_TYPE, + HeaderValue::from_static("text/event-stream"), + ); + h.insert( + header::CACHE_CONTROL, + HeaderValue::from_static("no-cache, no-transform"), + ); + h.insert("x-accel-buffering", HeaderValue::from_static("no")); + r +} + +fn plain(status: StatusCode, msg: &str) -> Response { + let mut r = Response::new( + Full::new(Bytes::from(format!("{msg}\n"))) + .map_err(|n| match n {}) + .boxed(), + ); + *r.status_mut() = status; + r.headers_mut().insert( + header::CONTENT_TYPE, + HeaderValue::from_static("text/plain; charset=utf-8"), + ); + r +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn routes() { + assert!(handles("/api/chat") && handles("/api/conversations/delete")); + assert!(!handles("/api/chat/upload") && !handles("/api/chat/export")); + assert!(relays_delta("/api/chat/mode") && !relays_delta("/api/chat")); + } +} diff --git a/rust/loki-rs/src/lib.rs b/rust/loki-rs/src/lib.rs index dbc8d2c..ff5cecd 100644 --- a/rust/loki-rs/src/lib.rs +++ b/rust/loki-rs/src/lib.rs @@ -6,12 +6,14 @@ //! - `ui` : UI web statique embarquée (lot 2) ; //! - `auth` : authentification de l'API de pilotage (lot 2) ; //! - `oai` : surface compatible OpenAI `/v1` relayée au moteur (lot 5) ; -//! - `chat` : cœur chat (lot 6, bibliothèque en cours de portage) ; +//! - `chat` : cœur chat (lot 6) ; `chatapi` : ses routes, servies quand la +//! conversation est tenue par la façade (`LOKI_CHAT=rust`) ; //! - `update` : mise à jour des binaires au démarrage du conteneur. pub mod api; pub mod auth; pub mod chat; +pub mod chatapi; pub mod child; pub mod engine; pub mod engstatus; diff --git a/rust/loki-rs/src/main.rs b/rust/loki-rs/src/main.rs index e134036..af10d8d 100644 --- a/rust/loki-rs/src/main.rs +++ b/rust/loki-rs/src/main.rs @@ -114,12 +114,30 @@ fn serve() -> ExitCode { return ExitCode::FAILURE; } }; - let sup = child::Supervisor::spawn( - go_bin, - go_port, - vec![(auth::INTERNAL_ENV.to_string(), token.clone())], - ); + // LOKI_CHAT=rust : la conversation est tenue par la façade. Le Go le + // sait (il remet alors à la façade le trafic de chat du tunnel) et + // connaît son adresse locale. + let chat_rust = chat_held_by_facade(); + let mut env = vec![(auth::INTERNAL_ENV.to_string(), token.clone())]; + if chat_rust { + env.push(("LOKI_CHAT".into(), "rust".into())); + env.push(("LOKI_FACADE_ADDR".into(), format!("127.0.0.1:{web_port}"))); + } + let sup = child::Supervisor::spawn(go_bin, go_port, env); wait_for(backend, Duration::from_secs(30)).await; + let engine_paths = engine::Paths::from_env(); + let chat = if chat_rust { + match start_chat(backend, &token, &engine_paths.home).await { + Ok(c) => Some(c), + Err(e) => { + eprintln!("[loki-rs] conversation : {e}"); + sup.shutdown(Duration::from_secs(10)).await; + return ExitCode::FAILURE; + } + } + } else { + None + }; let addr = SocketAddr::from(([0, 0, 0, 0], web_port)); let listener = match tokio::net::TcpListener::bind(addr).await { @@ -135,7 +153,8 @@ fn serve() -> ExitCode { backend, internal_token: token, host_env: auth::HostEnv::from_env(), - engine: engine::Paths::from_env(), + engine: engine_paths, + chat, }; let res = proxy::serve(listener, cfg, shutdown_signal()).await; eprintln!("[loki-rs] arrêt demandé — arrêt du serveur Go"); @@ -150,6 +169,33 @@ fn serve() -> ExitCode { }) } +/// `LOKI_CHAT=rust` : conversation tenue par la façade (Go par défaut). +fn chat_held_by_facade() -> bool { + std::env::var("LOKI_CHAT").is_ok_and(|v| v.trim().eq_ignore_ascii_case("rust")) +} + +/// Charge la discussion active et prépare le service des tours. +async fn start_chat( + backend: SocketAddr, + token: &str, + home: &std::path::Path, +) -> Result, String> { + use loki_rs::chat::{conv::Conversation, generate::ChatService, host::ChatShared, sessions}; + let store = loki_rs::store::Store::new(backend, token.to_string()); + let s = sessions::Sessions::new( + store.clone(), + Conversation::new(), + sessions::GoSessionHooks(store.clone()), + ); + s.load().await?; + let shared = ChatShared::for_sessions(&s, store, home.join("chatimg")); + eprintln!("[loki-rs] conversation tenue par la façade (LOKI_CHAT=rust)"); + Ok(ChatService::new( + std::sync::Arc::new(s), + std::sync::Arc::new(shared), + )) +} + fn internal_token() -> Result { let mut b = [0u8; 32]; getrandom::fill(&mut b)?; diff --git a/rust/loki-rs/src/proxy.rs b/rust/loki-rs/src/proxy.rs index 6562228..60104ff 100644 --- a/rust/loki-rs/src/proxy.rs +++ b/rust/loki-rs/src/proxy.rs @@ -29,6 +29,7 @@ use tokio::net::TcpListener; use crate::api; use crate::auth::{HostEnv, INTERNAL_HEADER, InternalClient}; +use crate::chatapi; use crate::engine; use crate::oai; use crate::store::Store; @@ -59,6 +60,9 @@ pub struct Config { pub host_env: HostEnv, /// `$LOKI_HOME` et service moteur, pour les routes portées. pub engine: engine::Paths, + /// Conversation tenue par la façade (`LOKI_CHAT=rust`) : ses routes sont + /// servies ici ; `None` : relayées au Go. + pub chat: Option>, } struct Proxy { @@ -85,6 +89,7 @@ pub async fn serve( internal: InternalClient::new(cfg.backend, cfg.internal_token), host_env: cfg.host_env, engine: cfg.engine, + chat: cfg.chat, }, }); tokio::pin!(shutdown); @@ -142,6 +147,16 @@ impl Proxy { if api::handles(req.method(), path) { return api::serve(&self.ctx, req).await; } + if let Some(chat) = &self.ctx.chat { + if chatapi::handles(path) { + return chatapi::serve(&self.ctx, chat, req).await; + } + if chatapi::relays_delta(path) { + let (method, path) = (req.method().clone(), path.to_string()); + let resp = self.forward(req).await; + return chatapi::after_forward(chat, &method, &path, resp).await; + } + } self.forward(req).await } diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 2f6a47e..7bd2642 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -102,6 +102,14 @@ async fn handle(st: Arc, req: Request) -> Response let path = req.uri().path().to_string(); if !path.starts_with("/internal/") { st.relayed.lock().unwrap().push(path.clone()); + if path == "/api/chat/mode" && req.method() == Method::POST { + let body = req.into_body().collect().await.unwrap().to_bytes(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default(); + return resp( + 200, + serde_json::json!({"ok": true, "mode": v["mode"]}).to_string(), + ); + } return resp(200, format!("go:{path}")); } if req.headers().get("x-loki-internal").map(|v| v.as_bytes()) != Some(JETON.as_bytes()) { @@ -326,7 +334,39 @@ async fn handle(st: Arc, req: Request) -> Response /// Lance le faux Go puis la façade devant lui ; rend (adresse de la façade, /// état du faux Go, garde d'arrêt). pub async fn start(home: std::path::PathBuf) -> (SocketAddr, Arc, oneshot::Sender<()>) { + let (f, st, tx, _) = start_opts(home, false, None).await; + (f, st, tx) +} + +/// Comme [`start`], avec la conversation tenue par la façade +/// (`LOKI_CHAT=rust`) ; `engine_port` : port du faux moteur (config PORT). +pub async fn start_chat( + home: std::path::PathBuf, + engine_port: u16, +) -> ( + SocketAddr, + Arc, + oneshot::Sender<()>, + Arc, +) { + let (f, st, tx, chat) = start_opts(home, true, Some(engine_port)).await; + (f, st, tx, chat.unwrap()) +} + +async fn start_opts( + home: std::path::PathBuf, + with_chat: bool, + engine_port: Option, +) -> ( + SocketAddr, + Arc, + oneshot::Sender<()>, + Option>, +) { let st = Arc::new(FakeGo::default()); + if let Some(p) = engine_port { + st.set("config", "PORT", &p.to_string()); + } let l = TcpListener::bind("127.0.0.1:0").await.unwrap(); let backend = l.local_addr().unwrap(); *st.backend.lock().unwrap() = Some(backend); @@ -345,6 +385,22 @@ pub async fn start(home: std::path::PathBuf) -> (SocketAddr, Arc, onesho }); } }); + let chat = if with_chat { + use loki_rs::chat::{ + conv::Conversation, generate::ChatService, host::ChatShared, sessions, + }; + let store = loki_rs::store::Store::new(backend, JETON.into()); + let s = sessions::Sessions::new( + store.clone(), + Conversation::new(), + sessions::GoSessionHooks(store.clone()), + ); + s.load().await.unwrap(); + let shared = ChatShared::for_sessions(&s, store, home.join("chatimg")); + Some(ChatService::new(Arc::new(s), Arc::new(shared))) + } else { + None + }; let fl = TcpListener::bind("127.0.0.1:0").await.unwrap(); let facade = fl.local_addr().unwrap(); let (tx, rx) = oneshot::channel::<()>(); @@ -356,11 +412,12 @@ pub async fn start(home: std::path::PathBuf) -> (SocketAddr, Arc, onesho home, service: "loki-engine".into(), }, + chat: chat.clone(), }; tokio::spawn(loki_rs::proxy::serve(fl, cfg, async move { let _ = rx.await; })); - (facade, st, tx) + (facade, st, tx, chat) } /// GET via la façade : (code, en-têtes, corps). diff --git a/rust/loki-rs/tests/lot2.rs b/rust/loki-rs/tests/lot2.rs index c8a8d90..5c789fa 100644 --- a/rust/loki-rs/tests/lot2.rs +++ b/rust/loki-rs/tests/lot2.rs @@ -93,6 +93,7 @@ async fn banc(cle: Option<&str>) -> Banc { home: std::env::temp_dir(), service: "loki-engine".into(), }, + chat: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index e1d7204..2e7bcd9 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -572,6 +572,7 @@ async fn tour_complet_par_le_go() { kwargs: Default::default(), }, queued: Box::new(Vec::new), + notice: Box::new(|_| {}), }; let bash = Tool { kind: "function".into(), @@ -872,3 +873,199 @@ async fn moteur_pas_pret_et_message_vide() { .await; assert_eq!(r.unwrap_err().status, 400); } + +// ---- les routes du chat servies par la façade (LOKI_CHAT=rust) ---- + +async fn call( + facade: std::net::SocketAddr, + method: &str, + path: &str, + body: &str, +) -> (u16, serde_json::Value) { + use http_body_util::{BodyExt, Full}; + let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .build_http::>(); + let req = hyper::Request::builder() + .method(method) + .uri(format!("http://{facade}{path}")) + .body(Full::new(bytes::Bytes::from(body.to_string()))) + .unwrap(); + let resp = client.request(req).await.unwrap(); + let status = resp.status().as_u16(); + let b = resp.into_body().collect().await.unwrap().to_bytes(); + ( + status, + serde_json::from_slice(&b).unwrap_or(serde_json::Value::String( + String::from_utf8_lossy(&b).into(), + )), + ) +} + +/// Ouvre le flux SSE et lit jusqu'à `caught_up` : rend les événements. +async fn read_stream( + facade: std::net::SocketAddr, + body: &str, +) -> (hyper::HeaderMap, Vec) { + use http_body_util::{BodyExt, Full}; + let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .build_http::>(); + let req = hyper::Request::builder() + .method("POST") + .uri(format!("http://{facade}/api/chat")) + .body(Full::new(bytes::Bytes::from(body.to_string()))) + .unwrap(); + let resp = client.request(req).await.unwrap(); + let headers = resp.headers().clone(); + let mut body = resp.into_body(); + let mut text = String::new(); + let mut events = Vec::new(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while let Some(Ok(f)) = body.frame().await { + if let Some(d) = f.data_ref() { + text.push_str(&String::from_utf8_lossy(d)); + } + while let Some(i) = text.find("\n\n") { + let frame: String = text.drain(..i + 2).collect(); + if let Some(data) = frame.trim().strip_prefix("data: ") { + let v: serde_json::Value = serde_json::from_str(data).unwrap(); + let delta = v["choices"][0]["delta"].clone(); + let done = delta.get("caught_up").is_some(); + events.push(delta); + if done { + return; + } + } + } + } + }) + .await + .expect("flux sans caught_up"); + (headers, events) +} + +#[tokio::test] +async fn routes_du_chat_par_la_facade() { + let (url, _seen) = fake_engine(tool_then_answer("voilà")).await; + let t = tempfile::tempdir().unwrap(); + let port: u16 = port_of(&url).parse().unwrap(); + let (facade, go, _stop, chat) = fakego::start_chat(t.path().into(), port).await; + + // Envoi, puis le flux rejoue le tour jusqu'à caught_up. + let (code, v) = call( + facade, + "POST", + "/api/chat/send", + r#"{"message":"salut","cid":"r1"}"#, + ) + .await; + assert_eq!( + (code, v.clone()), + (200, serde_json::json!({"ok": true, "queued": false})), + "{v}" + ); + until_done(&chat, 1).await; + let (headers, evs) = read_stream(facade, r#"{"from":0,"tail":20}"#).await; + assert_eq!(headers["content-type"], "text/event-stream"); + assert!(evs.iter().any(|e| e["user"] == "salut")); + assert!(evs.iter().any(|e| e["content"] == "voilà")); + assert!(evs.iter().any(|e| e.get("turn_done").is_some())); + // Les routes du chat ne partent plus au Go. + assert!( + !go.relayed + .lock() + .unwrap() + .iter() + .any(|p| p.starts_with("/api/chat")) + ); + + let (_, st) = call(facade, "GET", "/api/chat/state", "").await; + assert_eq!( + (st["generating"].clone(), st["turns"].clone()), + (serde_json::json!(false), serde_json::json!(1)) + ); + let (code, v) = call(facade, "POST", "/api/chat/send", "{").await; + assert_eq!((code, v["ok"].clone()), (400, serde_json::json!(false))); + assert_eq!( + call(facade, "GET", "/api/chat/tool-result", "").await.0, + 400 + ); + assert_eq!( + call(facade, "GET", "/api/chat/tool-result?id=c9.zz", "") + .await + .0, + 404 + ); + + // Discussions. + let (_, list) = call(facade, "GET", "/api/conversations", "").await; + let a = list["active"].as_str().unwrap().to_string(); + assert_eq!(list["conversations"].as_array().unwrap().len(), 1); + assert_eq!(list["busy"], false); + let (_, n) = call(facade, "POST", "/api/conversations/new", "").await; + let b = n["id"].as_str().unwrap().to_string(); + assert_ne!(a, b); + let (code, _) = call( + facade, + "POST", + "/api/conversations/switch", + &format!(r#"{{"id":"{a}"}}"#), + ) + .await; + assert_eq!(code, 200); + let (code, v) = call( + facade, + "POST", + "/api/conversations/switch", + r#"{"id":"inconnue"}"#, + ) + .await; + assert_eq!( + (code, v["error"].clone()), + (400, serde_json::json!("discussion introuvable")) + ); + assert_eq!( + call( + facade, + "POST", + "/api/conversations/rename", + &format!(r#"{{"id":"{a}","title":"Mon fil"}}"#) + ) + .await + .0, + 200 + ); + assert_eq!( + call( + facade, + "POST", + "/api/conversations/delete", + &format!(r#"{{"id":"{b}"}}"#) + ) + .await + .0, + 200 + ); + let (_, list) = call(facade, "GET", "/api/conversations", "").await; + assert_eq!(list["conversations"][0]["title"], "Mon fil"); + + // Bascule du mode Code : relayée au Go, journalisée ici. + let (code, v) = call(facade, "POST", "/api/chat/mode", r#"{"mode":"code"}"#).await; + assert_eq!((code, v["mode"].clone()), (200, serde_json::json!("code"))); + assert!( + go.relayed + .lock() + .unwrap() + .contains(&"/api/chat/mode".to_string()) + ); + assert_eq!(deltas(&chat, "mode"), [serde_json::json!("code")]); + + // Reset : fil vide, fichiers de la discussion effacés par le Go. + assert_eq!(call(facade, "POST", "/api/chat/reset", "").await.0, 200); + assert!(chat.sessions.conv().with_state(|st| st.log.is_empty())); + assert!( + go.conv_log + .lock() + .unwrap() + .contains(&format!("dropped {a}")) + ); +} diff --git a/rust/loki-rs/tests/proxy.rs b/rust/loki-rs/tests/proxy.rs index 3418dc2..575f6ca 100644 --- a/rust/loki-rs/tests/proxy.rs +++ b/rust/loki-rs/tests/proxy.rs @@ -99,6 +99,7 @@ async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) { home: std::env::temp_dir(), service: "loki-engine".into(), }, + chat: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await;