From 86f84c600361e4eb371e24946795f94ac41b78cc Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=8E=8B=E5=BE=B7=E5=AE=87?= <kvtodev@outlook.com>
Date: Tue, 15 Sep 2026 17:33:31 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3DirectProject=E4=BA=8B?=
 =?UTF-8?q?=E4=BB=B6=E9=98=9F=E5=88=97=E9=98=9F=E5=A4=B4=E6=B8=85=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

将运行态队列改为Vec加head索引\n仅清理可持久化事件的队头并保留未完成item事件\n增加定期compact避免已清理前缀长期占用内存
---
 .../src/agent/direct_thread_manager.rs        | 39 +++++++++++++------
 1 file changed, 28 insertions(+), 11 deletions(-)

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<StoredEvent>,
+    events: Vec<StoredEvent>,
+    head: usize,
     total_bytes: usize,
     active_items: HashSet<String>,
     unresolved_requests: HashSet<String>,
@@ -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::<Vec<_>>();
@@ -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::<Vec<_>>();
@@ -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) {