From ad46bb69a661dfedf4786e96ea137087ab8ddf17 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 11:19:42 +0000 Subject: [PATCH] =?UTF-8?q?Lot=206=20:=20la=20fa=C3=A7ade=20re=C3=A7oit=20?= =?UTF-8?q?le=20chat=20du=20Go=20et=20ses=20=C3=A9v=C3=A9nements?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Une requête qui porte le jeton interne vient du Go, qui l'a déjà authentifiée (tunnel du relais) : les routes du chat la servent sans redemander la clé. L'API interne de la façade, réservée à ce jeton, reçoit les événements d'affichage nés au Go (mode Code, critères) et le rechargement après déverrouillage : une discussion chargée vide mémoire verrouillée revient au déverrouillage, jamais par-dessus un fil rempli (sinon le premier message aurait écrasé la discussion chiffrée). Le relais par la façade de /api/chat/mode et /api/chat/criteria et les critères rendus par l'outil disparaissent : le Go pousse ses événements lui-même, une seule source quel que soit le chemin. README : LOKI_CHAT=rust et ce qu'il ne prend pas encore en charge. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv --- README.md | 1 + rust/loki-rs/src/api.rs | 13 ++++ rust/loki-rs/src/auth.rs | 3 +- rust/loki-rs/src/chat/generate.rs | 6 -- rust/loki-rs/src/chat/host.rs | 17 +---- rust/loki-rs/src/chat/sessions.rs | 18 +++++ rust/loki-rs/src/chatapi.rs | 109 ++++++++++++++++-------------- rust/loki-rs/src/proxy.rs | 26 ++++--- rust/loki-rs/tests/lot6.rs | 72 ++++++++++++++++++-- 9 files changed, 175 insertions(+), 90 deletions(-) diff --git a/README.md b/README.md index 6beea7f..39d4afc 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,7 @@ redémarrages (volume `/data`). Variables d'environnement du conteneur : | `LOKI_AUTO_UPDATE` | `0` : pas de mise à jour des binaires au démarrage | `1` | | `LOKI_FACADE` | `0` : sans façade Rust, le Go sert directement le port public (porte de sortie) | `1` | | `LOKI_GO_PORT` | port local (127.0.0.1) du serveur Go derrière la façade | `8091` | +| `LOKI_CHAT` | `rust` : la conversation est tenue par la façade Rust (expérimental, migration en cours). Le Go garde le mode Code, les pièces jointes, les fichiers et l'export ; sa boucle de vérification du mode Code et les clés `PROJ_SNAPSHOT`, `KEEP_TURN_IMAGES`, `NUDGE_IN_TOOL`, `SIDE_SLOT`, `SLOT_PERSIST` ne sont pas prises en charge dans ce mode | Go | **Accès par nom de domaine.** Sans clé de pilotage, l'API n'accepte que les hôtes locaux (IP, `localhost`, nom sans point, `.local`/`.lan`…) et refuse diff --git a/rust/loki-rs/src/api.rs b/rust/loki-rs/src/api.rs index f774f4b..020a999 100644 --- a/rust/loki-rs/src/api.rs +++ b/rust/loki-rs/src/api.rs @@ -26,6 +26,19 @@ pub struct Ctx { pub engine: engine::Paths, /// Conversation tenue par la façade (`LOKI_CHAT=rust`) ; `None` : au Go. pub chat: Option>, + /// Jeton de l'API interne : une requête qui le porte vient du Go. + pub internal_token: String, +} + +impl Ctx { + /// La requête vient du Go (jeton interne valide) : déjà authentifiée par + /// lui (tunnel du relais, redistribution E2E). + pub fn from_go(&self, headers: &HeaderMap) -> bool { + !self.internal_token.is_empty() + && headers.get(crate::auth::INTERNAL_HEADER).is_some_and(|v| { + auth::constant_time_eq(v.as_bytes(), self.internal_token.as_bytes()) + }) + } } impl Ctx { diff --git a/rust/loki-rs/src/auth.rs b/rust/loki-rs/src/auth.rs index b3fabb7..1beee73 100644 --- a/rust/loki-rs/src/auth.rs +++ b/rust/loki-rs/src/auth.rs @@ -225,7 +225,8 @@ pub fn check_bearer(headers: &HeaderMap, key: &str) -> bool { constant_time_eq(got.trim().as_bytes(), key.as_bytes()) } -fn constant_time_eq(a: &[u8], b: &[u8]) -> bool { +/// Comparaison en temps constant (secrets). +pub fn constant_time_eq(a: &[u8], b: &[u8]) -> bool { if a.len() != b.len() { return false; } diff --git a/rust/loki-rs/src/chat/generate.rs b/rust/loki-rs/src/chat/generate.rs index 79f6d40..8ebbdd7 100644 --- a/rust/loki-rs/src/chat/generate.rs +++ b/rust/loki-rs/src/chat/generate.rs @@ -752,12 +752,6 @@ 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 a8266c1..9228ea1 100644 --- a/rust/loki-rs/src/chat/host.rs +++ b/rust/loki-rs/src/chat/host.rs @@ -312,9 +312,6 @@ 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> ». @@ -399,19 +396,7 @@ impl TurnHost for GoTurnHost { .tool_exec(name, &Value::Object(args.clone()), &self.caps) .await { - 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) - } + Ok(v) => 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/chat/sessions.rs b/rust/loki-rs/src/chat/sessions.rs index f0740a7..43b8266 100644 --- a/rust/loki-rs/src/chat/sessions.rs +++ b/rust/loki-rs/src/chat/sessions.rs @@ -220,6 +220,24 @@ impl Sessions { Err(last_err) } + /// `reloadEncryptedStores` : après le déverrouillage de la mémoire, + /// recharge la discussion si elle a été chargée vide (mémoire verrouillée + /// au démarrage). Jamais un fil déjà rempli : écraser le tour en cours le + /// perdrait. Rend vrai si elle a été rechargée. + pub async fn reload_if_empty(&self) -> Result { + let _op = self.op.lock().await; + if self.conv.is_generating() || !self.conv.with_state(|st| st.log.is_empty()) { + return Ok(false); + } + let id = self.ensure_active_locked().await?; + let Some(body) = self.store.get_open(BK_CHAT, &conv_key(&id)).await? else { + return Ok(false); + }; + self.conv.load_from(&id, &body); + self.hooks.activated(&id).await; + Ok(true) + } + /// `convEnsureActive` : identifiant de la discussion active, créée au /// besoin (reprise silencieuse du fil unique d'avant les discussions). pub async fn ensure_active(&self) -> Result { diff --git a/rust/loki-rs/src/chatapi.rs b/rust/loki-rs/src/chatapi.rs index d0a696e..ba981e3 100644 --- a/rust/loki-rs/src/chatapi.rs +++ b/rust/loki-rs/src/chatapi.rs @@ -6,11 +6,11 @@ //! - `/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. +//! Le reste (mode Code et critères, pièces jointes, fichiers, images, +//! export) reste au Go, qui suit la discussion active et pousse ses propres +//! événements d'affichage dans le journal tenu ici ([`serve_internal`]). Le +//! trafic de chat qui arrive au Go sans passer par la façade (tunnel du +//! relais) lui est remis, authentifié par le jeton interne. use std::sync::Arc; use std::time::Duration; @@ -20,7 +20,7 @@ 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 serde_json::{Value, json}; use tokio::sync::mpsc; use crate::api::{Ctx, query_param}; @@ -54,9 +54,43 @@ pub fn handles(path: &str) -> bool { ) } -/// 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") +/// API interne de la façade, pour le Go : un événement d'affichage né chez +/// lui (bascule du mode Code, critères) à journaliser ici, et le +/// rechargement de la discussion après le déverrouillage de la mémoire. +pub fn is_internal(path: &str) -> bool { + matches!( + path, + "/internal/facade/v1/delta" | "/internal/facade/v1/reload" + ) +} + +pub async fn serve_internal(chat: &Arc, req: Request) -> Response { + let path = req.uri().path().to_string(); + if req.method() != Method::POST { + return send_json( + StatusCode::METHOD_NOT_ALLOWED, + &json!({ "error": "POST attendu" }), + ); + } + match path.as_str() { + "/internal/facade/v1/delta" => match read_json(req.into_body()).await { + Ok(Value::Object(d)) if !d.is_empty() => { + chat.push_delta(d); + Response::builder() + .status(StatusCode::NO_CONTENT) + .body(Full::new(Bytes::new()).map_err(|n| match n {}).boxed()) + .unwrap_or_default() + } + _ => send_json( + StatusCode::BAD_REQUEST, + &json!({ "error": "événement illisible" }), + ), + }, + _ => match chat.sessions.reload_if_empty().await { + Ok(reloaded) => send_json(StatusCode::OK, &json!({ "reloaded": reloaded })), + Err(e) => send_json(StatusCode::SERVICE_UNAVAILABLE, &json!({ "error": e })), + }, + } } /// Corps JSON de la requête ; vide → `{}` ; illisible → `Err(message)`. @@ -94,9 +128,15 @@ 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 { +/// `from_go` : relayée par le Go, qui l'a déjà authentifiée. +pub async fn serve( + ctx: &Ctx, + chat: &Arc, + req: Request, + from_go: bool, +) -> Response { let (parts, body) = req.into_parts(); - if let Err(refus) = ctx.guard(&parts.headers).await { + if !from_go && let Err(refus) = ctx.guard(&parts.headers).await { return *refus; } let path = parts.uri.path().to_string(); @@ -187,45 +227,6 @@ pub async fn serve(ctx: &Ctx, chat: &Arc, req: Request) - } } -/// 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. @@ -306,7 +307,11 @@ mod tests { #[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")); + assert!( + !handles("/api/chat/upload") + && !handles("/api/chat/export") + && !handles("/api/chat/mode") + ); + assert!(is_internal("/internal/facade/v1/delta") && !is_internal("/internal/v1/auth")); } } diff --git a/rust/loki-rs/src/proxy.rs b/rust/loki-rs/src/proxy.rs index 60104ff..b2b55da 100644 --- a/rust/loki-rs/src/proxy.rs +++ b/rust/loki-rs/src/proxy.rs @@ -86,10 +86,11 @@ pub async fn serve( backend: cfg.backend, ctx: api::Ctx { store: Store::new(cfg.backend, cfg.internal_token.clone()), - internal: InternalClient::new(cfg.backend, cfg.internal_token), + internal: InternalClient::new(cfg.backend, cfg.internal_token.clone()), host_env: cfg.host_env, engine: cfg.engine, chat: cfg.chat, + internal_token: cfg.internal_token.clone(), }, }); tokio::pin!(shutdown); @@ -127,12 +128,22 @@ pub async fn serve( impl Proxy { async fn handle(&self, mut req: Request) -> Response { + // Le Go (et lui seul : jeton tiré à chaque démarrage) remet à la + // façade le trafic de chat qui lui arrive sans passer par elle. + let from_go = self.ctx.chat.is_some() && self.ctx.from_go(req.headers()); // Jamais de confiance accordée sur la foi d'un en-tête du client : // ni « je viens du tunnel », ni « je suis la façade ». req.headers_mut().remove(VIA_TUNNEL); req.headers_mut().remove(INTERNAL_HEADER); let path = req.uri().path(); + // L'API interne de la façade : pour le Go seulement. + if from_go + && let Some(chat) = &self.ctx.chat + && chatapi::is_internal(path) + { + return chatapi::serve_internal(chat, req).await; + } // L'API interne du Go n'est jamais joignable de l'extérieur. if path.starts_with("/internal/") { return plain(StatusCode::NOT_FOUND, "404 page not found"); @@ -147,15 +158,10 @@ 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; - } + if let Some(chat) = &self.ctx.chat + && chatapi::handles(path) + { + return chatapi::serve(&self.ctx, chat, req, from_go).await; } self.forward(req).await } diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index 2e7bcd9..192d9ac 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -572,7 +572,6 @@ async fn tour_complet_par_le_go() { kwargs: Default::default(), }, queued: Box::new(Vec::new), - notice: Box::new(|_| {}), }; let bash = Tool { kind: "function".into(), @@ -901,6 +900,20 @@ async fn call( ) } +/// POST porteur du jeton interne (requête relayée par le Go) : le statut. +async fn call_internal(facade: std::net::SocketAddr, path: &str, body: &str) -> u16 { + use http_body_util::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}{path}")) + .header("x-loki-internal", fakego::JETON) + .body(Full::new(bytes::Bytes::from(body.to_string()))) + .unwrap(); + client.request(req).await.unwrap().status().as_u16() +} + /// Ouvre le flux SSE et lit jusqu'à `caught_up` : rend les événements. async fn read_stream( facade: std::net::SocketAddr, @@ -1048,19 +1061,45 @@ async fn routes_du_chat_par_la_facade() { 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"))); + // Mode Code : servi par le Go, qui pousse son événement au journal d'ici + // par l'API interne de la façade (jeton interne exigé). + assert_eq!( + call(facade, "POST", "/api/chat/mode", r#"{"mode":"code"}"#) + .await + .0, + 200 + ); assert!( go.relayed .lock() .unwrap() .contains(&"/api/chat/mode".to_string()) ); + let (code, _) = call( + facade, + "POST", + "/internal/facade/v1/delta", + r#"{"mode":"code"}"#, + ) + .await; + assert_eq!(code, 404, "sans le jeton, l'API interne n'existe pas"); + let code = call_internal(facade, "/internal/facade/v1/delta", r#"{"mode":"code"}"#).await; + assert_eq!(code, 204); assert_eq!(deltas(&chat, "mode"), [serde_json::json!("code")]); + // Fil déjà rempli : le déverrouillage ne le recharge pas. + assert_eq!( + call_internal(facade, "/internal/facade/v1/reload", "").await, + 200 + ); + assert!(!chat.sessions.conv().with_state(|st| st.log.is_empty())); + + // Trafic remis par le Go (tunnel) : déjà authentifié par lui. + go.set("state", "web_key", "cle-web"); + assert_eq!(call(facade, "GET", "/api/chat/state", "").await.0, 401); + assert_eq!(call_internal(facade, "/api/chat/state", "").await, 200); // Reset : fil vide, fichiers de la discussion effacés par le Go. - assert_eq!(call(facade, "POST", "/api/chat/reset", "").await.0, 200); + assert_eq!(call_internal(facade, "/api/chat/reset", "").await, 200); assert!(chat.sessions.conv().with_state(|st| st.log.is_empty())); assert!( go.conv_log @@ -1069,3 +1108,26 @@ async fn routes_du_chat_par_la_facade() { .contains(&format!("dropped {a}")) ); } + +#[tokio::test] +async fn deverrouillage_recharge_la_discussion_chargee_vide() { + let (s, _h, go, _stop, _t) = sessions().await; + go.set(BK_CHAT, "active", "c1"); + go.set( + BK_CHAT, + "conv:c1", + r#"SCELLE:{"messages":[{"role":"user","content":"fil chiffré"}],"log":[{"seq":1,"ts":0,"delta":{"user":"fil chiffré"}}],"seq":1,"ctx_used":0}"#, + ); + // Mémoire verrouillée au démarrage : fil vide, rien d'écrasé. + *go.locked.lock().unwrap() = true; + s.load().await.unwrap(); + assert_eq!(s.conv().id(), "c1"); + assert!(textes(&s).is_empty()); + assert_eq!(s.reload_if_empty().await, Ok(false), "toujours verrouillée"); + // Déverrouillée : le fil revient. + *go.locked.lock().unwrap() = false; + assert_eq!(s.reload_if_empty().await, Ok(true)); + assert_eq!(textes(&s), ["fil chiffré"]); + // Déjà rempli : jamais rechargé par-dessus. + assert_eq!(s.reload_if_empty().await, Ok(false)); +}