diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs index 27b4823..fa5701f 100644 --- a/rust/loki-rs/src/chat/mod.rs +++ b/rust/loki-rs/src/chat/mod.rs @@ -11,6 +11,7 @@ //! - [`conv`] : la discussion en mémoire (état, epoch, verrou du tour, //! abonnés) ; //! - [`persist`] : format en base, index des discussions, écrivain unique ; +//! - [`sessions`] : plusieurs discussions (index, bascules, chargement) ; //! - [`lease`] : bail du moteur (verrou de génération partagé avec le Go). pub mod compact; @@ -19,6 +20,7 @@ pub mod engine; pub mod journal; pub mod lease; pub mod persist; +pub mod sessions; pub mod stream; pub mod types; pub mod wire; diff --git a/rust/loki-rs/src/chat/persist.rs b/rust/loki-rs/src/chat/persist.rs index 233067c..81079eb 100644 --- a/rust/loki-rs/src/chat/persist.rs +++ b/rust/loki-rs/src/chat/persist.rs @@ -97,7 +97,7 @@ pub fn tool_res_conv(key: &str) -> &str { } } -fn now_unix() -> i64 { +pub(crate) fn now_unix() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs() as i64) @@ -783,16 +783,30 @@ async fn run(sh: Arc>) { #[derive(Clone)] pub struct StoreSink { pub store: Store, + /// Sérialise les lectures-modifications-écritures de l'index + /// (`convIndexMu`) : celles de l'écrivain et celles des opérations sur + /// les discussions ([`super::sessions`]), qui partagent ce verrou. + pub index_lock: Arc>, +} + +impl StoreSink { + pub fn new(store: Store) -> Self { + StoreSink { + store, + index_lock: Arc::new(tokio::sync::Mutex::new(())), + } + } } impl PersistSink for StoreSink { - fn commit( + async fn commit( &self, snaps: Vec, res: Vec, gate: ResGate, - ) -> impl Future + Send { - commit_batch(&self.store, snaps, res, gate, now_unix()) + ) -> CommitOutcome { + let _index = self.index_lock.lock().await; + commit_batch(&self.store, snaps, res, gate, now_unix()).await } } diff --git a/rust/loki-rs/src/chat/sessions.rs b/rust/loki-rs/src/chat/sessions.rs new file mode 100644 index 0000000..a3dc7c3 --- /dev/null +++ b/rust/loki-rs/src/chat/sessions.rs @@ -0,0 +1,352 @@ +//! Plusieurs discussions — port des opérations de `chat_sessions.go` et du +//! chargement au démarrage (`LoadConversation`, `chat_conversation.go`). +//! +//! Une seule discussion en mémoire ([`Conversation`]), rangée dans un tiroir +//! nommé (`chat/conv:`), listée dans `chat/index`, ouverte selon +//! `chat/active` — partagée par tous les appareils. Basculer = enregistrer la +//! courante, charger l'autre et incrémenter l'epoch : les abonnés reçoivent +//! `{reset}` et rejouent le nouveau fil. +//! +//! Les opérations de haut niveau (création, bascule, renommage, suppression) +//! sont sérialisées (`convOpMu`) ; les lectures-modifications-écritures de +//! l'index partagent leur verrou avec l'écrivain ([`StoreSink::index_lock`], +//! `convIndexMu`). +//! +//! Ce qui vit hors de la conversation reste derrière [`SessionHooks`] : les +//! fichiers, travaux, critères et mode d'une discussion supprimée, ses +//! résultats d'outils, et le préchauffage après une bascule. Le changement de +//! projet (`projectSwitch`) reste au Go pour l'instant. + +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use super::conv::Conversation; +use super::persist::{ + BK_CHAT, CK_ACTIVE, CK_INDEX, CK_LEGACY, ConvMeta, DEFAULT_PROJECT_SLUG, IndexError, Persister, + StoreSink, conv_key, decode_index, encode_index, index_for_project, new_conv_id, now_unix, + remove_from, rename_in, +}; +use crate::store::{BK_STATE, Store}; + +/// Projet actif (`stActiveProjectKey`, bucket `state`). +pub const ST_ACTIVE_PROJECT: &str = "active_project"; + +/// Essais de lecture au démarrage, et attente entre deux (`loadConvAttempts`, +/// `loadConvRetryWait`) : l'ancien process peut encore tenir la base. +pub const LOAD_ATTEMPTS: u32 = 4; +pub const LOAD_RETRY_WAIT: Duration = Duration::from_millis(250); + +/// Ce qui, autour d'une discussion, ne vit pas dans la conversation. +pub trait SessionHooks: Send + Sync + 'static { + /// Discussion supprimée : ses fichiers, travaux d'arrière-plan, critères, + /// mode et résultats d'outils partent avec elle. + fn dropped(&self, id: &str) -> impl Future + Send; + /// Bascule faite (`prewarmAfterSwitch`). + fn switched(&self, id: &str) -> impl Future + Send; +} + +/// Refus d'une opération sur les discussions. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SessionError { + /// « identifiant manquant » / « identifiant ou titre manquant » + Missing, + /// « discussion introuvable » + NotFound, + /// Base illisible. + Store(String), +} + +impl std::fmt::Display for SessionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SessionError::Missing => f.write_str("identifiant ou titre manquant"), + SessionError::NotFound => f.write_str("discussion introuvable"), + SessionError::Store(e) => write!(f, "base illisible : {e}"), + } + } +} + +impl From for SessionError { + fn from(e: IndexError) -> Self { + match e { + IndexError::Missing => SessionError::Missing, + IndexError::NotFound => SessionError::NotFound, + } + } +} + +/// Les discussions de l'utilisateur, autour de la discussion ouverte. +pub struct Sessions { + store: Store, + conv: Arc, + persister: Persister, + hooks: H, + /// `convOpMu`. + op: tokio::sync::Mutex<()>, +} + +impl Sessions { + pub fn new(store: Store, conv: Arc, hooks: H) -> Self { + let persister = Persister::new(StoreSink::new(store.clone())); + Sessions { + store, + conv, + persister, + hooks, + op: tokio::sync::Mutex::new(()), + } + } + + pub fn conv(&self) -> &Arc { + &self.conv + } + + pub fn persister(&self) -> &Persister { + &self.persister + } + + /// `activeProjectSlug` : projet actif, le projet par défaut à défaut. + pub async fn active_project(&self) -> String { + match self.store.get_str(BK_STATE, ST_ACTIVE_PROJECT).await.trim() { + "" => DEFAULT_PROJECT_SLUG.to_string(), + s => s.to_string(), + } + } + + /// `convIndex` : l'index, plus récente en tête. Illisible ou verrouillé : + /// vide — la liste est masquée, rien n'est perdu. + pub async fn read_index(&self) -> Vec { + match self.store.get_open(BK_CHAT, CK_INDEX).await { + Ok(Some(b)) => decode_index(&b), + _ => Vec::new(), + } + } + + async fn save_index(&self, idx: &[ConvMeta]) { + if let Err(e) = self + .store + .put_seal(BK_CHAT, CK_INDEX, encode_index(idx)) + .await + { + eprintln!("[conv] index non enregistré : {e:?}"); + } + } + + /// Ajoute une entrée à l'index (création, migration). + async fn index_add(&self, id: &str) { + let now = now_unix(); + let project = self.active_project().await; + let _g = self.persister.sink().index_lock.lock().await; + let mut idx = self.read_index().await; + idx.push(ConvMeta { + id: id.to_string(), + title: String::new(), + created: now, + updated: now, + turns: 0, + project, + }); + self.save_index(&idx).await; + } + + /// `LoadConversation` : recharge la discussion active au démarrage. Une + /// lecture RATÉE n'est pas une absence : on réessaie, et en cas d'échec + /// durable on ne touche à rien (ni identifiant forgé, ni fil écrasé). + pub async fn load(&self) -> Result<(), String> { + let _op = self.op.lock().await; + let mut last_err = String::new(); + for attempt in 0..LOAD_ATTEMPTS { + if attempt > 0 { + tokio::time::sleep(LOAD_RETRY_WAIT).await; + } + // Sonde : tant qu'elle échoue, surtout ne pas créer de discussion. + if let Err(e) = self.store.get(BK_CHAT, CK_ACTIVE).await { + last_err = e; + continue; + } + let id = self.ensure_active_locked().await?; + match self.store.get_open(BK_CHAT, &conv_key(&id)).await { + Ok(body) => { + self.conv.load_from(&id, &body.unwrap_or_default()); + return Ok(()); + } + Err(e) => last_err = e, + } + } + eprintln!( + "[conv] base illisible au démarrage ({last_err}) — aucune discussion chargée ; rien n'a été écrasé, l'historique est intact" + ); + Err(last_err) + } + + /// `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 { + let _op = self.op.lock().await; + self.ensure_active_locked().await + } + + async fn ensure_active_locked(&self) -> Result { + let cached = self.conv.id(); + if !cached.is_empty() { + return Ok(cached); + } + if let Some(raw) = self.store.get(BK_CHAT, CK_ACTIVE).await? { + let id = String::from_utf8_lossy(&raw).trim().to_string(); + if !id.is_empty() { + return Ok(id); + } + } + let id = new_conv_id(); + if let Some(legacy) = self.store.get(BK_CHAT, CK_LEGACY).await? + && !legacy.is_empty() + { + // Le fil unique devient la première discussion ; la clé d'origine + // reste intacte (retour arrière sans perte). + if let Err(e) = self.store.put_seal(BK_CHAT, &conv_key(&id), legacy).await { + eprintln!("[conv] reprise du fil unique non écrite : {e:?}"); + } + } + self.store + .put(BK_CHAT, CK_ACTIVE, id.as_bytes().to_vec()) + .await?; + self.index_add(&id).await; + Ok(id) + } + + /// `ConvList` : les discussions du projet actif, et l'active. + pub async fn list(&self) -> Result<(Vec, String), String> { + let project = self.active_project().await; + let idx = self.read_index().await; + let list = index_for_project(&idx, &project) + .into_iter() + .cloned() + .collect(); + Ok((list, self.ensure_active().await?)) + } + + /// `convActivate` : pointeur en base, identifiant et contenu changent + /// ensemble. + async fn activate(&self, id: &str) -> Result<(), SessionError> { + self.store + .put(BK_CHAT, CK_ACTIVE, id.as_bytes().to_vec()) + .await + .map_err(SessionError::Store)?; + let body = self + .store + .get_open(BK_CHAT, &conv_key(id)) + .await + .ok() + .flatten() + .unwrap_or_default(); + self.conv.load_from(id, &body); + Ok(()) + } + + /// `persist` : enregistre la discussion ouverte et ATTEND l'écriture. + pub async fn persist(&self) { + let project = self.active_project().await; + let snap = self.conv.snapshot(&self.persister, &project); + if snap.id.is_empty() { + self.persister.flush().await; + return; + } + self.persister.persist(snap).await; + } + + /// `persistAsync` : instantané tout de suite, écriture par l'écrivain. + pub async fn persist_async(&self) { + let project = self.active_project().await; + let snap = self.conv.snapshot(&self.persister, &project); + if !snap.id.is_empty() { + self.persister.enqueue_snap(snap); + } + } + + /// `convCreate` : discussion vide ouverte SANS enregistrer la courante. + async fn create_locked(&self) -> Result { + let id = new_conv_id(); + self.index_add(&id).await; + self.activate(&id).await?; + Ok(id) + } + + /// `convNew` : met la courante de côté puis en ouvre une neuve. + pub async fn new_conv(&self) -> Result { + let _op = self.op.lock().await; + self.persist().await; + self.create_locked().await + } + + /// `convSwitch` : enregistre la courante, charge la cible. + pub async fn switch(&self, id: &str) -> Result<(), SessionError> { + let _op = self.op.lock().await; + if id.is_empty() { + return Err(SessionError::Missing); + } + if !self.read_index().await.iter().any(|m| m.id == id) { + return Err(SessionError::NotFound); + } + if id + == self + .ensure_active_locked() + .await + .map_err(SessionError::Store)? + { + return Ok(()); + } + self.persist().await; + self.activate(id).await?; + self.hooks.switched(id).await; + Ok(()) + } + + /// `convRename`. + pub async fn rename(&self, id: &str, title: &str) -> Result<(), SessionError> { + let _op = self.op.lock().await; + let _g = self.persister.sink().index_lock.lock().await; + let mut idx = self.read_index().await; + rename_in(&mut idx, id, title)?; + self.save_index(&idx).await; + Ok(()) + } + + /// `convDelete` : supprimer celle qui est ouverte bascule sur la plus + /// récente restante, ou sur une neuve : il y a TOUJOURS une discussion + /// active. + pub async fn delete(&self, id: &str) -> Result<(), SessionError> { + let _op = self.op.lock().await; + if !self.read_index().await.iter().any(|m| m.id == id) { + return Err(SessionError::NotFound); + } + // Rien ne doit plus la viser, puis l'écriture en vol est attendue : + // sinon elle la ferait renaître, contenu et entrée d'index. + self.persister.forget(id); + self.persister.flush().await; + let next = { + let _g = self.persister.sink().index_lock.lock().await; + let mut idx = self.read_index().await; + remove_from(&mut idx, id); + self.save_index(&idx).await; + idx + }; + if let Err(e) = self.store.del(BK_CHAT, &conv_key(id)).await { + eprintln!("[conv] contenu de {id} non effacé : {e}"); + } + self.hooks.dropped(id).await; + if id + != self + .ensure_active_locked() + .await + .map_err(SessionError::Store)? + { + return Ok(()); + } + match next.first() { + // Surtout pas new_conv : il réenregistrerait la supprimée. + None => self.create_locked().await.map(|_| ()), + Some(m) => self.activate(&m.id.clone()).await, + } + } +} diff --git a/rust/loki-rs/src/store.rs b/rust/loki-rs/src/store.rs index 3697c0d..d9c466a 100644 --- a/rust/loki-rs/src/store.rs +++ b/rust/loki-rs/src/store.rs @@ -127,6 +127,31 @@ impl Store { } } + /// `getStoreBytesErr` : lit et ouvre une valeur scellée. Absente, ou + /// illisible (mémoire verrouillée, autre clé) → `Ok(None)` ; `Err` + /// seulement si la base n'a pas pu être lue (« rien » et « pas pu lire » + /// ne veulent pas dire la même chose). + pub async fn get_open(&self, bucket: &str, key: &str) -> Result>, String> { + match self.get(bucket, key).await? { + Some(raw) if !raw.is_empty() => Ok(self.open(raw).await.ok()), + _ => Ok(None), + } + } + + /// `putStoreBytes` : scelle puis écrit. Mémoire verrouillée : rien n'est + /// écrit (jamais en clair). + pub async fn put_seal( + &self, + bucket: &str, + key: &str, + plain: Vec, + ) -> Result<(), CryptError> { + let sealed = self.seal(plain).await?; + self.put(bucket, key, sealed) + .await + .map_err(CryptError::Other) + } + /// Exécute un appel d'outil par le Go (`/internal/v1/tool/exec`) : même /// fonction que sa boucle de chat. `caps` : objet JSON des capacités du /// tour (champs Go : Agent, Internet, Mem, Code, Role, ComputerUse). diff --git a/rust/loki-rs/tests/lot6.rs b/rust/loki-rs/tests/lot6.rs index e425f09..f1b9dca 100644 --- a/rust/loki-rs/tests/lot6.rs +++ b/rust/loki-rs/tests/lot6.rs @@ -73,7 +73,7 @@ fn opened(go: &fakego::FakeGo, bucket: &str, key: &str) -> Option> { #[tokio::test] async fn discussion_scellee_contenu_et_index() { let (st, go, _stop, _t) = store().await; - let p = Persister::new(StoreSink { store: st }); + let p = Persister::new(StoreSink::new(st)); let body = ConvState { seq: 3, ..Default::default() @@ -114,7 +114,7 @@ async fn discussion_scellee_contenu_et_index() { #[tokio::test] async fn memoire_verrouillee_rien_en_clair() { let (st, go, _stop, _t) = store().await; - let p = Persister::new(StoreSink { store: st }); + let p = Persister::new(StoreSink::new(st)); *go.locked.lock().unwrap() = true; p.enqueue_tool_res(ToolResJob { key: "c1.ab".into(), @@ -139,7 +139,7 @@ async fn memoire_verrouillee_rien_en_clair() { async fn index_illisible_jamais_reecrit() { let (st, go, _stop, _t) = store().await; go.set(BK_CHAT, CK_INDEX, "SCELLE:pas du json"); - let p = Persister::new(StoreSink { store: st }); + let p = Persister::new(StoreSink::new(st)); p.persist(snap("c1", p.next_seq(), "t", b"{}".to_vec())) .await; assert_eq!(opened(&go, BK_CHAT, "conv:c1").unwrap(), b"{}"); @@ -241,3 +241,177 @@ async fn moteur_tenu_par_une_tache_go() { .unwrap(); lease.release().await; } + +// ---- plusieurs discussions ---- + +use loki_rs::chat::conv::Conversation; +use loki_rs::chat::sessions::{SessionError, SessionHooks, Sessions}; +use std::sync::Mutex as StdMutex; + +#[derive(Default, Clone)] +struct Crochets(Arc>>); + +impl SessionHooks for Crochets { + async fn dropped(&self, id: &str) { + self.0.lock().unwrap().push(format!("dropped {id}")); + } + async fn switched(&self, id: &str) { + self.0.lock().unwrap().push(format!("switched {id}")); + } +} + +async fn sessions() -> ( + Sessions, + Crochets, + Arc, + tokio::sync::oneshot::Sender<()>, + tempfile::TempDir, +) { + let (st, go, stop, t) = store().await; + let h = Crochets::default(); + ( + Sessions::new(st, Conversation::new(), h.clone()), + h, + go, + stop, + t, + ) +} + +fn dire(s: &Sessions, texte: &str) { + let c = s.conv(); + c.append_delta( + c.epoch(), + serde_json::from_value(serde_json::json!({"user": texte})).unwrap(), + ); + c.with_state(|st| { + st.messages.push(loki_rs::chat::types::Message { + role: "user".into(), + content: Some(serde_json::json!(texte)), + ..Default::default() + }) + }); +} + +fn textes(s: &Sessions) -> Vec { + s.conv().with_state(|st| { + st.messages + .iter() + .filter_map(|m| m.content.as_ref()?.as_str().map(str::to_string)) + .collect() + }) +} + +#[tokio::test] +async fn demarrage_sur_base_neuve() { + let (s, _h, go, _stop, _t) = sessions().await; + s.load().await.unwrap(); + let id = s.conv().id(); + assert!(id.starts_with('c')); + assert_eq!(go.get(BK_CHAT, "active").unwrap(), id); + let (list, active) = s.list().await.unwrap(); + assert_eq!(active, id); + assert_eq!(list.len(), 1); + assert_eq!(list[0].project, "generale"); + // Rechargé : même discussion, rien de recréé. + s.load().await.unwrap(); + assert_eq!(s.read_index().await.len(), 1); +} + +#[tokio::test] +async fn demarrage_reprend_le_fil_unique() { + let (s, _h, go, _stop, _t) = sessions().await; + go.set( + BK_CHAT, + "conversation", + r#"{"messages":[{"role":"user","content":"ancien fil"}],"log":null,"seq":0,"ctx_used":0}"#, + ); + s.load().await.unwrap(); + assert_eq!(textes(&s), ["ancien fil"]); + assert!( + go.get(BK_CHAT, "conversation").is_some(), + "clé d'origine gardée" + ); +} + +#[tokio::test] +async fn demarrage_base_illisible_n_ecrit_rien() { + let (s, _h, go, _stop, _t) = sessions().await; + *go.broken.lock().unwrap() = true; + assert!(s.load().await.is_err()); + *go.broken.lock().unwrap() = false; + assert!( + go.store.lock().unwrap().is_empty(), + "écrit malgré la base illisible" + ); + assert_eq!(s.conv().id(), ""); +} + +#[tokio::test] +async fn creer_basculer_renommer() { + let (s, h, _go, _stop, _t) = sessions().await; + s.load().await.unwrap(); + let a = s.conv().id(); + dire(&s, "seulement dans A"); + let b = s.new_conv().await.unwrap(); + assert_ne!(a, b); + assert!(textes(&s).is_empty()); + dire(&s, "seulement dans B"); + s.switch(&a).await.unwrap(); + assert_eq!(textes(&s), ["seulement dans A"]); + s.switch(&b).await.unwrap(); + assert_eq!(textes(&s), ["seulement dans B"]); + assert_eq!( + *h.0.lock().unwrap(), + [format!("switched {a}"), format!("switched {b}")] + ); + // L'index porte les titres déduits. + let idx = s.read_index().await; + let titre = |id: &str| idx.iter().find(|m| m.id == id).unwrap().title.clone(); + assert_eq!(titre(&a), "seulement dans A"); + assert_eq!(s.switch("inconnue").await, Err(SessionError::NotFound)); + assert_eq!(s.switch("").await, Err(SessionError::Missing)); + s.rename(&a, " Mon fil ").await.unwrap(); + assert_eq!( + s.read_index() + .await + .iter() + .find(|m| m.id == a) + .unwrap() + .title, + "Mon fil" + ); + assert_eq!(s.rename(&a, " ").await, Err(SessionError::Missing)); +} + +#[tokio::test] +async fn supprimer_l_ouverte_puis_la_derniere() { + let (s, h, go, _stop, _t) = sessions().await; + s.load().await.unwrap(); + let a = s.conv().id(); + dire(&s, "fil A"); + s.persist().await; + let b = s.new_conv().await.unwrap(); + dire(&s, "fil B"); + s.persist_async().await; // une écriture en attente qui vise B + s.delete(&b).await.unwrap(); + assert!( + go.get(BK_CHAT, &format!("conv:{b}")).is_none(), + "contenu de B resté" + ); + assert!(!s.read_index().await.iter().any(|m| m.id == b)); + assert_eq!(s.conv().id(), a, "bascule sur la plus récente restante"); + assert_eq!(textes(&s), ["fil A"]); + assert!(h.0.lock().unwrap().contains(&format!("dropped {b}"))); + // B ne renaît pas, même après une écriture. + s.persist().await; + assert!(go.get(BK_CHAT, &format!("conv:{b}")).is_none()); + + s.delete(&a).await.unwrap(); + let c = s.conv().id(); + assert!(c != a && !c.is_empty(), "toujours une discussion active"); + let idx = s.read_index().await; + assert_eq!(idx.len(), 1); + assert_eq!(idx[0].id, c); + assert_eq!(s.delete("inconnue").await, Err(SessionError::NotFound)); +}