From 74c2d5105cd7fbdc50adf7cd20efdb90be9ac084 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 15 Oct 2026 10:05:33 +0000
Subject: [PATCH] =?UTF-8?q?Lot=206=20:=20la=20discussion=20en=20m=C3=A9moi?=
 =?UTF-8?q?re=20port=C3=A9e=20en=20Rust=20(chat::conv)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Port de l'état de Conversation et de subscribeSink : état persisté et
discussion chargée, epoch (reset, bascule), verrou du tour et son arrêt,
ajout au journal avec rejet des deltas d'un tour périmé, coalescence de fin
de tour, instantané numéroté sous le même verrou, /api/chat/state.

Subscription rend les lots à émettre : rejeu (padding, reset si le curseur
vient d'une autre discussion, pagination, caught_up, contexte), puis le
direct, avec {reset} et l'en-tête paginé quand l'epoch change. Réveil par
compteur de version marqué lu avant la relecture de l'état : aucun réveil
perdu.

Tests Go de l'abonnement portés.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv
---
 rust/loki-rs/src/chat/conv.rs | 482 ++++++++++++++++++++++++++++++++++
 rust/loki-rs/src/chat/mod.rs  |   3 +
 2 files changed, 485 insertions(+)
 create mode 100644 rust/loki-rs/src/chat/conv.rs

diff --git a/rust/loki-rs/src/chat/conv.rs b/rust/loki-rs/src/chat/conv.rs
new file mode 100644
index 0000000..5cf5c66
--- /dev/null
+++ b/rust/loki-rs/src/chat/conv.rs
@@ -0,0 +1,482 @@
+//! La discussion en mémoire — port de l'état de `Conversation`
+//! (`chat_conversation.go`) et de l'abonnement `subscribeSink`.
+//!
+//! Une seule discussion ouverte, partagée par tous les appareils. Elle porte :
+//! - l'état persisté ([`ConvState`] : vue du modèle, journal d'affichage,
+//!   Seq, contexte mesuré, bloc projet) et l'identifiant de la discussion
+//!   chargée ;
+//! - l'epoch, incrémenté à chaque reset ou bascule : il invalide les abonnés
+//!   (qui reçoivent `{reset}` et rejouent) et les deltas d'un tour périmé ;
+//! - le verrou du tour (`Generating`) et de quoi l'arrêter.
+//!
+//! Les abonnés ([`Subscription`]) lisent le journal : rejeu coalescé, puis
+//! direct. Ils sont réveillés par un compteur de version (`watch`), marqué lu
+//! AVANT de relire l'état : aucun réveil n'est perdu, contrairement à une
+//! simple notification qui peut partir entre la lecture et l'attente.
+//!
+//! Hors de ce module : la boucle d'un tour, la file d'attente des messages
+//! envoyés pendant un tour, les discussions multiples (index, bascules) et la
+//! base — l'appelant charge et enregistre (voir [`super::persist`]).
+
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use serde_json::{Map, Value};
+use tokio::sync::watch;
+
+use super::journal::{self, Delta, live_batch, live_reset_plan, replay_plan};
+use super::persist::{ConvSnap, ConvState, PersistSink, Persister, decode_state, take_snapshot};
+
+/// Arrêt du tour en cours (`c.cancel`).
+pub type CancelTurn = Box<dyn FnOnce() + Send>;
+
+/// Un tour est déjà en cours (`ErrBusy`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Busy;
+
+impl std::fmt::Display for Busy {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("génération en cours")
+    }
+}
+
+#[derive(Default)]
+struct Inner {
+    st: ConvState,
+    /// Discussion chargée (`ckActive`) : celle qu'annoncent `{reset}` et
+    /// `{caught_up}`, et sous laquelle s'enregistrent les instantanés.
+    id: String,
+    generating: bool,
+    epoch: i64,
+    cancel: Option<CancelTurn>,
+}
+
+/// La discussion ouverte. À partager dans un [`Arc`] (abonnés, tour).
+pub struct Conversation {
+    inner: Mutex<Inner>,
+    version: watch::Sender<u64>,
+}
+
+impl Default for Conversation {
+    fn default() -> Self {
+        Conversation {
+            inner: Mutex::new(Inner::default()),
+            version: watch::Sender::new(0),
+        }
+    }
+}
+
+fn now_ms() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+impl Conversation {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self::default())
+    }
+
+    fn lock(&self) -> MutexGuard<'_, Inner> {
+        self.inner.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    /// Réveille les abonnés (`cond.Broadcast`).
+    fn bump(&self) {
+        self.version.send_modify(|v| *v = v.wrapping_add(1));
+    }
+
+    /// Identifiant de la discussion chargée.
+    pub fn id(&self) -> String {
+        self.lock().id.clone()
+    }
+
+    pub fn epoch(&self) -> i64 {
+        self.lock().epoch
+    }
+
+    /// `isGenerating`.
+    pub fn is_generating(&self) -> bool {
+        self.lock().generating
+    }
+
+    /// `loadFrom` : remplace l'état par la discussion `id` (`body` vide = fil
+    /// neuf), arrête le tour en cours et invalide les abonnés, qui vident
+    /// l'écran et rejouent le nouveau fil.
+    pub fn load_from(&self, id: &str, body: &[u8]) {
+        self.stop();
+        let st = if body.is_empty() {
+            ConvState::default()
+        } else {
+            decode_state(body)
+        };
+        {
+            let mut g = self.lock();
+            g.st = st;
+            g.id = id.to_string();
+            g.generating = false;
+            g.cancel = None;
+            g.epoch += 1;
+        }
+        self.bump();
+    }
+
+    /// `Reset` : fil vide pour tous les appareils, tour arrêté et verrou
+    /// rendu — même un tour coincé (moteur redémarré sous ses pieds) : c'est
+    /// le geste qui débloque. L'enregistrement et les fichiers de la
+    /// discussion restent à l'appelant.
+    pub fn reset(&self) {
+        self.stop();
+        {
+            let mut g = self.lock();
+            g.st = ConvState::default();
+            g.generating = false;
+            g.cancel = None;
+            g.epoch += 1;
+        }
+        self.bump();
+    }
+
+    /// `Stop` : arrête le tour en cours (le verrou est rendu par sa fin).
+    pub fn stop(&self) {
+        let cancel = self.lock().cancel.take();
+        if let Some(c) = cancel {
+            c();
+        }
+    }
+
+    /// Prend le verrou du tour ; rend l'epoch du tour, que ses deltas portent.
+    pub fn begin_turn(&self, cancel: CancelTurn) -> Result<i64, Busy> {
+        let mut g = self.lock();
+        if g.generating {
+            return Err(Busy);
+        }
+        g.generating = true;
+        g.cancel = Some(cancel);
+        Ok(g.epoch)
+    }
+
+    /// Fin d'un tour : le journal du tour est coalescé (`compactLogLocked`)
+    /// et le verrou rendu. Un tour dont l'epoch est périmé (reset, bascule)
+    /// ne touche à rien : le verrou a déjà été rendu, peut-être repris.
+    pub fn end_turn(&self, epoch: i64) -> bool {
+        {
+            let mut g = self.lock();
+            if g.epoch != epoch {
+                return false;
+            }
+            let log = std::mem::take(&mut g.st.log);
+            g.st.log = journal::compact_log_owned(log);
+            g.generating = false;
+            g.cancel = None;
+        }
+        self.bump();
+        true
+    }
+
+    /// `appendDelta` : journalise un événement d'affichage et réveille les
+    /// abonnés. Un delta d'un epoch périmé appartient à l'ancien fil : jeté.
+    pub fn append_delta(&self, epoch: i64, delta: Delta) -> bool {
+        {
+            let mut g = self.lock();
+            if g.epoch != epoch {
+                return false;
+            }
+            let Inner { st, .. } = &mut *g;
+            journal::append_delta(&mut st.log, &mut st.seq, now_ms(), delta);
+        }
+        self.bump();
+        true
+    }
+
+    /// Accès à l'état (vue du modèle, contexte…) sous le verrou, pour le tour.
+    /// Ne réveille personne : le journal se modifie par [`Self::append_delta`].
+    pub fn with_state<R>(&self, f: impl FnOnce(&mut ConvState) -> R) -> R {
+        f(&mut self.lock().st)
+    }
+
+    /// Instantané à enregistrer, numéroté sous le même verrou que l'état.
+    pub fn snapshot<S: PersistSink>(&self, p: &Persister<S>, project: &str) -> ConvSnap {
+        let g = self.lock();
+        take_snapshot(&g.st, g.generating, &g.id, p.next_seq(), project)
+    }
+
+    /// `state` : instantané léger pour `/api/chat/state`. La tâche Go qui
+    /// tiendrait le moteur est ajoutée par l'appelant.
+    pub fn state_json(&self) -> Map<String, Value> {
+        let g = self.lock();
+        let turns =
+            g.st.log
+                .iter()
+                .filter(|ev| ev.delta.contains_key("user"))
+                .count();
+        let mut m = Map::new();
+        m.insert("seq".into(), g.st.seq.into());
+        m.insert("generating".into(), g.generating.into());
+        m.insert("ctx_used".into(), g.st.ctx_used.into());
+        m.insert("turns".into(), turns.into());
+        m
+    }
+
+    /// Abonnement au journal. `from` : dernier Seq vu par le client ; `tail` :
+    /// échanges à rejouer pour un client paginé (`< 0` : tout) ; `conv_id` :
+    /// discussion affichée par le client (vide : inconnue).
+    pub fn subscribe(self: &Arc<Self>, from: i64, tail: i64, conv_id: &str) -> Subscription {
+        Subscription {
+            conv: self.clone(),
+            rx: self.version.subscribe(),
+            from,
+            tail,
+            conv_id: conv_id.to_string(),
+            epoch: 0,
+            last: 0,
+            started: false,
+        }
+    }
+}
+
+/// Un abonné au journal (`subscribeSink`). Chaque [`Self::next`] rend le lot
+/// suivant à émettre, dans l'ordre : d'abord le rejeu (padding, reset éventuel,
+/// pagination, rejeu coalescé, `caught_up`, contexte, padding), puis les
+/// événements du direct dès qu'il y en a. Lâcher l'abonnement l'arrête.
+pub struct Subscription {
+    conv: Arc<Conversation>,
+    rx: watch::Receiver<u64>,
+    from: i64,
+    tail: i64,
+    conv_id: String,
+    epoch: i64,
+    last: i64,
+    started: bool,
+}
+
+impl Subscription {
+    /// Lot suivant ; attend s'il n'y a rien de neuf. Jamais vide.
+    pub async fn next(&mut self) -> Vec<Delta> {
+        loop {
+            // Marqué lu AVANT de relire l'état : un ajout qui suit réveille.
+            self.rx.borrow_and_update();
+            if let Some(events) = self.poll() {
+                return events;
+            }
+            if self.rx.changed().await.is_err() {
+                // Discussion détruite : plus rien ne viendra.
+                std::future::pending::<()>().await;
+            }
+        }
+    }
+
+    fn poll(&mut self) -> Option<Vec<Delta>> {
+        let g = self.conv.lock();
+        if !self.started {
+            self.started = true;
+            // Le journal est copié sous le verrou ; le rejeu se bâtit hors.
+            let snapshot = g.st.log.clone();
+            let (id, ctx_used) = (g.id.clone(), g.st.ctx_used);
+            self.epoch = g.epoch;
+            drop(g);
+            let plan = replay_plan(
+                &snapshot,
+                self.from,
+                self.tail,
+                &self.conv_id,
+                &id,
+                ctx_used,
+            );
+            self.last = plan.last;
+            return Some(plan.events);
+        }
+        if g.epoch != self.epoch {
+            self.epoch = g.epoch;
+            let plan = live_reset_plan(&g.st.log, self.tail, &g.id, g.st.ctx_used);
+            self.last = plan.last;
+            return Some(plan.events);
+        }
+        let plan = live_batch(&g.st.log, self.last);
+        if plan.events.is_empty() {
+            return None;
+        }
+        self.last = plan.last;
+        Some(plan.events)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::chat::persist::{CommitOutcome, ResGate, ToolResJob};
+    use serde_json::json;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::time::Duration;
+
+    fn d(v: Value) -> Delta {
+        let Value::Object(m) = v else {
+            panic!("objet attendu")
+        };
+        m
+    }
+
+    async fn next(sub: &mut Subscription) -> Vec<Delta> {
+        tokio::time::timeout(Duration::from_secs(5), sub.next())
+            .await
+            .expect("aucun lot")
+    }
+
+    fn seqs(evs: &[Delta]) -> Vec<i64> {
+        evs.iter()
+            .filter_map(|e| e.get("seq").and_then(Value::as_i64))
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn rejeu_puis_direct() {
+        let c = Conversation::new();
+        c.append_delta(0, d(json!({"user": "salut"})));
+        c.append_delta(0, d(json!({"content": "bon"})));
+        let mut sub = c.subscribe(0, -1, "");
+        let first = next(&mut sub).await;
+        assert_eq!(seqs(&first), vec![1, 2]);
+        assert!(
+            first
+                .iter()
+                .any(|e| e.get("caught_up") == Some(&json!(true)))
+        );
+        c.append_delta(0, d(json!({"content": "jour"})));
+        assert_eq!(seqs(&next(&mut sub).await), vec![3]);
+    }
+
+    #[tokio::test]
+    async fn abonne_depuis_un_seq() {
+        let c = Conversation::new();
+        c.append_delta(0, d(json!({"user": "a"})));
+        c.append_delta(0, d(json!({"user": "b"})));
+        let mut sub = c.subscribe(1, -1, "");
+        assert_eq!(seqs(&next(&mut sub).await), vec![2]);
+    }
+
+    #[tokio::test]
+    async fn curseur_d_une_autre_discussion_rejoue_tout() {
+        let c = Conversation::new();
+        c.load_from("c1", b"");
+        let e = c.epoch();
+        c.append_delta(e, d(json!({"user": "premier message"})));
+        c.append_delta(e, d(json!({"content": "réponse"})));
+        let mut sub = c.subscribe(9999, -1, "c0");
+        let evs = next(&mut sub).await;
+        assert_eq!(evs[1], d(json!({"reset": true, "id": "c1"})));
+        assert!(
+            evs.iter()
+                .any(|e| e.get("user") == Some(&json!("premier message")))
+        );
+    }
+
+    #[tokio::test]
+    async fn reset_prevenu_et_deltas_perimes_jetes() {
+        let c = Conversation::new();
+        let e0 = c.epoch();
+        c.append_delta(e0, d(json!({"user": "x"})));
+        let mut sub = c.subscribe(0, -1, "");
+        next(&mut sub).await;
+        c.reset();
+        let evs = next(&mut sub).await;
+        assert_eq!(evs[0].get("reset"), Some(&json!(true)));
+        assert!(!c.append_delta(e0, d(json!({"content": "fantôme"}))));
+        assert_eq!(c.state_json()["seq"], 0);
+        assert!(c.append_delta(c.epoch(), d(json!({"user": "nouveau"}))));
+        assert_eq!(seqs(&next(&mut sub).await), vec![1]);
+    }
+
+    #[tokio::test]
+    async fn bascule_charge_le_fil_et_previent() {
+        let c = Conversation::new();
+        c.load_from("a", b"");
+        c.append_delta(c.epoch(), d(json!({"user": "dans A"})));
+        let mut sub = c.subscribe(0, 20, "a");
+        next(&mut sub).await;
+        let body = ConvState {
+            log: vec![crate::chat::types::LogEvent {
+                seq: 1,
+                ts: 5,
+                delta: d(json!({"user": "dans B"})),
+            }],
+            seq: 1,
+            ctx_used: 77,
+            ..Default::default()
+        }
+        .to_json();
+        c.load_from("b", &body);
+        let evs = next(&mut sub).await;
+        assert_eq!(evs[0], d(json!({"reset": true, "id": "b"})));
+        assert_eq!(c.id(), "b");
+        assert_eq!(c.state_json()["ctx_used"], 77);
+        // Fil court : rien de masqué, le direct rejoue tout depuis 0.
+        assert_eq!(seqs(&next(&mut sub).await), vec![1]);
+    }
+
+    #[tokio::test]
+    async fn verrou_du_tour() {
+        let c = Conversation::new();
+        let stopped = Arc::new(AtomicBool::new(false));
+        let s = stopped.clone();
+        let e = c
+            .begin_turn(Box::new(move || s.store(true, Ordering::SeqCst)))
+            .unwrap();
+        assert!(c.is_generating());
+        assert_eq!(c.begin_turn(Box::new(|| {})), Err(Busy));
+        c.append_delta(e, d(json!({"content": "Bon"})));
+        c.append_delta(e, d(json!({"content": "jour"})));
+        assert!(c.end_turn(e));
+        assert!(!c.is_generating());
+        // Fin de tour : journal coalescé.
+        c.with_state(|st| {
+            assert_eq!(st.log.len(), 1);
+            assert_eq!(st.log[0].delta["content"], "Bonjour");
+        });
+        // Reset pendant un tour : arrêté, verrou rendu, fin périmée sans effet.
+        let s = stopped.clone();
+        let e = c
+            .begin_turn(Box::new(move || s.store(true, Ordering::SeqCst)))
+            .unwrap();
+        c.reset();
+        assert!(stopped.load(Ordering::SeqCst));
+        assert!(!c.is_generating());
+        let e2 = c.begin_turn(Box::new(|| {})).unwrap();
+        assert!(!c.end_turn(e), "fin d'un tour périmé acceptée");
+        assert!(c.is_generating());
+        assert!(c.end_turn(e2));
+    }
+
+    #[tokio::test]
+    async fn etat_et_instantane() {
+        struct Rien;
+        impl PersistSink for Rien {
+            async fn commit(
+                &self,
+                _: Vec<ConvSnap>,
+                _: Vec<ToolResJob>,
+                _: ResGate,
+            ) -> CommitOutcome {
+                CommitOutcome::default()
+            }
+        }
+        let p = Persister::new(Rien);
+        let c = Conversation::new();
+        c.load_from("c9", b"");
+        let e = c.begin_turn(Box::new(|| {})).unwrap();
+        c.append_delta(e, d(json!({"user": "q"})));
+        c.append_delta(e, d(json!({"content": "a"})));
+        c.append_delta(e, d(json!({"content": "b"})));
+        c.with_state(|st| st.ctx_used = 12);
+        assert_eq!(
+            Value::Object(c.state_json()),
+            json!({"seq": 3, "generating": true, "ctx_used": 12, "turns": 1})
+        );
+        let s1 = c.snapshot(&p, "generale");
+        let s2 = c.snapshot(&p, "generale");
+        assert_eq!((s1.id.as_str(), s1.turns), ("c9", 1));
+        assert!(s2.seq > s1.seq);
+        // En plein tour, l'instantané porte le journal compacté.
+        assert_eq!(decode_state(&s1.body).log.len(), 2);
+    }
+}
diff --git a/rust/loki-rs/src/chat/mod.rs b/rust/loki-rs/src/chat/mod.rs
index 3dc6b37..27b4823 100644
--- a/rust/loki-rs/src/chat/mod.rs
+++ b/rust/loki-rs/src/chat/mod.rs
@@ -8,10 +8,13 @@
 //!   d'outils, statistiques) ;
 //! - [`compact`] : estimation du contexte et compactage de l'historique ;
 //! - [`journal`] : journal d'affichage (compactage, rejeu, pagination) ;
+//! - [`conv`] : la discussion en mémoire (état, epoch, verrou du tour,
+//!   abonnés) ;
 //! - [`persist`] : format en base, index des discussions, écrivain unique ;
 //! - [`lease`] : bail du moteur (verrou de génération partagé avec le Go).
 
 pub mod compact;
+pub mod conv;
 pub mod engine;
 pub mod journal;
 pub mod lease;