diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index a8d8525..99dfdee 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -33,6 +33,11 @@ use domain::ports::ReplyEvent; use crate::dto::ReplyChunk; +/// Maximum number of recent chat chunks retained for transport reattach replay. +pub const MAX_CHAT_SCROLLBACK_CHUNKS: usize = 2_000; +/// Maximum estimated bytes retained for transport reattach replay. +pub const MAX_CHAT_SCROLLBACK_BYTES: usize = 512 * 1024; + /// Per-session transport state: the current attach generation, its output /// channel, and the conversation scrollback recorded so far. struct ChatEntry { @@ -42,9 +47,9 @@ struct ChatEntry { /// when the session has been recorded (scrollback kept) but no view is /// currently attached — the pump then only appends to the scrollback. channel: Option>, - /// Every chunk routed for this session, in order — the conversation - /// scrollback replayed on re-attach. Bounded only by the conversation length - /// (a turn count), like the PTY ring buffer is bounded by its byte capacity. + /// Recent chunks retained for reattach, bounded by chunk count and estimated + /// byte size; older chunks are dropped from the head. This is a transport + /// replay buffer, not durable conversation history. scrollback: Vec, } @@ -159,6 +164,7 @@ impl ChatBridge { return false; }; entry.scrollback.push(chunk.clone()); + trim_scrollback(entry); match &entry.channel { Some(channel) => channel.send(chunk).is_ok(), None => false, @@ -172,6 +178,27 @@ impl ChatBridge { } } +fn trim_scrollback(entry: &mut ChatEntry) { + while !entry.scrollback.is_empty() + && (entry.scrollback.len() > MAX_CHAT_SCROLLBACK_CHUNKS + || scrollback_bytes(&entry.scrollback) > MAX_CHAT_SCROLLBACK_BYTES) + { + entry.scrollback.remove(0); + } +} + +fn scrollback_bytes(chunks: &[ReplyChunk]) -> usize { + chunks.iter().map(reply_chunk_bytes).sum() +} + +fn reply_chunk_bytes(chunk: &ReplyChunk) -> usize { + match chunk { + ReplyChunk::TextDelta { text } => text.len(), + ReplyChunk::ToolActivity { label } => label.len(), + ReplyChunk::Final { content } => content.len(), + } +} + /// Maps a domain [`ReplyEvent`] to its wire [`ReplyChunk`]. Pure translation, no /// I/O — the single point where the typed turn event becomes a serialisable chunk /// (kept here so the pump and any test share one mapping). diff --git a/crates/app-tauri/tests/chat_bridge.rs b/crates/app-tauri/tests/chat_bridge.rs index 88f8667..3645145 100644 --- a/crates/app-tauri/tests/chat_bridge.rs +++ b/crates/app-tauri/tests/chat_bridge.rs @@ -13,7 +13,9 @@ use std::sync::{Arc, Mutex}; use tauri::ipc::{Channel, InvokeResponseBody}; -use app_tauri_lib::chat::{chunk_from_event, ChatBridge}; +use app_tauri_lib::chat::{ + chunk_from_event, ChatBridge, MAX_CHAT_SCROLLBACK_BYTES, MAX_CHAT_SCROLLBACK_CHUNKS, +}; use app_tauri_lib::dto::ReplyChunk; use domain::ids::SessionId; use domain::ports::ReplyEvent; @@ -49,6 +51,14 @@ fn final_chunk(s: &str) -> ReplyChunk { } } +fn chunk_bytes(chunk: &ReplyChunk) -> usize { + match chunk { + ReplyChunk::TextDelta { text } => text.len(), + ReplyChunk::ToolActivity { label } => label.len(), + ReplyChunk::Final { content } => content.len(), + } +} + // --------------------------------------------------------------------------- // chunk_from_event — exhaustive mapping (zone 6) // --------------------------------------------------------------------------- @@ -311,6 +321,131 @@ fn scrollback_of_unknown_session_is_empty() { assert!(bridge.scrollback(&sid()).is_empty()); } +#[test] +fn scrollback_drops_oldest_chunks_when_chunk_cap_is_exceeded() { + let bridge = ChatBridge::new(); + let session = sid(); + let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + bridge.detach_if(&session, gen); + + for i in 0..(MAX_CHAT_SCROLLBACK_CHUNKS + 10) { + bridge.send_output(&session, delta(&format!("chunk-{i}"))); + } + + let scrollback = bridge.scrollback(&session); + assert_eq!(scrollback.len(), MAX_CHAT_SCROLLBACK_CHUNKS); + assert_eq!(scrollback.first(), Some(&delta("chunk-10"))); + assert_eq!( + scrollback.last(), + Some(&delta(&format!("chunk-{}", MAX_CHAT_SCROLLBACK_CHUNKS + 9))) + ); + assert_eq!( + scrollback.iter().take(3).cloned().collect::>(), + vec![delta("chunk-10"), delta("chunk-11"), delta("chunk-12")], + "remaining chunks keep their relative order and exact content" + ); +} + +#[test] +fn scrollback_drops_oldest_whole_chunks_when_byte_cap_is_exceeded() { + let bridge = ChatBridge::new(); + let session = sid(); + let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + bridge.detach_if(&session, gen); + let chunk_size = MAX_CHAT_SCROLLBACK_BYTES / 4; + + for i in 0..6 { + let text = format!("{i:02}:{}", "x".repeat(chunk_size - 3)); + bridge.send_output(&session, delta(&text)); + } + + let scrollback = bridge.scrollback(&session); + let retained_bytes: usize = scrollback.iter().map(chunk_bytes).sum(); + assert!( + retained_bytes <= MAX_CHAT_SCROLLBACK_BYTES, + "retained bytes {retained_bytes} must stay under cap {MAX_CHAT_SCROLLBACK_BYTES}" + ); + assert_eq!(scrollback.len(), 4); + assert!(matches!( + scrollback.first(), + Some(ReplyChunk::TextDelta { text }) if text.starts_with("02:") + )); + assert!(matches!( + scrollback.last(), + Some(ReplyChunk::TextDelta { text }) if text.starts_with("05:") + )); + assert!(scrollback.iter().all(|chunk| matches!( + chunk, + ReplyChunk::TextDelta { text } if text.len() == chunk_size + ))); +} + +#[test] +fn truncated_scrollback_replays_valid_unmodified_chunks_in_order() { + let bridge = ChatBridge::new(); + let session = sid(); + let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + bridge.detach_if(&session, gen); + + for i in 0..(MAX_CHAT_SCROLLBACK_CHUNKS + 3) { + bridge.send_output(&session, delta(&format!("delta-{i}"))); + } + bridge.send_output( + &session, + ReplyChunk::ToolActivity { + label: "tool".into(), + }, + ); + bridge.send_output(&session, final_chunk("done")); + + let scrollback = bridge.scrollback(&session); + assert_eq!(scrollback.len(), MAX_CHAT_SCROLLBACK_CHUNKS); + assert_eq!(scrollback[0], delta("delta-5")); + assert_eq!( + &scrollback[(MAX_CHAT_SCROLLBACK_CHUNKS - 2)..], + &[ + ReplyChunk::ToolActivity { + label: "tool".into() + }, + final_chunk("done") + ], + "chunks are retained whole and unmodified" + ); + + let replay_sink = Arc::new(Mutex::new(Vec::new())); + let replay = ChatBridge::new(); + let replay_session = sid(); + replay.register(replay_session, capturing_channel(Arc::clone(&replay_sink))); + for chunk in scrollback.clone() { + assert!(replay.send_output(&replay_session, chunk)); + } + assert_eq!(*replay_sink.lock().unwrap(), scrollback); +} + +#[test] +fn detach_keeps_scrollback_bounded_and_unregister_purges_it() { + let bridge = ChatBridge::new(); + let session = sid(); + let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + bridge.detach_if(&session, gen); + + for i in 0..(MAX_CHAT_SCROLLBACK_CHUNKS + 25) { + bridge.send_output(&session, delta(&format!("detached-{i}"))); + } + + assert_eq!(bridge.active_sessions(), 1); + assert_eq!( + bridge.scrollback(&session).len(), + MAX_CHAT_SCROLLBACK_CHUNKS, + "detached sessions keep bounded replay buffers" + ); + + bridge.unregister(&session); + + assert_eq!(bridge.active_sessions(), 0); + assert!(bridge.scrollback(&session).is_empty()); +} + // --------------------------------------------------------------------------- // close_agent_session: unregister purges channel AND scrollback (zone 4) // ---------------------------------------------------------------------------