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

接入 DirectProject 原始历史回放

Browse Source 
新增 Codex Responses item JSONL 读写模块并启用 raw 事件

DirectProject 新线程使用 thread/inject_items 恢复历史

异常终态收尾累计 assistant 文本并移除旧文本 replay 写入
This commit is contained in:
王德宇
2026-09-04 17:17:35 +08:00
parent 9ce0e89e43
commit 3d44af0a7d
5 changed files with 221 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/ai-game-creator-shell/src-tauri/src/agent.rs
+2
View File
@@ -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::*;
apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs
+79 -26
View File
@@ -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__"),
&current_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),
}
};
apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs
+135
View File
@@ -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<String, String>,
}
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<Value> {
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<String, String> {
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<Vec<Value>, 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)
}
apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs
-60
View File
@@ -4171,48 +4171,6 @@ fn normalize_direct_client_turn_id(client_turn_id: Option<&str>) -> Result<Strin
Ok(client_turn_id.to_string())
}
fn persist_direct_codex_assistant_reply_at(
root: &Path,
client_turn_id: &str,
reply: &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: "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)
apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs
+5 -5
View File
@@ -4318,12 +4318,12 @@ fn project_append_locks() -> &'static Mutex<BTreeMap<PathBuf, Arc<Mutex<()>>>> {
PROJECT_APPEND_LOCKS.get_or_init(|| Mutex::new(BTreeMap::new()))
}
pub(super) struct ProjectAppendLock {
pub(crate) struct ProjectAppendLock {
process_lock: Arc<Mutex<()>>,
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<ProjectAppendGuard<'_>, String> {
pub(crate) fn lock(&self, error_label: &str) -> Result<ProjectAppendGuard<'_>, String> {
let process_guard = self
.process_lock
.lock()
@@ -4348,7 +4348,7 @@ impl ProjectAppendLock {
}
}
pub(super) fn project_append_lock_for(path: &Path) -> Result<ProjectAppendLock, String> {
pub(crate) fn project_append_lock_for(path: &Path) -> Result<ProjectAppendLock, String> {
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<Opt
))
}
pub(super) fn append_jsonl_line_unlocked(
pub(crate) fn append_jsonl_line_unlocked(
path: &Path,
line: &str,
error_label: &str,