GenarrativeAI/Genarrative
11
0
Fork 0
Code Issues 22 Pull Requests 11 Actions 3 Packages Projects 1 Releases Wiki Activity

接入DirectProject线程事件消费命令

Browse Source 
将app-server item与turn事件接入Thread Manager

新增subscribe、consume和历史切片Tauri命令

保证响应item持久化成功后才发布完成事件
This commit is contained in:
王德宇
2026-09-15 17:18:10 +08:00
parent 8058ab7810
commit b0d85eb158
5 changed files with 198 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server/mod.rs
+72 -2
View File
@@ -2771,6 +2771,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 {
@@ -2822,6 +2837,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() {
@@ -2864,6 +2888,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)) => {
@@ -2926,6 +2964,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)) => {
@@ -2942,11 +2999,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())
apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs
+18
View File
@@ -512,6 +512,24 @@ pub(crate) fn read_direct_project_history_items_at(root: &Path) -> Result<Vec<Va
Ok(items)
}
pub(crate) fn read_direct_project_history_items_slice_at(
root: &Path,
before_item_id: Option<&str>,
limit: usize,
) -> Result<(Vec<Value>, 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<LocalConversationResult, String> {
apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs
+67 -1
View File
@@ -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<Mutex<DirectThreadManager>> = OnceLock::new();
static DIRECT_THREAD_MANAGER_APP_HANDLE: OnceLock<tauri::AppHandle> = 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<DirectThreadRawEvent>,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct DirectThreadHistorySlice {
pub(crate) items: Vec<Value>,
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<String> {
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<DirectThreadManager> {
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<DirectThreadConsumeResult, String> {
global_direct_thread_manager()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.consume(subscription_id)
}
fn request_id(event: &DirectThreadRawEvent) -> Option<String> {
event
.payload
apps/ai-game-creator-shell/src-tauri/src/commands.rs
+37
View File
@@ -5241,6 +5241,43 @@ pub(crate) async fn read_direct_project_conversation(
.map_err(|error| format!("读取 DirectProject 历史后台任务失败：{error}"))?
}
#[tauri::command]
pub(crate) fn subscribe_direct_project_thread(
project_path: String,
) -> Result<DirectThreadSubscriptionBootstrap, String> {
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<DirectThreadConsumeResult, String> {
consume_direct_thread(subscription_id.trim())
}
#[tauri::command]
pub(crate) async fn read_direct_project_history_slice(
project_path: String,
before_item_id: Option<String>,
limit: Option<usize>,
) -> Result<DirectThreadHistorySlice, String> {
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]
pub(crate) fn append_local_conversation_message(
project_path: String,
apps/ai-game-creator-shell/src-tauri/src/main.rs
+4
View File
@@ -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)
@@ -2767,6 +2768,9 @@ fn main() {
archive_game_creator_agent_session,
read_local_conversation,
read_direct_project_conversation,
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,