diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent.rs b/apps/ai-game-creator-shell/src-tauri/src/agent.rs
index d084e6174..0e5076fef 100644
--- a/apps/ai-game-creator-shell/src-tauri/src/agent.rs
+++ b/apps/ai-game-creator-shell/src-tauri/src/agent.rs
@@ -20,6 +20,7 @@ mod direct_codex_user_item;
 mod direct_project_history;
 mod direct_project_turn_history;
 mod direct_runtime;
+mod direct_thread_manager;
 mod direct_tool_bridge;
 mod direct_tools_mcp;
 mod generation;
@@ -49,6 +50,7 @@ pub(crate) use direct_codex_user_item::*;
 pub(crate) use direct_project_history::*;
 pub(crate) use direct_project_turn_history::*;
 pub(crate) use direct_runtime::*;
+pub(crate) use direct_thread_manager::*;
 pub(crate) use direct_tool_bridge::*;
 pub(crate) use direct_tools_mcp::*;
 pub(crate) use generation::*;
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
new file mode 100644
index 000000000..647c3b9e2
--- /dev/null
+++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs
@@ -0,0 +1,507 @@
+//! DirectProject 运行态事件队列。
+//!
+//! 这个模块只维护 Thread Manager 的内存事实：每个 thread 一个全局事件序列，
+//! 每个 subscriber 一个受保护的消费游标。它不理解前端 reducer，也不负责 JSONL
+//! 持久化；调用方必须在完成 item 持久化成功后再追加对应完成事件。
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::collections::{HashMap, HashSet, VecDeque};
+use uuid::Uuid;
+
+const DEFAULT_MAX_EVENTS: usize = 8_192;
+const DEFAULT_MAX_BYTES: usize = 8 * 1024 * 1024;
+
+pub(crate) const SUBSCRIPTION_EXPIRED: &str = "SUBSCRIPTION_EXPIRED";
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct DirectThreadRawEvent {
+    pub(crate) seq: u64,
+    #[serde(rename = "type")]
+    pub(crate) event_type: String,
+    pub(crate) turn_id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) item_id: Option<String>,
+    pub(crate) payload: Value,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) struct DirectThreadRawEventDraft {
+    pub(crate) event_type: String,
+    pub(crate) turn_id: String,
+    pub(crate) item_id: Option<String>,
+    pub(crate) payload: Value,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct DirectThreadSubscriptionBootstrap {
+    pub(crate) subscription_id: String,
+    pub(crate) last_completed_item_id: Option<String>,
+    pub(crate) events: Vec<DirectThreadRawEvent>,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct DirectThreadConsumeResult {
+    pub(crate) events: Vec<DirectThreadRawEvent>,
+}
+
+#[derive(Clone, Debug)]
+struct StoredEvent {
+    event: DirectThreadRawEvent,
+    bytes: usize,
+    cleanable: bool,
+}
+
+#[derive(Clone, Debug)]
+struct SubscriberState {
+    cursor: u64,
+}
+
+#[derive(Clone, Debug)]
+struct ThreadState {
+    next_seq: u64,
+    events: VecDeque<StoredEvent>,
+    total_bytes: usize,
+    active_items: HashSet<String>,
+    unresolved_requests: HashSet<String>,
+    lifecycle_anchor: Option<DirectThreadRawEvent>,
+    last_completed_item_id: Option<String>,
+    subscribers: HashMap<String, SubscriberState>,
+}
+
+impl Default for ThreadState {
+    fn default() -> Self {
+        Self {
+            next_seq: 0,
+            events: VecDeque::new(),
+            total_bytes: 0,
+            active_items: HashSet::new(),
+            unresolved_requests: HashSet::new(),
+            lifecycle_anchor: None,
+            last_completed_item_id: None,
+            subscribers: HashMap::new(),
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct DirectThreadManager {
+    threads: HashMap<String, ThreadState>,
+    max_events: usize,
+    max_bytes: usize,
+}
+
+impl Default for DirectThreadManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl DirectThreadManager {
+    pub(crate) fn new() -> Self {
+        Self::with_limits(DEFAULT_MAX_EVENTS, DEFAULT_MAX_BYTES)
+    }
+
+    #[cfg(test)]
+    fn with_limits(max_events: usize, max_bytes: usize) -> Self {
+        Self {
+            threads: HashMap::new(),
+            max_events: max_events.max(1),
+            max_bytes: max_bytes.max(1),
+        }
+    }
+
+    pub(crate) fn append(
+        &mut self,
+        thread_id: &str,
+        draft: DirectThreadRawEventDraft,
+    ) -> DirectThreadRawEvent {
+        let thread = self.threads.entry(thread_id.to_string()).or_default();
+        thread.next_seq = thread.next_seq.saturating_add(1);
+        let event = DirectThreadRawEvent {
+            seq: thread.next_seq,
+            event_type: draft.event_type,
+            turn_id: draft.turn_id,
+            item_id: draft.item_id,
+            payload: draft.payload,
+        };
+        let cleanable = Self::observe_event(thread, &event);
+        let bytes = serde_json::to_vec(&event)
+            .map(|value| value.len())
+            .unwrap_or_default();
+        thread.total_bytes = thread.total_bytes.saturating_add(bytes);
+        thread.events.push_back(StoredEvent {
+            event: event.clone(),
+            bytes,
+            cleanable,
+        });
+        Self::mark_item_events_cleanable(thread, event.item_id.as_deref());
+        self.evict(thread_id);
+        event
+    }
+
+    pub(crate) fn subscribe(&mut self, thread_id: &str) -> DirectThreadSubscriptionBootstrap {
+        let thread = self.threads.entry(thread_id.to_string()).or_default();
+        let subscription_id = Uuid::new_v4().to_string();
+        let cursor = thread.next_seq;
+        thread
+            .subscribers
+            .insert(subscription_id.clone(), SubscriberState { cursor });
+
+        let mut events = thread
+            .events
+            .iter()
+            .filter(|stored| Self::is_bootstrap_event(thread, &stored.event, stored.cleanable))
+            .map(|stored| stored.event.clone())
+            .collect::<Vec<_>>();
+        if let Some(anchor) = thread.lifecycle_anchor.as_ref() {
+            if !events.iter().any(|event| event.seq == anchor.seq) {
+                events.push(anchor.clone());
+            }
+        }
+        events.sort_by_key(|event| event.seq);
+        DirectThreadSubscriptionBootstrap {
+            subscription_id,
+            last_completed_item_id: thread.last_completed_item_id.clone(),
+            events,
+        }
+    }
+
+    pub(crate) fn consume(
+        &mut self,
+        subscription_id: &str,
+    ) -> Result<DirectThreadConsumeResult, String> {
+        let Some((_, thread)) = self
+            .threads
+            .iter_mut()
+            .find(|(_, thread)| thread.subscribers.contains_key(subscription_id))
+        else {
+            return Err(SUBSCRIPTION_EXPIRED.to_string());
+        };
+        let cursor = thread
+            .subscribers
+            .get(subscription_id)
+            .map(|subscriber| subscriber.cursor)
+            .expect("subscriber checked above");
+        let oldest_seq = thread
+            .events
+            .front()
+            .map(|stored| stored.event.seq)
+            .unwrap_or(thread.next_seq.saturating_add(1));
+        if cursor.saturating_add(1) < oldest_seq {
+            thread.subscribers.remove(subscription_id);
+            return Err(SUBSCRIPTION_EXPIRED.to_string());
+        }
+        let events = thread
+            .events
+            .iter()
+            .filter(|stored| stored.event.seq > cursor)
+            .map(|stored| stored.event.clone())
+            .collect::<Vec<_>>();
+        if let Some(last) = events.last() {
+            thread
+                .subscribers
+                .get_mut(subscription_id)
+                .expect("subscriber remains registered")
+                .cursor = last.seq;
+        }
+        let result = DirectThreadConsumeResult { events };
+        Self::trim_prefix(thread);
+        Ok(result)
+    }
+
+    #[cfg(test)]
+    fn thread_debug(&self, thread_id: &str) -> Option<(usize, usize, usize)> {
+        self.threads.get(thread_id).map(|thread| {
+            (
+                thread.events.len(),
+                thread.total_bytes,
+                thread.subscribers.len(),
+            )
+        })
+    }
+
+    fn observe_event(thread: &mut ThreadState, event: &DirectThreadRawEvent) -> bool {
+        match event.event_type.as_str() {
+            "item.started" => {
+                if let Some(item_id) = event.item_id.as_deref() {
+                    thread.active_items.insert(item_id.to_string());
+                }
+                false
+            }
+            "item.completed" => {
+                if let Some(item_id) = event.item_id.as_deref() {
+                    thread.active_items.remove(item_id);
+                    thread.last_completed_item_id = Some(item_id.to_string());
+                }
+                true
+            }
+            "turn.started" | "turn.completed" => {
+                thread.lifecycle_anchor = Some(event.clone());
+                event.event_type == "turn.completed"
+            }
+            "approval.requested" | "request.requested" | "ask.requested" => {
+                if let Some(request_id) = request_id(event) {
+                    thread.unresolved_requests.insert(request_id);
+                }
+                false
+            }
+            "approval.resolved" | "request.resolved" | "ask.resolved" => {
+                if let Some(request_id) = request_id(event) {
+                    thread.unresolved_requests.remove(&request_id);
+                }
+                true
+            }
+            _ => true,
+        }
+    }
+
+    fn is_bootstrap_event(
+        thread: &ThreadState,
+        event: &DirectThreadRawEvent,
+        cleanable: bool,
+    ) -> bool {
+        if thread
+            .lifecycle_anchor
+            .as_ref()
+            .is_some_and(|anchor| anchor.seq == event.seq)
+        {
+            return true;
+        }
+        if let Some(item_id) = event.item_id.as_deref() {
+            return thread.active_items.contains(item_id);
+        }
+        if let Some(request_id) = request_id(event) {
+            return thread.unresolved_requests.contains(&request_id);
+        }
+        !cleanable
+    }
+
+    fn mark_item_events_cleanable(thread: &mut ThreadState, item_id: Option<&str>) {
+        let Some(item_id) = item_id else {
+            return;
+        };
+        if thread.active_items.contains(item_id) {
+            return;
+        }
+        for stored in &mut thread.events {
+            if stored.event.item_id.as_deref() == Some(item_id) {
+                stored.cleanable = true;
+            }
+        }
+    }
+
+    fn trim_prefix(thread: &mut ThreadState) {
+        let min_cursor = thread
+            .subscribers
+            .values()
+            .map(|subscriber| subscriber.cursor)
+            .min()
+            .unwrap_or(thread.next_seq);
+        loop {
+            let can_pop = thread
+                .events
+                .front()
+                .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);
+            }
+        }
+    }
+
+    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
+            });
+            if !over_limit {
+                if let Some(thread) = self.threads.get_mut(thread_id) {
+                    Self::trim_prefix(thread);
+                }
+                return;
+            }
+            let Some(thread) = self.threads.get_mut(thread_id) else {
+                return;
+            };
+            let slowest = thread
+                .subscribers
+                .iter()
+                .min_by_key(|(_, subscriber)| subscriber.cursor)
+                .map(|(id, _)| id.clone());
+            if let Some(subscription_id) = slowest {
+                thread.subscribers.remove(&subscription_id);
+                Self::trim_prefix(thread);
+                continue;
+            }
+            // 未完成 item 的事件可能暂时 pin 住队头；不能为了满足上限截断它。
+            return;
+        }
+    }
+}
+
+fn request_id(event: &DirectThreadRawEvent) -> Option<String> {
+    event
+        .payload
+        .get("requestId")
+        .and_then(Value::as_str)
+        .or_else(|| event.payload.get("id").and_then(Value::as_str))
+        .filter(|value| !value.is_empty())
+        .map(str::to_string)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn draft(event_type: &str, turn_id: &str, item_id: Option<&str>) -> DirectThreadRawEventDraft {
+        DirectThreadRawEventDraft {
+            event_type: event_type.to_string(),
+            turn_id: turn_id.to_string(),
+            item_id: item_id.map(str::to_string),
+            payload: serde_json::json!({}),
+        }
+    }
+
+    #[test]
+    fn subscribers_have_independent_cursors_on_one_global_queue() {
+        let mut manager = DirectThreadManager::with_limits(100, 100_000);
+        manager.append("thread-1", draft("turn.started", "turn-1", None));
+        let first = manager.subscribe("thread-1");
+        let second = manager.subscribe("thread-1");
+        manager.append("thread-1", draft("item.started", "turn-1", Some("item-1")));
+        manager.append("thread-1", draft("item.delta", "turn-1", Some("item-1")));
+
+        let first_batch = manager
+            .consume(&first.subscription_id)
+            .expect("first consume");
+        assert_eq!(first_batch.events.len(), 2);
+        let second_batch = manager
+            .consume(&second.subscription_id)
+            .expect("second consume");
+        assert_eq!(second_batch.events, first_batch.events);
+        assert!(manager
+            .consume(&first.subscription_id)
+            .expect("empty consume")
+            .events
+            .is_empty());
+    }
+
+    #[test]
+    fn bootstrap_contains_lifecycle_anchor_and_unfinished_events_only() {
+        let mut manager = DirectThreadManager::with_limits(100, 100_000);
+        manager.append("thread-1", draft("turn.started", "turn-1", None));
+        manager.append("thread-1", draft("item.started", "turn-1", Some("item-1")));
+        manager.append("thread-1", draft("item.delta", "turn-1", Some("item-1")));
+        manager.append(
+            "thread-1",
+            draft("item.completed", "turn-1", Some("item-1")),
+        );
+        manager.append("thread-1", draft("item.started", "turn-1", Some("item-2")));
+
+        let bootstrap = manager.subscribe("thread-1");
+        assert_eq!(bootstrap.last_completed_item_id.as_deref(), Some("item-1"));
+        assert_eq!(
+            bootstrap
+                .events
+                .iter()
+                .map(|event| event.event_type.as_str())
+                .collect::<Vec<_>>(),
+            vec!["turn.started", "item.started"]
+        );
+    }
+
+    #[test]
+    fn completion_releases_item_events_only_after_the_completion_event_is_appended() {
+        let mut manager = DirectThreadManager::with_limits(100, 100_000);
+        manager.append("thread-1", draft("item.started", "turn-1", Some("item-1")));
+        manager.append("thread-1", draft("item.delta", "turn-1", Some("item-1")));
+        let bootstrap = manager.subscribe("thread-1");
+        manager.append(
+            "thread-1",
+            draft("item.completed", "turn-1", Some("item-1")),
+        );
+        let events = manager
+            .consume(&bootstrap.subscription_id)
+            .expect("consume completion")
+            .events;
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].event_type, "item.completed");
+    }
+
+    #[test]
+    fn slow_subscriber_is_expired_when_queue_limit_is_reached() {
+        let mut manager = DirectThreadManager::with_limits(2, 100_000);
+        let subscription = manager.subscribe("thread-1");
+        manager.append("thread-1", draft("approval.resolved", "turn-1", None));
+        manager.append("thread-1", draft("approval.resolved", "turn-1", None));
+        manager.append("thread-1", draft("approval.resolved", "turn-1", None));
+        assert_eq!(
+            manager.consume(&subscription.subscription_id),
+            Err(SUBSCRIPTION_EXPIRED.to_string())
+        );
+    }
+
+    #[test]
+    fn unresolved_approval_is_kept_in_bootstrap_until_resolved() {
+        let mut manager = DirectThreadManager::with_limits(100, 100_000);
+        manager.append(
+            "thread-1",
+            DirectThreadRawEventDraft {
+                event_type: "approval.requested".to_string(),
+                turn_id: "turn-1".to_string(),
+                item_id: None,
+                payload: serde_json::json!({"requestId": "request-1"}),
+            },
+        );
+        let bootstrap = manager.subscribe("thread-1");
+        assert_eq!(bootstrap.events.len(), 1);
+        manager.append(
+            "thread-1",
+            DirectThreadRawEventDraft {
+                event_type: "approval.resolved".to_string(),
+                turn_id: "turn-1".to_string(),
+                item_id: None,
+                payload: serde_json::json!({"requestId": "request-1"}),
+            },
+        );
+        assert_eq!(
+            manager
+                .consume(&bootstrap.subscription_id)
+                .expect("consume resolution")
+                .events
+                .len(),
+            1
+        );
+    }
+
+    #[test]
+    fn turn_completed_anchor_survives_empty_queue_for_new_subscriber() {
+        let mut manager = DirectThreadManager::with_limits(100, 100_000);
+        manager.append("thread-1", draft("turn.completed", "turn-1", None));
+        let bootstrap = manager.subscribe("thread-1");
+        assert_eq!(bootstrap.events.len(), 1);
+        assert_eq!(bootstrap.events[0].event_type, "turn.completed");
+    }
+
+    #[test]
+    fn queue_cleanup_only_removes_a_cleanable_prefix() {
+        let mut manager = DirectThreadManager::with_limits(100, 100_000);
+        manager.append("thread-1", draft("item.started", "turn-1", Some("item-1")));
+        manager.append("thread-1", draft("approval.resolved", "turn-1", None));
+        let subscription = manager.subscribe("thread-1");
+        manager
+            .consume(&subscription.subscription_id)
+            .expect("consume");
+        let (events, _, _) = manager.thread_debug("thread-1").expect("thread");
+        assert_eq!(
+            events, 2,
+            "unfinished item at queue head blocks middle cleanup"
+        );
+    }
+}