diff --git a/rust/loki-rs/src/chat/journal.rs b/rust/loki-rs/src/chat/journal.rs index 2677102..f1267e1 100644 --- a/rust/loki-rs/src/chat/journal.rs +++ b/rust/loki-rs/src/chat/journal.rs @@ -1 +1,1193 @@ -//! Lot 6 : module en cours de portage. +//! Journal d'affichage du chat — port PUR de `chat_conversation.go` et des +//! trames SSE de `web_chat.go`. +//! +//! L'interface ne voit la discussion qu'à travers ce journal : une suite de +//! [`LogEvent`] (un delta SSE + un Seq monotone + un horodatage serveur en ms), +//! ajoutés par `appendDelta`, coalescés en fin de tour (`compactLog`) et +//! rejoués au rafraîchissement de la page (`subscribeSink`). +//! +//! Ce module ne contient QUE la logique déterministe : aucune goroutine, aucun +//! verrou, aucune base, aucun chiffrement. Ce que le Go lisait dans un état +//! global (discussion active `ckActive`, `c.CtxUsed`, horloge) arrive ici en +//! paramètre explicite, pour que l'acteur de conversation se bâtisse dessus : +//! +//! - ajout au journal : [`append_delta`] (plafond [`MAX_LOG_EVENTS`], [`cap_log`]) ; +//! - compaction de fin de tour : [`compact_log`] / [`compact_log_owned`] ; +//! - rejeu à l'abonnement : [`replay_plan`] (padding, reset, pagination, +//! rejeu coalescé, caught_up), bâti sur [`coalesce_replay`], [`history_cut`], +//! [`history_head`] ; +//! - direct : [`live_batch`] (dichotomie [`first_after`] + [`decorate_event`]) +//! et [`live_reset_plan`] (reset survenu pendant l'abonnement) ; +//! - trames SSE : [`sse_payload`], [`sse_data`], [`sse_data_frame`], +//! [`SSE_HEARTBEAT`], et le regroupement [`SseBatcher`]. +//! +//! Fidélité JSON : les deltas sont des `map[string]any` en Go ; l'UI lit les +//! clés par nom. Les valeurs (nombre / chaîne, présence / absence d'une clé) +//! sont reproduites à l'identique. Les nombres posés par ce module (`toks`, +//! `ts0`, `seq0`, `seq`, `ts`, `history_more`, `ctx_used`) sont des entiers. + +use std::time::Duration; + +use serde::Serialize; +use serde_json::{Map, Value}; + +use super::types::LogEvent; + +/// Un delta d'affichage (`map[string]any` en Go). +pub type Delta = Map; + +/// Plafond du journal d'AFFICHAGE (pas de la vue modèle), `maxLogEvents` en +/// Go. La coalescence de fin de tour réduit un tour terminé à une poignée +/// d'événements ; la marge sert à absorber UN tour en cours, streamé token par +/// token, sans en tronquer le début avant sa compaction. +pub const MAX_LOG_EVENTS: usize = 200_000; + +/// Échanges rejoués à l'ouverture d'une session (reset) pour un client qui +/// gère la pagination de l'historique (`historyTailDefault`). +pub const HISTORY_TAIL_DEFAULT: i64 = 20; + +/// Padding d'amorçage anti-buffering envoyé en tout premier : 2048 fois « · » +/// (U+00B7, 2 octets en UTF-8). Force un proxy (Cloudflare) à passer en mode +/// streaming tout de suite. Le client ignore la clé `pad`. +pub const PAD_OPEN_REPEAT: usize = 2048; + +/// Padding envoyé APRÈS caught_up : 16384 fois « · ». Ce sont ses octets, et +/// non le dernier vrai message, qui restent coincés dans le tampon du proxy. +pub const PAD_CAUGHT_UP_REPEAT: usize = 16384; + +/// Battement de cœur SSE : un commentaire, ignoré par le parseur navigateur, +/// sans contenu (donc rien à chiffrer). +pub const SSE_HEARTBEAT: &[u8] = b": ping\n\n"; + +/// Période du battement de cœur (4 s : borne le temps qu'un dernier bout de +/// flux reste coincé dans un tampon de proxy). +pub const SSE_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(4); + +/// Taille d'un envoi groupé en octets (`sseBatchBytes`, 128 Kio). +pub const SSE_BATCH_BYTES: usize = 128 << 10; + +/// Taille d'un envoi groupé en événements (`sseBatchEvents`). +pub const SSE_BATCH_EVENTS: usize = 256; + +// --------------------------------------------------------------------------- +// Lecture des champs d'un delta texte +// --------------------------------------------------------------------------- + +/// Lit un nombre JSON comme entier (troncature, comme `int(float64)` en Go). +fn num_i64(v: Option<&Value>) -> Option { + match v { + Some(Value::Number(n)) => n.as_i64().or_else(|| n.as_f64().map(|f| f as i64)), + _ => None, + } +} + +/// `evToks` : nombre de tokens porté par un événement texte — 1 pour un delta +/// brut (streaming), ou la valeur `toks` accumulée d'un événement coalescé. +pub fn ev_toks(d: &Delta) -> i64 { + num_i64(d.get("toks")).unwrap_or(1) +} + +/// `evTS0` : horodatage de DÉBUT d'un événement texte (présent seulement sur +/// les événements coalescés), sinon `fallback` (le TS de l'événement). +pub fn ev_ts0(d: &Delta, fallback: i64) -> i64 { + num_i64(d.get("ts0")).unwrap_or(fallback) +} + +/// `evSeq0` : seq de DÉBUT d'un bloc texte — sur un événement coalescé (dont +/// le Seq est celui du DERNIER delta fusionné), le seq du PREMIER delta ; sur +/// un delta brut, son propre Seq. Sert à poser `replace`. +pub fn ev_seq0(d: &Delta, fallback: i64) -> i64 { + num_i64(d.get("seq0")).unwrap_or(fallback) +} + +/// Clé texte d'un delta : `content` (chaîne) d'abord, puis +/// `reasoning_content` (chaîne). +fn text_key(d: &Delta) -> Option<&'static str> { + if matches!(d.get("content"), Some(Value::String(_))) { + Some("content") + } else if matches!(d.get("reasoning_content"), Some(Value::String(_))) { + Some("reasoning_content") + } else { + None + } +} + +/// `isTextDelta` : événement porteur de texte (content / reasoning_content). +pub fn is_text_delta(d: &Delta) -> bool { + text_key(d).is_some() +} + +fn text_of<'a>(d: &'a Delta, key: &str) -> &'a str { + match d.get(key) { + Some(Value::String(s)) => s, + _ => "", + } +} + +/// `tool_used` est un objet : (est un outil, done == true). +fn tool_state(d: &Delta) -> Option { + match d.get("tool_used") { + Some(Value::Object(tu)) => Some(tu.get("done") == Some(&Value::Bool(true))), + _ => None, + } +} + +fn obj1(k: &str, v: Value) -> Delta { + let mut m = Delta::new(); + m.insert(k.to_owned(), v); + m +} + +/// `{"pad": "·"×n}` — événement de padding ignoré par le client. +pub fn pad_event(n: usize) -> Delta { + obj1("pad", Value::String("·".repeat(n))) +} + +// --------------------------------------------------------------------------- +// Ajout au journal +// --------------------------------------------------------------------------- + +/// Partie pure de `appendDelta` : incrémente `seq`, ajoute l'événement +/// horodaté `ts` (ms, `time.Now().UnixMilli()` en Go) et applique le plafond +/// [`MAX_LOG_EVENTS`]. Rend le Seq attribué. +/// +/// Le contrôle d'epoch (delta d'un tour périmé jeté après un reset) et le +/// réveil des abonnés restent à l'acteur. +pub fn append_delta(log: &mut Vec, seq: &mut i64, ts: i64, delta: Delta) -> i64 { + *seq += 1; + log.push(LogEvent { + seq: *seq, + ts, + delta, + }); + cap_log(log, MAX_LOG_EVENTS); + *seq +} + +/// Troncature du journal à ses `max` DERNIERS événements (les plus vieux +/// partent). En Go, `Log[len-max:]` est O(1) ; ici le `drain` décale le +/// vecteur — coût payé seulement quand le plafond est atteint. +pub fn cap_log(log: &mut Vec, max: usize) { + if log.len() > max { + let extra = log.len() - max; + log.drain(..extra); + } +} + +/// Premier indice dont le Seq dépasse `last` (le journal est trié par Seq +/// croissant) : la dichotomie du direct (`sort.Search` en Go), au lieu d'un +/// balayage complet à chaque token. +pub fn first_after(log: &[LogEvent], last: i64) -> usize { + log.partition_point(|e| e.seq <= last) +} + +// --------------------------------------------------------------------------- +// Compaction de fin de tour +// --------------------------------------------------------------------------- + +/// Bloc texte en cours de fusion. +struct TextAcc { + key: &'static str, + buf: String, + seq: i64, + ts: i64, + toks: i64, + ts0: i64, + seq0: i64, +} + +impl TextAcc { + fn start(key: &'static str, ev: &LogEvent) -> Self { + TextAcc { + key, + buf: String::new(), + seq: ev.seq, + ts: ev.ts, + toks: 0, + ts0: ev_ts0(&ev.delta, ev.ts), + seq0: ev_seq0(&ev.delta, ev.seq), + } + } + + fn add(&mut self, ev: &LogEvent) { + self.buf.push_str(text_of(&ev.delta, self.key)); + self.seq = ev.seq; + self.ts = ev.ts; + self.toks += ev_toks(&ev.delta); + } +} + +/// `compactLog` sur un journal possédé (pas de copie des événements non +/// fusionnés) : la forme de `compactLogLocked`. +/// +/// Coalesce les suites d'événements content / reasoning_content en UN +/// événement chacune, `{clé: texte, toks: somme, ts0: premier, seq0: premier}` +/// avec le Seq/TS du DERNIER delta fusionné. Des `tool_used`, seul le +/// `done=true` reste (état final) ; on lui retire l'aperçu navigateur `shot` +/// (le delta devient alors `{tool_used}` seul). Un journal de moins de deux +/// événements est rendu tel quel. +pub fn compact_log_owned(log: Vec) -> Vec { + if log.len() < 2 { + return log; + } + let mut out = Vec::with_capacity(log.len()); + let mut acc: Option = None; + let flush = |acc: &mut Option, out: &mut Vec| { + if let Some(a) = acc.take() { + let mut d = Delta::new(); + d.insert(a.key.to_owned(), Value::String(a.buf)); + d.insert("toks".to_owned(), a.toks.into()); + d.insert("ts0".to_owned(), a.ts0.into()); + d.insert("seq0".to_owned(), a.seq0.into()); + out.push(LogEvent { + seq: a.seq, + ts: a.ts, + delta: d, + }); + } + }; + for mut ev in log { + if let Some(done) = tool_state(&ev.delta) { + flush(&mut acc, &mut out); + if done { + let has_shot = matches!(ev.delta.get("tool_used"), + Some(Value::Object(tu)) if tu.contains_key("shot")); + if has_shot && let Some(Value::Object(mut cp)) = ev.delta.remove("tool_used") { + cp.remove("shot"); + ev.delta = obj1("tool_used", Value::Object(cp)); + } + out.push(ev); + } + continue; + } + let Some(key) = text_key(&ev.delta) else { + flush(&mut acc, &mut out); + out.push(ev); + continue; + }; + if acc.as_ref().is_some_and(|a| a.key != key) { + flush(&mut acc, &mut out); + } + acc.get_or_insert_with(|| TextAcc::start(key, &ev)).add(&ev); + } + flush(&mut acc, &mut out); + out +} + +/// `compactLog` : compaction PURE, sans toucher au journal passé (sert à +/// écrire en plein tour la forme qu'aura le journal à la fin du tour). +pub fn compact_log(log: &[LogEvent]) -> Vec { + compact_log_owned(log.to_vec()) +} + +// --------------------------------------------------------------------------- +// Rejeu +// --------------------------------------------------------------------------- + +/// `{"seq", "ts"}` puis les clés du delta (qui l'emportent en cas de doublon). +fn flatten(ev: &LogEvent) -> Delta { + let mut m = Delta::new(); + m.insert("seq".to_owned(), ev.seq.into()); + m.insert("ts".to_owned(), ev.ts.into()); + for (k, v) in &ev.delta { + m.insert(k.clone(), v.clone()); + } + m +} + +/// `decorateEvent` : aplatit un événement pour l'émission SSE et pose +/// `replace: true` quand le client a DÉJÀ vu le début du bloc texte (`from` +/// tombe dedans) : le texte envoyé est le bloc entier, à substituer. +pub fn decorate_event(ev: &LogEvent, from: i64) -> Delta { + let mut out = flatten(ev); + if is_text_delta(&ev.delta) && ev_seq0(&ev.delta, ev.seq) <= from { + out.insert("replace".to_owned(), Value::Bool(true)); + } + out +} + +/// `coalesceReplay` : rejeu léger des événements de Seq > `from`. +/// +/// Les deltas texte consécutifs d'un même bloc deviennent UN événement +/// `{clé, seq, ts, ts0, toks, seq0}` (+ `replace` si `seq0 <= from`). Des +/// outils, seul l'état final compte : chaque `done=true` est émis ; le dernier +/// `tool_used` non terminé est retenu (sans être émis sur un événement +/// non-outil, sinon un outil apparaîtrait deux fois) et n'est émis qu'en fin +/// de fil si aucun `done` ne l'a remplacé. Le reste passe aplati tel quel. +pub fn coalesce_replay(events: &[LogEvent], from: i64) -> Vec { + let mut out = Vec::new(); + let mut acc: Option = None; + let mut pending_tool: Option = None; + let flush = |acc: &mut Option, out: &mut Vec| { + if let Some(a) = acc.take() { + let mut m = Delta::new(); + m.insert(a.key.to_owned(), Value::String(a.buf)); + m.insert("seq".to_owned(), a.seq.into()); + m.insert("ts".to_owned(), a.ts.into()); + m.insert("ts0".to_owned(), a.ts0.into()); + m.insert("toks".to_owned(), a.toks.into()); + m.insert("seq0".to_owned(), a.seq0.into()); + if a.seq0 <= from { + m.insert("replace".to_owned(), Value::Bool(true)); + } + out.push(m); + } + }; + for ev in events { + if ev.seq <= from { + continue; + } + let tool = tool_state(&ev.delta); + if let Some(key) = text_key(&ev.delta) { + if acc.as_ref().is_some_and(|a| a.key != key) { + flush(&mut acc, &mut out); + } + acc.get_or_insert_with(|| TextAcc::start(key, ev)).add(ev); + continue; + } + flush(&mut acc, &mut out); + let m = flatten(ev); + match tool { + Some(true) => { + pending_tool = None; + out.push(m); + } + Some(false) => pending_tool = Some(m), + None => out.push(m), + } + } + flush(&mut acc, &mut out); + out.extend(pending_tool); + out +} + +/// `historyCut` : où couper le rejeu pour ne garder que les `tail` derniers +/// échanges (un échange commence à un événement portant la clé `user`). +/// Rend le Seq à partir duquel rejouer (exclu) et le nombre d'échanges +/// masqués ; `(0, 0)` = tout rejouer. +pub fn history_cut(log: &[LogEvent], tail: i64) -> (i64, i64) { + if tail <= 0 { + return (0, 0); + } + let mut seen = 0; + for i in (0..log.len()).rev() { + if !log[i].delta.contains_key("user") { + continue; + } + seen += 1; + if seen == tail { + let hidden = log[..i] + .iter() + .filter(|e| e.delta.contains_key("user")) + .count() as i64; + if hidden == 0 { + return (0, 0); + } + return (log[i].seq - 1, hidden); + } + } + (0, 0) +} + +/// `historyHead` : événements qui annoncent un historique tronqué — +/// `{history_more: hidden}`, puis les derniers deltas `mode` et `criteria` +/// posés dans la partie masquée (Seq <= `cut`), tels quels. +pub fn history_head(log: &[LogEvent], cut: i64, hidden: i64) -> Vec { + let mut out = vec![obj1("history_more", hidden.into())]; + let mut mode = None; + let mut crit = None; + for ev in log { + if ev.seq > cut { + break; + } + if ev.delta.contains_key("mode") { + mode = Some(&ev.delta); + } + if ev.delta.contains_key("criteria") { + crit = Some(&ev.delta); + } + } + out.extend([mode, crit].into_iter().flatten().cloned()); + out +} + +/// Ce qu'un abonné doit émettre, dans l'ordre, et le Seq à partir duquel +/// suivre le direct ensuite. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct ReplayPlan { + pub events: Vec, + /// Dernier Seq couvert : le direct reprend aux événements de Seq > `last`. + pub last: i64, +} + +fn reset_event(cur_id: &str) -> Delta { + let mut m = obj1("reset", Value::Bool(true)); + m.insert("id".to_owned(), cur_id.into()); + m +} + +/// Partie pure du début de `subscribeSink` (étapes 0 et 1) sur un instantané +/// du journal. +/// +/// - `from` : dernier Seq vu par le client ; `tail` : échanges à rejouer pour +/// un client paginé (`< 0` = ancien client, tout rejouer) ; `conv_id` : +/// discussion affichée par le client (vide = inconnue) ; +/// - `cur_id` : discussion active (`getStr(bkChat, ckActive)` en Go) ; +/// - `ctx_used` : taille du contexte (`c.CtxUsed`, que le Go relit APRÈS le +/// rejeu : l'acteur passe la valeur courante). +/// +/// Ordre émis : pad 2048, `{reset, id}` si le curseur vient d'une autre +/// discussion (id différent, ou `from` au-delà du dernier Seq), en-tête de +/// pagination, rejeu coalescé, `{caught_up, id}`, `{ctx_used}` si > 0, pad +/// 16384. Le client s'arrête au premier envoi raté. +pub fn replay_plan( + snapshot: &[LogEvent], + from: i64, + tail: i64, + conv_id: &str, + cur_id: &str, + ctx_used: i64, +) -> ReplayPlan { + let mut events = vec![pad_event(PAD_OPEN_REPEAT)]; + let mut from = from; + let mut stale = from > 0 && !conv_id.is_empty() && conv_id != cur_id; + if let Some(last) = snapshot.last() + && from > last.seq + { + from = 0; + stale = stale || !conv_id.is_empty(); + } + if stale { + from = 0; + events.push(reset_event(cur_id)); + } + if from == 0 && tail > 0 { + let (cut, hidden) = history_cut(snapshot, tail); + if hidden > 0 { + from = cut; + events.extend(history_head(snapshot, cut, hidden)); + } + } + let mut last = from; + for ev in coalesce_replay(snapshot, from) { + if let Some(s) = ev.get("seq").and_then(Value::as_i64) { + last = s; + } + events.push(ev); + } + let mut caught = obj1("caught_up", Value::Bool(true)); + caught.insert("id".to_owned(), cur_id.into()); + events.push(caught); + if ctx_used > 0 { + events.push(obj1("ctx_used", ctx_used.into())); + } + events.push(pad_event(PAD_CAUGHT_UP_REPEAT)); + ReplayPlan { events, last } +} + +/// Reset survenu pendant l'abonnement (epoch changé) : `{reset, id}`, puis, +/// chez un client paginé (`tail >= 0`), l'en-tête des +/// [`HISTORY_TAIL_DEFAULT`] derniers échanges suivi de `{ctx_used}` (émis même +/// à 0, comme en Go). `last` = Seq de coupe, ou 0 pour tout suivre. +pub fn live_reset_plan(log: &[LogEvent], tail: i64, cur_id: &str, ctx_used: i64) -> ReplayPlan { + let mut events = vec![reset_event(cur_id)]; + let mut last = 0; + if tail >= 0 { + let (cut, hidden) = history_cut(log, HISTORY_TAIL_DEFAULT); + if hidden > 0 { + last = cut; + events.extend(history_head(log, cut, hidden)); + events.push(obj1("ctx_used", ctx_used.into())); + } + } + ReplayPlan { events, last } +} + +/// Un tour de la boucle du direct : les événements de Seq > `last`, décorés +/// avec `last` comme repère de `replace` (un bloc fusionné par la compaction +/// de fin de tour que le client suivait en direct arrive avec un Seq plus +/// grand que le sien). `last` du résultat = Seq du dernier événement ; +/// inchangé (et `events` vide) s'il n'y a rien de neuf — l'acteur attend. +pub fn live_batch(log: &[LogEvent], last: i64) -> ReplayPlan { + let pending = &log[first_after(log, last)..]; + ReplayPlan { + events: pending.iter().map(|ev| decorate_event(ev, last)).collect(), + last: pending.last().map_or(last, |ev| ev.seq), + } +} + +// --------------------------------------------------------------------------- +// Trames SSE +// --------------------------------------------------------------------------- + +#[derive(Serialize)] +struct SseChoice<'a> { + delta: &'a Delta, +} + +#[derive(Serialize)] +struct SsePayload<'a> { + choices: [SseChoice<'a>; 1], +} + +/// Échappement HTML de `encoding/json` : `<`, `>`, `&`, U+2028 et U+2029 +/// deviennent `<`, `>`, `&`, `
`, `
`. Ces octets +/// n'apparaissent que dans des chaînes JSON : un remplacement global est sûr. +fn go_html_escape(b: Vec) -> Vec { + let needs = b.iter().any(|&c| matches!(c, b'<' | b'>' | b'&')) + || b.windows(3) + .any(|w| w[0] == 0xE2 && w[1] == 0x80 && (w[2] == 0xA8 || w[2] == 0xA9)); + if !needs { + return b; + } + let mut out = Vec::with_capacity(b.len() + 32); + let mut i = 0; + while i < b.len() { + match b[i] { + b'<' => out.extend_from_slice(b"\\u003c"), + b'>' => out.extend_from_slice(b"\\u003e"), + b'&' => out.extend_from_slice(b"\\u0026"), + 0xE2 if i + 2 < b.len() && b[i + 1] == 0x80 && matches!(b[i + 2], 0xA8 | 0xA9) => { + out.extend_from_slice(if b[i + 2] == 0xA8 { + b"\\u2028" + } else { + b"\\u2029" + }); + i += 3; + continue; + } + c => out.push(c), + } + i += 1; + } + out +} + +/// Charge JSON d'un événement du flux : `{"choices":[{"delta":obj}]}`, clés +/// triées et échappement HTML comme `json.Marshal`. C'est ce que le relais +/// E2E scelle (un sceau par événement). +pub fn sse_payload(obj: &Delta) -> Vec { + let p = SsePayload { + choices: [SseChoice { delta: obj }], + }; + // Une Map de Value se sérialise toujours. + go_html_escape(serde_json::to_vec(&p).unwrap_or_default()) +} + +/// Ligne SSE `data: \n\n` (corps JSON en clair, ou base64 du sceau E2E). +pub fn sse_data(body: &[u8]) -> Vec { + let mut out = Vec::with_capacity(body.len() + 8); + out.extend_from_slice(b"data: "); + out.extend_from_slice(body); + out.extend_from_slice(b"\n\n"); + out +} + +/// Trame en clair d'un événement (`frame` de `handleChat`). +pub fn sse_data_frame(obj: &Delta) -> Vec { + sse_data(&sse_payload(obj)) +} + +/// Regroupement des trames d'un abonné (`sseStream` sans l'écrivain) : un lot +/// = une écriture. Un client en retard peut trouver des milliers +/// d'événements en attente ; on écrit par tranches de [`SSE_BATCH_BYTES`] +/// octets ou [`SSE_BATCH_EVENTS`] événements. +/// +/// Les lots sont RENDUS à l'appelant (qui écrit et pousse sous le verrou du +/// battement de cœur) : `queue` rend `Some(lot)` quand le tampon déborde, +/// `flush` rend ce qui attend. `emit` du Go = `queue` puis `flush`. Le lot +/// rendu emporte son tampon : la règle Go « un gros tampon (> 4 lots) n'est +/// pas gardé à vie » est satisfaite d'office. +#[derive(Debug, Default)] +pub struct SseBatcher { + buf: Vec, + n: usize, +} + +impl SseBatcher { + pub fn new() -> Self { + Self::default() + } + + /// Une trame de plus ; `Some(lot)` = tampon plein, à écrire tout de suite. + pub fn queue(&mut self, frame: &[u8]) -> Option> { + self.buf.extend_from_slice(frame); + self.n += 1; + if self.buf.len() >= SSE_BATCH_BYTES || self.n >= SSE_BATCH_EVENTS { + return self.flush(); + } + None + } + + /// Tout ce qui attend, en un lot ; `None` si rien. + pub fn flush(&mut self) -> Option> { + if self.buf.is_empty() { + return None; + } + self.n = 0; + Some(std::mem::take(&mut self.buf)) + } + + /// Nombre de trames en attente. + pub fn pending(&self) -> usize { + self.n + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn d(v: Value) -> Delta { + match v { + Value::Object(m) => m, + _ => panic!("objet attendu"), + } + } + + fn ev(seq: i64, v: Value) -> LogEvent { + LogEvent { + seq, + ts: 0, + delta: d(v), + } + } + + fn evts(seq: i64, ts: i64, v: Value) -> LogEvent { + LogEvent { + seq, + ts, + delta: d(v), + } + } + + /// Conversation de test : journal + Seq, comme `newTestConv`. + #[derive(Default)] + struct Conv { + log: Vec, + seq: i64, + } + + impl Conv { + fn add(&mut self, v: Value) { + append_delta(&mut self.log, &mut self.seq, 0, d(v)); + } + } + + fn seqs(evs: &[Delta]) -> Vec { + evs.iter() + .filter_map(|m| m.get("seq").and_then(Value::as_i64)) + .collect() + } + + // --- chat_conversation_test.go --- + + // TestSubscribeReplayAndLive : rejeu de l'existant PUIS direct. + #[test] + fn subscribe_replay_and_live() { + let mut c = Conv::default(); + c.add(json!({"user": "salut"})); + c.add(json!({"content": "bon"})); + let plan = replay_plan(&c.log, 0, -1, "", "", 0); + assert_eq!(seqs(&plan.events), vec![1, 2]); + c.add(json!({"content": "jour"})); + let live = live_batch(&c.log, plan.last); + assert_eq!(seqs(&live.events), vec![3]); + assert_eq!(live.last, 3); + } + + // TestSubscribeFromOffset : from=N → seulement ce qui est plus récent. + #[test] + fn subscribe_from_offset() { + let mut c = Conv::default(); + c.add(json!({"user": "a"})); + c.add(json!({"user": "b"})); + let plan = replay_plan(&c.log, 1, -1, "", "", 0); + assert_eq!(seqs(&plan.events).first(), Some(&2)); + } + + // TestResetNotifiesSubscribers (partie abonné) : un reset pendant + // l'abonnement pousse {reset:true} ; le journal vidé, on repart de 0. + #[test] + fn reset_notifies_subscribers() { + let plan = live_reset_plan(&[], -1, "", 0); + assert!(plan.events.iter().any(|m| m.contains_key("reset"))); + assert_eq!(plan.last, 0); + } + + // TestReplayMarksReplaceWhenClientSawPartOfBlock + #[test] + fn replay_marks_replace_when_client_saw_part_of_block() { + let mut c = Conv::default(); + c.add(json!({"user": "salut"})); // seq 1 + c.add(json!({"content": "Oui, "})); // seq 2 + c.add(json!({"content": "j'ai "})); // seq 3 + c.add(json!({"content": "accès"})); // seq 4 + c.log = compact_log_owned(std::mem::take(&mut c.log)); + + let out = coalesce_replay(&c.log, 3); + assert_eq!(out.len(), 1, "{out:?}"); + assert_eq!(out[0]["content"], json!("Oui, j'ai accès")); + assert_eq!(out[0].get("replace"), Some(&json!(true))); + + let out = coalesce_replay(&c.log, 1); + assert_eq!(out.len(), 1); + assert!(!out[0].contains_key("replace")); + } + + // TestLiveSelectionSurJournalTronque + #[test] + fn live_selection_sur_journal_tronque() { + let mut c = Conv { + log: vec![ + ev(500, json!({"user": "a"})), + ev(501, json!({"user": "b"})), + ev(502, json!({"user": "c"})), + ], + seq: 502, + }; + for (from, first) in [(0, 500), (500, 501), (501, 502)] { + let plan = replay_plan(&c.log, from, -1, "", "", 0); + assert_eq!(seqs(&plan.events).first(), Some(&first), "from={from}"); + // La dichotomie du direct tombe au même endroit. + assert_eq!(c.log[first_after(&c.log, from)].seq, first); + } + // Client déjà à jour : rien à rejouer, mais le direct suit. + let plan = replay_plan(&c.log, 502, -1, "", "", 0); + assert!(seqs(&plan.events).is_empty()); + assert!(live_batch(&c.log, plan.last).events.is_empty()); + c.add(json!({"content": "suite"})); + assert_eq!(seqs(&live_batch(&c.log, plan.last).events), vec![503]); + } + + // TestCompactLogCoalesceTools + #[test] + fn compact_log_coalesce_tools() { + let log = vec![ + ev(1, json!({"user": "édite"})), + ev(2, json!({"tool_used": {"name": "edit", "label": "f"}})), + ev( + 3, + json!({"tool_used": {"name": "edit", "label": "f", "typing": true, "body": "a"}}), + ), + ev( + 4, + json!({"tool_used": {"name": "edit", "label": "f", "result": "ok", "done": true}}), + ), + ev(5, json!({"content": "fini"})), + ]; + let out = compact_log_owned(log); + let mut n_tool = 0; + for e in &out { + if let Some(Value::Object(tu)) = e.delta.get("tool_used") { + n_tool += 1; + assert_eq!( + tu.get("done"), + Some(&json!(true)), + "un tool_used non terminé a survécu à la compaction" + ); + } + } + assert_eq!(n_tool, 1); + } + + // TestCoalesceReplayNoDoubleTool + #[test] + fn coalesce_replay_no_double_tool() { + let events = vec![ + ev(1, json!({"tool_used": {"name": "bash", "label": "ls"}})), + ev(2, json!({"stats": {"gen_tokens": 3}})), + ev( + 3, + json!({"tool_used": {"name": "bash", "label": "ls", "result": "x", "done": true}}), + ), + ]; + let n = coalesce_replay(&events, 0) + .iter() + .filter(|m| m.contains_key("tool_used")) + .count(); + assert_eq!(n, 1); + } + + // TestSubscribeStaleFromCrossSession + #[test] + fn subscribe_stale_from_cross_session() { + let mut c = Conv::default(); + c.add(json!({"user": "premier message"})); + c.add(json!({"content": "réponse"})); + let plan = replay_plan(&c.log, 9999, -1, "", "", 0); + let seen: String = plan + .events + .iter() + .filter_map(|m| m.get("user").and_then(Value::as_str)) + .collect(); + assert!(seen.contains("premier message"), "reçu : {seen:?}"); + } + + // --- chat_history_tail_test.go --- + + fn five_exchanges() -> Vec { + let mut c = Conv::default(); + for _ in 0..5 { + c.add(json!({"user": "q"})); + c.add(json!({"content": "r"})); + c.add(json!({"turn_done": true})); + } + c.log + } + + // TestHistoryCut + #[test] + fn history_cut_garde_les_n_derniers() { + let log = five_exchanges(); + assert_eq!(history_cut(&log, 2), (9, 3)); + assert_eq!(history_cut(&log, 5), (0, 0)); + assert_eq!(history_cut(&log, 0), (0, 0)); + let evs = coalesce_replay(&log, 9); + assert_eq!(evs[0]["user"], json!("q")); + assert_eq!(evs[0]["seq"], json!(10)); + } + + // TestHistoryHeadCarriesMode + #[test] + fn history_head_carries_mode() { + let log = vec![ + ev(1, json!({"mode": "code"})), + ev(2, json!({"user": "q1"})), + ev(3, json!({"user": "q2"})), + ]; + let head = history_head(&log, 2, 1); + assert_eq!(head.len(), 2, "{head:?}"); + assert_eq!(head[0]["history_more"], json!(1)); + assert_eq!(head[1]["mode"], json!("code")); + } + + // TestSubscribeTailPaginates + #[test] + fn subscribe_tail_paginates() { + let mut c = Conv::default(); + for _ in 0..5 { + c.add(json!({"user": "q"})); + c.add(json!({"content": "r"})); + } + let collect = |tail: i64| { + let plan = replay_plan(&c.log, 0, tail, "", "", 0); + let mut more = None; + let mut users = 0; + for m in &plan.events { + if let Some(v) = m.get("history_more") { + more = Some(v.clone()); + } + if m.contains_key("user") { + users += 1; + } + if m.contains_key("caught_up") { + break; + } + } + (more, users) + }; + assert_eq!(collect(2), (Some(json!(3)), 2)); + assert_eq!(collect(-1), (None, 5)); + } + + // --- cu_autoshot_log_test.go --- + + // TestCompactLogRetireLAperçuNavigateur + #[test] + fn compact_log_retire_l_apercu_navigateur() { + let log = vec![ + ev( + 1, + json!({"tool_used": {"name": "browser_click", "done": false}}), + ), + ev( + 2, + json!({"tool_used": {"name": "browser_click", "done": true, "result": "ok", "shot": "abc"}}), + ), + ]; + let out = compact_log(&log); + assert_eq!(out.len(), 1); + let tu = out[0].delta["tool_used"].as_object().unwrap(); + assert!( + !tu.contains_key("shot"), + "l'aperçu aurait dû quitter le journal" + ); + assert_eq!(tu["result"], json!("ok")); + assert_eq!(tu["name"], json!("browser_click")); + assert!( + log[1].delta["tool_used"] + .as_object() + .unwrap() + .contains_key("shot"), + "le journal d'origine ne doit pas être modifié en place" + ); + } + + // --- chat_persist_test.go --- + + // TestCompactLogPurEgalFinDeTour + #[test] + fn compact_log_pur_egal_fin_de_tour() { + let log = vec![ + evts(1, 10, json!({"user": "q"})), + evts(2, 11, json!({"reasoning_content": "je ", "toks": 1})), + evts(3, 12, json!({"reasoning_content": "pense", "toks": 1})), + evts(4, 13, json!({"tool_used": {"name": "bash", "done": false}})), + evts(5, 14, json!({"tool_used": {"name": "bash", "done": true}})), + evts(6, 15, json!({"content": "bon", "toks": 1})), + evts(7, 16, json!({"content": "jour", "toks": 1})), + ]; + let orig = log.clone(); + let got = compact_log(&log); + assert_eq!(log, orig, "compact_log a modifié le journal d'origine"); + assert_eq!(got, compact_log_owned(log.clone())); + assert_eq!(got.len(), 4); + // Valeurs exactes du bloc de réflexion fusionné (entiers, comme en Go). + assert_eq!((got[1].seq, got[1].ts), (3, 12)); + assert_eq!( + Value::Object(got[1].delta.clone()), + json!({"reasoning_content": "je pense", "toks": 2, "ts0": 11, "seq0": 2}) + ); + } + + // TestPersistEnPleinTourEcritLeJournalCompacte (partie pure : la forme + // compactée écrite en plein tour, journal d'origine intact). + #[test] + fn persist_en_plein_tour_ecrit_le_journal_compacte() { + let log: Vec = (0..50) + .map(|i| ev(i + 1, json!({"content": "x", "toks": 1}))) + .collect(); + let st = compact_log(&log); + assert_eq!(st.len(), 1); + assert_eq!(st[0].delta["content"], json!("x".repeat(50))); + assert_eq!(log.len(), 50); + } + + // --- stream_hotpath_test.go --- + + // TestSSEStreamBatches : un lot = une écriture, ordre gardé, gros lot découpé. + #[test] + fn sse_stream_batches() { + let frame = |i: usize| format!("data: {i}\n\n").into_bytes(); + let mut writes: Vec> = Vec::new(); + let mut s = SseBatcher::new(); + for i in 1..=10 { + writes.extend(s.queue(&frame(i))); + } + assert_eq!(writes.len(), 0, "écritures avant flush"); + writes.extend(s.flush()); + assert_eq!(writes.len(), 1); + for i in 11..=10 + 2 * SSE_BATCH_EVENTS { + writes.extend(s.queue(&frame(i))); + } + // emit = queue puis flush + writes.extend(s.queue(&frame(11 + 2 * SSE_BATCH_EVENTS))); + writes.extend(s.flush()); + assert_eq!(writes.len(), 4); + let want: Vec = (1..=11 + 2 * SSE_BATCH_EVENTS).flat_map(frame).collect(); + assert_eq!(writes.concat(), want, "ordre des événements perdu"); + assert_eq!(s.pending(), 0); + assert!(s.flush().is_none()); + } + + // TestSubscribeSinkLiveBatchOrder (partie pure) : le direct rend les + // événements dans l'ordre des seq. + #[test] + fn subscribe_sink_live_batch_order() { + let mut c = Conv::default(); + let plan = replay_plan(&c.log, 0, -1, "", "", 0); + assert!(plan.events.iter().any(|m| m.contains_key("caught_up"))); + const N: i64 = 500; + for _ in 0..N { + c.add(json!({"content": "t"})); + } + let live = live_batch(&c.log, plan.last); + assert_eq!(seqs(&live.events), (1..=N).collect::>()); + assert_eq!(live.last, N); + } + + // --- Tests propres au port --- + + #[test] + fn plafond_du_journal() { + let mut log = Vec::new(); + let mut seq = 0; + for _ in 0..5 { + seq += 1; + log.push(ev(seq, json!({"content": "t"}))); + cap_log(&mut log, 3); + } + assert_eq!(log.iter().map(|e| e.seq).collect::>(), vec![3, 4, 5]); + let mut big = Vec::new(); + let mut s = 0; + assert_eq!( + append_delta(&mut big, &mut s, 42, d(json!({"user": "a"}))), + 1 + ); + assert_eq!((big[0].seq, big[0].ts), (1, 42)); + } + + #[test] + fn compaction_journal_court_inchange() { + // Moins de deux événements : rendu tel quel (ni toks ni ts0). + let log = vec![ev(1, json!({"content": "a"}))]; + assert_eq!(compact_log(&log), log); + let log = vec![ev(1, json!({"tool_used": {"name": "x"}}))]; + assert_eq!(compact_log(&log), log); + } + + #[test] + fn compaction_relit_les_nombres_du_json() { + // Journal relu depuis la base : toks/ts0/seq0 en flottants. + let log = vec![ + evts( + 5, + 50, + json!({"content": "ab", "toks": 2.0, "ts0": 40.0, "seq0": 3.0}), + ), + evts(6, 60, json!({"content": "c"})), + ]; + let out = compact_log(&log); + assert_eq!( + Value::Object(out[0].delta.clone()), + json!({"content": "abc", "toks": 3, "ts0": 40, "seq0": 3}) + ); + assert_eq!((out[0].seq, out[0].ts), (6, 60)); + } + + #[test] + fn decoration_du_direct() { + // Bloc fusionné seq0=2..seq=4 que le client suivait (dernier vu : 3). + let e = evts( + 4, + 9, + json!({"content": "abc", "toks": 3, "ts0": 7, "seq0": 2}), + ); + let m = decorate_event(&e, 3); + assert_eq!( + Value::Object(m), + json!({"seq": 4, "ts": 9, "content": "abc", "toks": 3, "ts0": 7, "seq0": 2, "replace": true}) + ); + assert!(!decorate_event(&e, 1).contains_key("replace")); + // Non-texte : jamais de replace. + let u = ev(2, json!({"user": "q"})); + assert!(!decorate_event(&u, 5).contains_key("replace")); + } + + #[test] + fn plan_de_rejeu_complet() { + let log = vec![ + evts(1, 1, json!({"user": "q"})), + evts(2, 2, json!({"content": "a"})), + evts(3, 3, json!({"content": "b"})), + ]; + let p = replay_plan(&log, 0, -1, "c1", "c1", 77); + let ev = &p.events; + assert_eq!(ev[0]["pad"].as_str().unwrap().chars().count(), 2048); + assert_eq!(ev[0]["pad"].as_str().unwrap().len(), 4096); + assert_eq!( + Value::Object(ev[1].clone()), + json!({"seq": 1, "ts": 1, "user": "q"}) + ); + assert_eq!( + Value::Object(ev[2].clone()), + json!({"content": "ab", "seq": 3, "ts": 3, "ts0": 2, "toks": 2, "seq0": 2}) + ); + assert_eq!( + Value::Object(ev[3].clone()), + json!({"caught_up": true, "id": "c1"}) + ); + assert_eq!(Value::Object(ev[4].clone()), json!({"ctx_used": 77})); + assert_eq!(ev[5]["pad"].as_str().unwrap().chars().count(), 16384); + assert_eq!(ev.len(), 6); + assert_eq!(p.last, 3); + } + + #[test] + fn plan_de_rejeu_discussion_perimee() { + let log = vec![ev(1, json!({"user": "q"})), ev(2, json!({"content": "r"}))]; + // Id différent : reset puis tout le fil. + let p = replay_plan(&log, 1, -1, "ancienne", "neuve", 0); + assert_eq!( + Value::Object(p.events[1].clone()), + json!({"reset": true, "id": "neuve"}) + ); + assert_eq!(seqs(&p.events), vec![1, 2]); + // Même id, curseur au-delà : tout le fil ET reset (id connu). + let p = replay_plan(&log, 50, -1, "neuve", "neuve", 0); + assert!(p.events[1].contains_key("reset")); + assert_eq!(seqs(&p.events), vec![1, 2]); + // Curseur au-delà sans id : tout le fil, sans reset. + let p = replay_plan(&log, 50, -1, "", "neuve", 0); + assert!(!p.events.iter().any(|m| m.contains_key("reset"))); + assert_eq!(seqs(&p.events), vec![1, 2]); + // Journal vide, from > 0 : rien, last = from. + let p = replay_plan(&[], 7, -1, "", "x", 0); + assert_eq!(p.last, 7); + assert_eq!(p.events.len(), 3); + } + + #[test] + fn plan_de_reset_pagine() { + let mut c = Conv::default(); + c.add(json!({"criteria": ["a"]})); + for _ in 0..(HISTORY_TAIL_DEFAULT + 2) { + c.add(json!({"user": "q"})); + } + let p = live_reset_plan(&c.log, 0, "id", 0); + assert_eq!( + Value::Object(p.events[0].clone()), + json!({"reset": true, "id": "id"}) + ); + assert_eq!(p.events[1]["history_more"], json!(2)); + assert_eq!(p.events[2]["criteria"], json!(["a"])); + assert_eq!(Value::Object(p.events[3].clone()), json!({"ctx_used": 0})); + assert_eq!(p.last, 3); + // Ancien client (tail < 0) : reset seul. + let p = live_reset_plan(&c.log, -1, "id", 0); + assert_eq!(p.events.len(), 1); + assert_eq!(p.last, 0); + } + + #[test] + fn coalescence_outil_interrompu_et_bascule_texte() { + let log = vec![ + ev(1, json!({"reasoning_content": "r"})), + ev(2, json!({"content": "a"})), + ev(3, json!({"tool_used": {"name": "bash", "done": false}})), + ev(4, json!({"content": "b"})), + ]; + let out = coalesce_replay(&log, 0); + let keys: Vec<_> = out + .iter() + .map(|m| { + ["reasoning_content", "content", "tool_used"] + .into_iter() + .find(|k| m.contains_key(*k)) + .unwrap() + }) + .collect(); + // L'outil jamais terminé n'est émis qu'une fois, en fin de fil. + assert_eq!( + keys, + vec!["reasoning_content", "content", "content", "tool_used"] + ); + } + + #[test] + fn trames_sse_comme_le_go() { + let m = d(json!({"content": "\u{2028}", "seq": 3})); + assert_eq!( + sse_data_frame(&m), + b"data: {\"choices\":[{\"delta\":{\"content\":\"\\u003ca \\u0026 b\\u003e\\u2028\",\"seq\":3}}]}\n\n" + .to_vec() + ); + assert_eq!( + sse_payload(&d(json!({"x": 1}))), + br#"{"choices":[{"delta":{"x":1}}]}"#.to_vec() + ); + assert_eq!(SSE_HEARTBEAT, b": ping\n\n"); + } +}