Merge feature/ticket34-chatbridge-scrollback-bounded into develop (#34)
Scrollback de ChatBridge borné (double cap chunks + octets, drop par la tête) pour stopper la croissance mémoire du buffer de replay. Aucun changement DTO. QA vert 19/19. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -33,6 +33,11 @@ use domain::ports::ReplyEvent;
|
|||||||
|
|
||||||
use crate::dto::ReplyChunk;
|
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
|
/// Per-session transport state: the current attach generation, its output
|
||||||
/// channel, and the conversation scrollback recorded so far.
|
/// channel, and the conversation scrollback recorded so far.
|
||||||
struct ChatEntry {
|
struct ChatEntry {
|
||||||
@ -42,9 +47,9 @@ struct ChatEntry {
|
|||||||
/// when the session has been recorded (scrollback kept) but no view is
|
/// when the session has been recorded (scrollback kept) but no view is
|
||||||
/// currently attached — the pump then only appends to the scrollback.
|
/// currently attached — the pump then only appends to the scrollback.
|
||||||
channel: Option<Channel<ReplyChunk>>,
|
channel: Option<Channel<ReplyChunk>>,
|
||||||
/// Every chunk routed for this session, in order — the conversation
|
/// Recent chunks retained for reattach, bounded by chunk count and estimated
|
||||||
/// scrollback replayed on re-attach. Bounded only by the conversation length
|
/// byte size; older chunks are dropped from the head. This is a transport
|
||||||
/// (a turn count), like the PTY ring buffer is bounded by its byte capacity.
|
/// replay buffer, not durable conversation history.
|
||||||
scrollback: Vec<ReplyChunk>,
|
scrollback: Vec<ReplyChunk>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -159,6 +164,7 @@ impl ChatBridge {
|
|||||||
return false;
|
return false;
|
||||||
};
|
};
|
||||||
entry.scrollback.push(chunk.clone());
|
entry.scrollback.push(chunk.clone());
|
||||||
|
trim_scrollback(entry);
|
||||||
match &entry.channel {
|
match &entry.channel {
|
||||||
Some(channel) => channel.send(chunk).is_ok(),
|
Some(channel) => channel.send(chunk).is_ok(),
|
||||||
None => false,
|
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
|
/// 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
|
/// 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).
|
/// (kept here so the pump and any test share one mapping).
|
||||||
|
|||||||
@ -13,7 +13,9 @@ use std::sync::{Arc, Mutex};
|
|||||||
|
|
||||||
use tauri::ipc::{Channel, InvokeResponseBody};
|
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 app_tauri_lib::dto::ReplyChunk;
|
||||||
use domain::ids::SessionId;
|
use domain::ids::SessionId;
|
||||||
use domain::ports::ReplyEvent;
|
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)
|
// chunk_from_event — exhaustive mapping (zone 6)
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@ -311,6 +321,131 @@ fn scrollback_of_unknown_session_is_empty() {
|
|||||||
assert!(bridge.scrollback(&sid()).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<_>>(),
|
||||||
|
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)
|
// close_agent_session: unregister purges channel AND scrollback (zone 4)
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user