diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server/mod.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server/mod.rs index a86ef26b5..8e73bb857 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server/mod.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server/mod.rs @@ -2816,6 +2816,21 @@ impl CodexAppServerConnection { } }; turn_start_guard.armed = false; + let direct_thread_id = history_root.to_string_lossy().into_owned(); + if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject { + append_direct_thread_event( + &direct_thread_id, + DirectThreadRawEventDraft { + event_type: "turn.started".to_string(), + turn_id: turn_id.clone(), + item_id: None, + payload: serde_json::json!({ + "threadId": thread_id, + "turnId": turn_id, + }), + }, + ); + } let mut receiver = self.register_turn(&turn_id).await; let mut direct_project_history = DirectProjectHistoryAccumulator::default(); let mut guard = CodexTurnGuard { @@ -2867,6 +2882,15 @@ impl CodexAppServerConnection { Some(CodexTurnEvent::AgentMessageDelta { item_id, delta }) => { if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject { direct_project_history.observe_delta(&item_id, &delta); + append_direct_thread_event( + &direct_thread_id, + DirectThreadRawEventDraft { + event_type: "item.delta".to_string(), + turn_id: turn_id.clone(), + item_id: Some(item_id.clone()), + payload: serde_json::json!({ "delta": delta.clone() }), + }, + ); } streamed_text.push_str(&delta); if let Some(observer) = direct_observer.as_deref_mut() { @@ -2909,6 +2933,20 @@ impl CodexAppServerConnection { })? .map_err(platform_llm::LlmError::InvalidRequest)?; direct_project_history.complete_item(&item); + let item_id = item + .get("id") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_string); + append_direct_thread_event( + &direct_thread_id, + DirectThreadRawEventDraft { + event_type: "item.completed".to_string(), + turn_id: turn_id.clone(), + item_id, + payload: serde_json::json!({ "item": item }), + }, + ); } } Some(CodexTurnEvent::Activity(activity)) => { @@ -2971,6 +3009,25 @@ impl CodexAppServerConnection { self.inner.workspace_mode.passive_item_boundary_name(), ))); } + if !completed + && self.inner.workspace_mode + == CodexAppServerWorkspaceMode::DirectProject + { + let item_id = item + .get("id") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_string); + append_direct_thread_event( + &direct_thread_id, + DirectThreadRawEventDraft { + event_type: "item.started".to_string(), + turn_id: turn_id.clone(), + item_id, + payload: serde_json::json!({ "item": item }), + }, + ); + } } } Some(CodexTurnEvent::Terminal(params)) => { @@ -2987,11 +3044,24 @@ impl CodexAppServerConnection { }) }); } - match turn + let status = turn .get("status") .and_then(serde_json::Value::as_str) - .unwrap_or_default() + .unwrap_or_default(); + if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject + && matches!(status, "completed" | "interrupted" | "failed") { + append_direct_thread_event( + &direct_thread_id, + DirectThreadRawEventDraft { + event_type: "turn.completed".to_string(), + turn_id: turn_id.clone(), + item_id: None, + payload: params.clone(), + }, + ); + } + match status { "completed" => { return final_text .filter(|text| !text.trim().is_empty()) diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs index 857f3521b..2f3a6f62c 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs @@ -512,6 +512,24 @@ pub(crate) fn read_direct_project_history_items_at(root: &Path) -> Result, + limit: usize, +) -> Result<(Vec, bool), String> { + let items = read_direct_project_history_items_at(root)?; + let end = match before_item_id { + Some(item_id) => items + .iter() + .position(|item| item.get("id").and_then(Value::as_str) == Some(item_id)) + .ok_or_else(|| format!("DirectProject 历史中不存在 item：{item_id}"))?, + None => items.len(), + }; + let bounded_limit = limit.clamp(1, 200); + let start = end.saturating_sub(bounded_limit); + Ok((items[start..end].to_vec(), start > 0)) +} + pub(crate) fn read_direct_project_chat_history_at( root: &Path, ) -> Result { 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 647c3b9e2..8b9edebd3 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 @@ -7,12 +7,17 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{Mutex, OnceLock}; 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"; +pub(crate) const DIRECT_THREAD_NOTIFY_EVENT: &str = "game-creator-direct-thread-notify"; + +static DIRECT_THREAD_MANAGER: OnceLock> = OnceLock::new(); +static DIRECT_THREAD_MANAGER_APP_HANDLE: OnceLock = OnceLock::new(); #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -48,6 +53,13 @@ pub(crate) struct DirectThreadConsumeResult { pub(crate) events: Vec, } +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct DirectThreadHistorySlice { + pub(crate) items: Vec, + pub(crate) has_more: bool, +} + #[derive(Clone, Debug)] struct StoredEvent { event: DirectThreadRawEvent, @@ -105,7 +117,6 @@ impl DirectThreadManager { 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(), @@ -170,6 +181,13 @@ impl DirectThreadManager { } } + fn subscriber_ids(&self, thread_id: &str) -> Vec { + self.threads + .get(thread_id) + .map(|thread| thread.subscribers.keys().cloned().collect()) + .unwrap_or_default() + } + pub(crate) fn consume( &mut self, subscription_id: &str, @@ -345,6 +363,54 @@ impl DirectThreadManager { } } +fn global_direct_thread_manager() -> &'static Mutex { + DIRECT_THREAD_MANAGER.get_or_init(|| Mutex::new(DirectThreadManager::new())) +} + +pub(crate) fn set_direct_thread_manager_app_handle(app: tauri::AppHandle) { + let _ = DIRECT_THREAD_MANAGER_APP_HANDLE.set(app); +} + +pub(crate) fn append_direct_thread_event( + thread_id: &str, + draft: DirectThreadRawEventDraft, +) -> DirectThreadRawEvent { + let (event, subscriber_ids) = { + let mut manager = global_direct_thread_manager() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let event = manager.append(thread_id, draft); + let subscriber_ids = manager.subscriber_ids(thread_id); + (event, subscriber_ids) + }; + if let Some(app) = DIRECT_THREAD_MANAGER_APP_HANDLE.get() { + for subscription_id in subscriber_ids { + let _ = tauri::Emitter::emit( + app, + DIRECT_THREAD_NOTIFY_EVENT, + serde_json::json!({ "subscriptionId": subscription_id }), + ); + } + } + event +} + +pub(crate) fn subscribe_direct_thread(thread_id: &str) -> DirectThreadSubscriptionBootstrap { + global_direct_thread_manager() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .subscribe(thread_id) +} + +pub(crate) fn consume_direct_thread( + subscription_id: &str, +) -> Result { + global_direct_thread_manager() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .consume(subscription_id) +} + fn request_id(event: &DirectThreadRawEvent) -> Option { event .payload diff --git a/apps/ai-game-creator-shell/src-tauri/src/commands.rs b/apps/ai-game-creator-shell/src-tauri/src/commands.rs index 53c77d151..0ca57dc67 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/commands.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/commands.rs @@ -5274,6 +5274,42 @@ pub(crate) async fn read_agent_runtime_error_detail( }) .await .map_err(|error| format!("读取统一错误诊断后台任务失败：{error}"))? + +#[tauri::command] +pub(crate) fn subscribe_direct_project_thread( + project_path: String, +) -> Result { + let root = Path::new(project_path.trim()); + enforce_project_permission_policy(root, "conversation.read")?; + let thread_id = root.to_string_lossy().into_owned(); + Ok(subscribe_direct_thread(&thread_id)) +} + +#[tauri::command] +pub(crate) fn consume_direct_project_thread( + subscription_id: String, +) -> Result { + consume_direct_thread(subscription_id.trim()) +} + +#[tauri::command] +pub(crate) async fn read_direct_project_history_slice( + project_path: String, + before_item_id: Option, + limit: Option, +) -> Result { + tauri::async_runtime::spawn_blocking(move || { + let root = Path::new(project_path.trim()); + enforce_project_permission_policy(root, "conversation.read")?; + let (items, has_more) = read_direct_project_history_items_slice_at( + root, + before_item_id.as_deref(), + limit.unwrap_or(20), + )?; + Ok(DirectThreadHistorySlice { items, has_more }) + }) + .await + .map_err(|error| format!("读取 DirectProject 历史切片后台任务失败：{error}"))? } #[tauri::command] diff --git a/apps/ai-game-creator-shell/src-tauri/src/main.rs b/apps/ai-game-creator-shell/src-tauri/src/main.rs index b95ebf358..e68dee027 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/main.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/main.rs @@ -2607,6 +2607,7 @@ fn main() { app.manage(gui_owner_lock); setup_log.append("startup.runner.start.begin"); set_game_creator_agent_runtime_update_app_handle(app.handle().clone()); + set_direct_thread_manager_app_handle(app.handle().clone()); let manifest_event_sink = start_game_creator_manifest_invalidation_event_sink(app.handle().clone())?; attach_external_agent_runner_gui_owner(&manifest_event_sink, &gui_owner_epoch) @@ -2769,6 +2770,9 @@ fn main() { read_local_conversation, read_direct_project_conversation, read_agent_runtime_error_detail, + subscribe_direct_project_thread, + consume_direct_project_thread, + read_direct_project_history_slice, append_local_conversation_message, append_direct_project_conversation_message, build_local_project_index,