From 3d44af0a7d0292c2cf40976668920c9d27cbc591 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BE=B7=E5=AE=87?= Date: Fri, 4 Sep 2026 17:17:35 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=85=A5=20DirectProject=20=E5=8E=9F?= =?UTF-8?q?=E5=A7=8B=E5=8E=86=E5=8F=B2=E5=9B=9E=E6=94=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 Codex Responses item JSONL 读写模块并启用 raw 事件 DirectProject 新线程使用 thread/inject_items 恢复历史 异常终态收尾累计 assistant 文本并移除旧文本 replay 写入 --- .../src-tauri/src/agent.rs | 2 + .../src-tauri/src/agent/codex_app_server.rs | 105 ++++++++++---- .../src/agent/direct_project_history.rs | 135 ++++++++++++++++++ .../src-tauri/src/agent/direct_runtime.rs | 60 -------- .../src-tauri/src/project/agent_db.rs | 10 +- 5 files changed, 221 insertions(+), 91 deletions(-) create mode 100644 apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs 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 bb0f70aee..526857c4c 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent.rs @@ -14,6 +14,7 @@ mod codex_cli; mod codex_provider_proxy; mod direct_codex_attachments; mod direct_codex_audit; +mod direct_project_history; mod direct_runtime; mod direct_tool_bridge; mod direct_tools_mcp; @@ -38,6 +39,7 @@ pub(crate) use codex_cli::{ pub(crate) use codex_provider_proxy::*; pub(crate) use direct_codex_attachments::*; pub(crate) use direct_codex_audit::*; +pub(crate) use direct_project_history::*; pub(crate) use direct_runtime::*; pub(crate) use direct_tool_bridge::*; pub(crate) use direct_tools_mcp::*; diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs index 84504d2d8..692ac87bc 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs @@ -374,12 +374,16 @@ impl From<&AgentRuntimeProviderRequestSnapshot> for CodexNodeThreadKey { #[derive(Clone, Debug)] enum CodexTurnEvent { - AgentMessageDelta(String), + AgentMessageDelta { + item_id: String, + delta: String, + }, Activity(&'static str), Item { completed: bool, params: serde_json::Value, }, + RawItem(serde_json::Value), Terminal(serde_json::Value), TransportClosed(String), } @@ -961,6 +965,9 @@ fn codex_app_server_thread_start_params( params["modelProvider"] = serde_json::Value::String(GAME_CREATOR_CODEX_APP_SERVER_PROVIDER_ID.to_string()); } + if workspace_mode == CodexAppServerWorkspaceMode::DirectProject { + params["experimentalRawEvents"] = serde_json::Value::Bool(true); + } params } @@ -2114,7 +2121,7 @@ impl CodexAppServerConnection { llm: &GameCreatorLlmConfig, request: LlmRunRequest, direct_history_root: Option<&std::path::Path>, - direct_client_turn_id: Option<&str>, + _direct_client_turn_id: Option<&str>, mut on_agent_message_delta: Option<&mut (dyn FnMut(&platform_llm::LlmStreamDelta) + Send)>, mut direct_observer: Option<&mut (dyn FnMut(DirectCodexTurnObservation) + Send)>, mut audit: Option<&mut DirectCodexTurnAudit>, @@ -2124,29 +2131,29 @@ impl CodexAppServerConnection { self.wait_for_initial_client_mcp_startup().await; let thread_id = thread_lease.thread_id.clone(); let mut request = request; - if thread_created && self.inner.workspace_mode.uses_direct_conversation() { - // DirectProject owns the append-only project history in AGC. The - // replay builder derives a bounded prompt without mutating that - // durable fact source, so a new ephemeral thread can recover the - // newest contiguous context within the model budget. - let current_prompt = direct_codex_current_user_prompt(&request).to_string(); - let history_root = direct_history_root.unwrap_or(&self.inner.workspace_path); - let history_prompt = build_direct_codex_history_prompt( + let history_root = direct_history_root.unwrap_or(&self.inner.workspace_path); + if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject { + let current_prompt = direct_codex_current_user_prompt(&request).trim(); + if current_prompt.is_empty() { + return Err(platform_llm::LlmError::InvalidRequest( + "DirectProject 用户消息不能为空".to_string(), + )); + } + if thread_created { + let items = read_direct_project_history_items_at(history_root) + .map_err(platform_llm::LlmError::InvalidRequest)?; + self.request( + "thread/inject_items", + serde_json::json!({"threadId": thread_id, "items": items}), + ) + .await + .map_err(platform_llm::LlmError::Transport)?; + } + append_direct_project_history_item_at( history_root, - direct_client_turn_id.unwrap_or("__none__"), - ¤t_prompt, - &request, - llm, + &direct_project_user_message_item(current_prompt), ) .map_err(platform_llm::LlmError::InvalidRequest)?; - if let Some(message) = request - .messages - .iter_mut() - .rev() - .find(|message| message.role == LlmMessageRole::User) - { - message.content = history_prompt; - } } let prompt = if self.inner.workspace_mode.uses_direct_conversation() { direct_codex_user_prompt(&request) @@ -2228,6 +2235,7 @@ impl CodexAppServerConnection { }; turn_start_guard.armed = false; let mut receiver = self.register_turn(&turn_id).await; + let mut direct_project_history = DirectProjectHistoryAccumulator::default(); let mut guard = CodexTurnGuard { connection: self.clone(), thread_id: thread_id.clone(), @@ -2274,7 +2282,8 @@ impl CodexAppServerConnection { } }; match event { - Some(CodexTurnEvent::AgentMessageDelta(delta)) => { + Some(CodexTurnEvent::AgentMessageDelta { item_id, delta }) => { + direct_project_history.observe_delta(&item_id, &delta); streamed_text.push_str(&delta); if let Some(observer) = direct_observer.as_deref_mut() { observer(DirectCodexTurnObservation::AccumulatedText( @@ -2289,6 +2298,15 @@ impl CodexAppServerConnection { }); } } + Some(CodexTurnEvent::RawItem(item)) => { + if self.inner.workspace_mode + == CodexAppServerWorkspaceMode::DirectProject + { + append_direct_project_history_item_at(history_root, &item) + .map_err(platform_llm::LlmError::Transport)?; + direct_project_history.complete_item(&item); + } + } Some(CodexTurnEvent::Activity(activity)) => { if let Some(observer) = direct_observer.as_deref_mut() { observer(DirectCodexTurnObservation::Activity(activity)); @@ -2396,7 +2414,22 @@ impl CodexAppServerConnection { } } }; - let text = collect.await?; + let text = match collect.await { + Ok(text) => text, + Err(error) => { + if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject { + if let Err(persist_error) = persist_direct_project_partial_items_at( + history_root, + &mut direct_project_history, + ) { + return Err(platform_llm::LlmError::Transport(format!( + "DirectProject 异常收尾历史失败：{persist_error}；原始回合错误：{error}" + ))); + } + } + return Err(error); + } + }; guard.armed = false; self.inner.turns.lock().await.remove(&turn_id); let response = parse_game_creator_codex_app_server_text(&text, &thread_id, &request)?; @@ -2764,7 +2797,11 @@ async fn read_game_creator_codex_app_server_stdout( let safe_activity = direct_codex_safe_activity_for_notification(method); if !matches!( method, - "item/agentMessage/delta" | "item/started" | "item/completed" | "turn/completed" + "item/agentMessage/delta" + | "item/started" + | "item/completed" + | "rawResponseItem/completed" + | "turn/completed" ) && safe_activity.is_none() { continue; @@ -2805,12 +2842,28 @@ async fn read_game_creator_codex_app_server_stdout( else { continue; }; - CodexTurnEvent::AgentMessageDelta(delta.to_string()) + let Some(item_id) = params + .get("itemId") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + else { + continue; + }; + CodexTurnEvent::AgentMessageDelta { + item_id: item_id.to_string(), + delta: delta.to_string(), + } } "item/started" | "item/completed" => CodexTurnEvent::Item { completed: method == "item/completed", params, }, + "rawResponseItem/completed" => CodexTurnEvent::RawItem( + params + .get("item") + .cloned() + .unwrap_or(serde_json::Value::Null), + ), _ => CodexTurnEvent::Terminal(params), } }; 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 new file mode 100644 index 000000000..f4c19615a --- /dev/null +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs @@ -0,0 +1,135 @@ +use super::*; +use crate::project::{append_jsonl_line_unlocked, enforce_project_permission_policy, project_append_lock_for}; +use serde_json::Value; +use std::collections::BTreeMap; +use std::io::{BufRead, BufReader}; + +const DIRECT_PROJECT_HISTORY_RECORD_TYPE: &str = "response_item"; + +#[derive(Default)] +pub(crate) struct DirectProjectHistoryAccumulator { + text_by_item_id: BTreeMap, +} + +impl DirectProjectHistoryAccumulator { + pub(crate) fn observe_delta(&mut self, item_id: &str, delta: &str) { + self.text_by_item_id + .entry(item_id.to_string()) + .or_default() + .push_str(delta); + } + + pub(crate) fn complete_item(&mut self, item: &Value) { + if let Some(item_id) = item.get("id").and_then(Value::as_str) { + self.text_by_item_id.remove(item_id); + } + } + + pub(crate) fn take_partial_items(&mut self) -> Vec { + std::mem::take(&mut self.text_by_item_id) + .into_iter() + .filter(|(_, text)| !text.is_empty()) + .map(|(item_id, text)| { + serde_json::json!({ + "type": "message", + "role": "assistant", + "id": item_id, + "content": [{"type": "output_text", "text": text}], + }) + }) + .collect() + } +} + +pub(crate) fn persist_direct_project_partial_items_at( + root: &Path, + accumulator: &mut DirectProjectHistoryAccumulator, +) -> Result<(), String> { + for item in accumulator.take_partial_items() { + append_direct_project_history_item_at(root, &item)?; + } + Ok(()) +} + +fn history_path(root: &Path) -> PathBuf { + root.join(".agent/conversations/project.jsonl") +} + +fn record(item: &Value) -> Result { + serde_json::to_string(&serde_json::json!({ + "type": DIRECT_PROJECT_HISTORY_RECORD_TYPE, + "payload": item, + })) + .map_err(|error| format!("序列化 DirectProject 历史失败：{error}")) +} + +pub(crate) fn append_direct_project_history_item_at( + root: &Path, + item: &Value, +) -> Result<(), String> { + enforce_project_permission_policy(root, "conversation.write")?; + let _project_lock = crate::project::acquire_project_write_lock(root, "conversation.write")?; + let path = history_path(root); + let lock = project_append_lock_for(&path)?; + let _append_guard = lock.lock("DirectProject 历史追加写")?; + let line = record(item)?; + append_jsonl_line_unlocked(&path, &line, "DirectProject 历史") +} + +pub(crate) fn direct_project_user_message_item(prompt: &str) -> Value { + serde_json::json!({ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": prompt}], + }) +} + +pub(crate) fn read_direct_project_history_items_at( + root: &Path, +) -> Result, String> { + let path = history_path(root); + if !path.exists() { + return Ok(Vec::new()); + } + let file = File::open(&path) + .map_err(|error| format!("打开 DirectProject 历史失败：{}: {error}", path.display()))?; + let mut items = Vec::new(); + let mut reader = BufReader::new(file); + loop { + let mut line = String::new(); + let bytes = reader + .read_line(&mut line) + .map_err(|error| format!("读取 DirectProject 历史失败：{}: {error}", path.display()))?; + if bytes == 0 { + break; + } + let had_newline = line.ends_with('\n'); + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let parsed: Value = match serde_json::from_str(trimmed) { + Ok(value) => value, + Err(_error) if !had_newline => break, + Err(error) => { + return Err(format!( + "解析 DirectProject 历史失败：{}: {error}", + path.display() + )); + } + }; + if parsed.get("type").and_then(Value::as_str) != Some(DIRECT_PROJECT_HISTORY_RECORD_TYPE) + { + return Err(format!( + "DirectProject 历史记录类型无效：{}", + path.display() + )); + } + let item = parsed + .get("payload") + .cloned() + .ok_or_else(|| format!("DirectProject 历史记录缺少 payload：{}", path.display()))?; + items.push(item); + } + Ok(items) +} diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs index fc36389d8..2098314b9 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs @@ -4171,48 +4171,6 @@ fn normalize_direct_client_turn_id(client_turn_id: Option<&str>) -> Result Result<(), String> { - enforce_project_permission_policy(root, "conversation.write")?; - let _lock = acquire_project_write_lock(root, "conversation.write")?; - append_local_conversation_message_for_session_idempotent_at( - root, - None, - None, - LocalConversationMessage { - role: "assistant".to_string(), - content: reply.to_string(), - agent_id: None, - }, - &format!("direct-codex:{client_turn_id}:assistant"), - ) - .map(|_| ()) -} - -fn persist_direct_codex_user_prompt_at( - root: &Path, - client_turn_id: &str, - prompt: &str, -) -> Result<(), String> { - enforce_project_permission_policy(root, "conversation.write")?; - let _lock = acquire_project_write_lock(root, "conversation.write")?; - append_local_conversation_message_for_session_idempotent_at( - root, - None, - None, - LocalConversationMessage { - role: "user".to_string(), - content: prompt.trim().to_string(), - agent_id: None, - }, - &format!("direct-codex:{client_turn_id}:user"), - ) - .map(|_| ()) -} - #[tauri::command] pub(crate) async fn chat_with_game_creator_direct_codex( project_path: String, @@ -4244,15 +4202,6 @@ pub(crate) async fn chat_with_game_creator_direct_codex( return Err(error); } }; - if let Err(error) = persist_direct_codex_user_prompt_at(root, &turn_id, &user_prompt) { - audit.finish(false); - turn_emitter.emit("failed", Some("none"), None); - return Err(redact_agent_runtime_error( - root, - &format!("Direct 用户消息持久化失败，已拒绝发起回合：{error}"), - 500, - )); - } let reply = match run_direct_game_creator_turn_at_with_creation_type_and_emitter( root, &user_prompt, @@ -4268,15 +4217,6 @@ pub(crate) async fn chat_with_game_creator_direct_codex( return Err(error); } }; - if let Err(error) = persist_direct_codex_assistant_reply_at(root, &turn_id, &reply) { - audit.finish(false); - turn_emitter.emit("failed", Some("none"), None); - return Err(redact_agent_runtime_error( - root, - &format!("Direct 成功回复持久化失败，已拒绝以未落盘状态返回：{error}"), - 500, - )); - } audit.finish(true); turn_emitter.emit("completed", Some("none"), Some(reply.clone())); Ok(reply) diff --git a/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs b/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs index bd2f3e18b..4fd654a80 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs @@ -4318,12 +4318,12 @@ fn project_append_locks() -> &'static Mutex>>> { PROJECT_APPEND_LOCKS.get_or_init(|| Mutex::new(BTreeMap::new())) } -pub(super) struct ProjectAppendLock { +pub(crate) struct ProjectAppendLock { process_lock: Arc>, os_lock_path: PathBuf, } -pub(super) struct ProjectAppendGuard<'a> { +pub(crate) struct ProjectAppendGuard<'a> { _process_guard: std::sync::MutexGuard<'a, ()>, _os_lock: File, } @@ -4335,7 +4335,7 @@ impl ProjectAppendLock { .map_err(|_| format!("获取{error_label}进程内锁失败：锁已损坏")) } - pub(super) fn lock(&self, error_label: &str) -> Result, String> { + pub(crate) fn lock(&self, error_label: &str) -> Result, String> { let process_guard = self .process_lock .lock() @@ -4348,7 +4348,7 @@ impl ProjectAppendLock { } } -pub(super) fn project_append_lock_for(path: &Path) -> Result { +pub(crate) fn project_append_lock_for(path: &Path) -> Result { let mut locks = project_append_locks() .lock() .map_err(|_| "获取本地追加写锁失败：锁已损坏".to_string())?; @@ -4539,7 +4539,7 @@ fn try_open_project_append_os_lock(path: &Path, error_label: &str) -> Result