diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs index 8b9edebd3..071f65da4 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet}; use std::sync::{Mutex, OnceLock}; use uuid::Uuid; @@ -75,7 +75,8 @@ struct SubscriberState { #[derive(Clone, Debug)] struct ThreadState { next_seq: u64, - events: VecDeque, + events: Vec, + head: usize, total_bytes: usize, active_items: HashSet, unresolved_requests: HashSet, @@ -88,7 +89,8 @@ impl Default for ThreadState { fn default() -> Self { Self { next_seq: 0, - events: VecDeque::new(), + events: Vec::new(), + head: 0, total_bytes: 0, active_items: HashSet::new(), unresolved_requests: HashSet::new(), @@ -144,7 +146,7 @@ impl DirectThreadManager { .map(|value| value.len()) .unwrap_or_default(); thread.total_bytes = thread.total_bytes.saturating_add(bytes); - thread.events.push_back(StoredEvent { + thread.events.push(StoredEvent { event: event.clone(), bytes, cleanable, @@ -165,6 +167,7 @@ impl DirectThreadManager { let mut events = thread .events .iter() + .skip(thread.head) .filter(|stored| Self::is_bootstrap_event(thread, &stored.event, stored.cleanable)) .map(|stored| stored.event.clone()) .collect::>(); @@ -206,7 +209,7 @@ impl DirectThreadManager { .expect("subscriber checked above"); let oldest_seq = thread .events - .front() + .get(thread.head) .map(|stored| stored.event.seq) .unwrap_or(thread.next_seq.saturating_add(1)); if cursor.saturating_add(1) < oldest_seq { @@ -216,6 +219,7 @@ impl DirectThreadManager { let events = thread .events .iter() + .skip(thread.head) .filter(|stored| stored.event.seq > cursor) .map(|stored| stored.event.clone()) .collect::>(); @@ -235,7 +239,7 @@ impl DirectThreadManager { fn thread_debug(&self, thread_id: &str) -> Option<(usize, usize, usize)> { self.threads.get(thread_id).map(|thread| { ( - thread.events.len(), + thread.events.len().saturating_sub(thread.head), thread.total_bytes, thread.subscribers.len(), ) @@ -322,21 +326,34 @@ impl DirectThreadManager { loop { let can_pop = thread .events - .front() + .get(thread.head) .is_some_and(|stored| stored.event.seq <= min_cursor && stored.cleanable); if !can_pop { break; } - if let Some(stored) = thread.events.pop_front() { - thread.total_bytes = thread.total_bytes.saturating_sub(stored.bytes); - } + let bytes = thread + .events + .get(thread.head) + .map(|stored| stored.bytes) + .unwrap_or_default(); + thread.total_bytes = thread.total_bytes.saturating_sub(bytes); + thread.head = thread.head.saturating_add(1); + } + Self::compact(thread); + } + + fn compact(thread: &mut ThreadState) { + if thread.head >= 1024 && thread.head.saturating_mul(2) >= thread.events.len() { + thread.events.drain(..thread.head); + thread.head = 0; } } fn evict(&mut self, thread_id: &str) { loop { let over_limit = self.threads.get(thread_id).is_some_and(|thread| { - thread.events.len() > self.max_events || thread.total_bytes > self.max_bytes + thread.events.len().saturating_sub(thread.head) > self.max_events + || thread.total_bytes > self.max_bytes }); if !over_limit { if let Some(thread) = self.threads.get_mut(thread_id) {