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

拆分 DirectProject 历史收尾并停用浏览器写入

Browse Source 
DirectProject 历史累计逻辑拆到独立回合模块

浏览器仅显示项目对话，Rust 负责原始 item 持久化

拒绝空 rawResponseItem 并更新测试适配
This commit is contained in:
王德宇
2026-09-04 17:56:02 +08:00
parent b116a5d75c
commit 2a0a15639c
6 changed files with 160 additions and 389 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
@@ -15,6 +15,7 @@ mod codex_provider_proxy;
mod direct_codex_attachments;
mod direct_codex_audit;
mod direct_project_history;
mod direct_project_turn_history;
mod direct_runtime;
mod direct_tool_bridge;
mod direct_tools_mcp;
@@ -40,6 +41,7 @@ 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_project_turn_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
+23 -118
View File
@@ -2299,9 +2299,12 @@ impl CodexAppServerConnection {
}
}
Some(CodexTurnEvent::RawItem(item)) => {
if self.inner.workspace_mode
== CodexAppServerWorkspaceMode::DirectProject
{
if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject {
if item.is_null() {
return Err(platform_llm::LlmError::Deserialize(
"rawResponseItem/completed 缺少 item".to_string(),
));
}
append_direct_project_history_item_at(history_root, &item)
.map_err(platform_llm::LlmError::Transport)?;
direct_project_history.complete_item(&item);
@@ -3353,125 +3356,27 @@ pub(in crate::agent) fn shutdown_game_creator_codex_app_servers_impl() -> Result
})
}
#[cfg(test)]
pub(crate) fn build_direct_codex_history_prompt(
root: &Path,
_client_turn_id: &str,
current_prompt: &str,
_request: &LlmRunRequest,
_llm: &GameCreatorLlmConfig,
) -> Result<String, String> {
let mut lines = read_direct_project_chat_history_at(root)?
.messages
.into_iter()
.map(|message| format!("{}: {}", message.role, message.content))
.collect::<Vec<_>>();
lines.push(format!("user: {}", current_prompt.trim()));
Ok(lines.join("\n"))
}
#[cfg(test)]
mod tests {
use super::*;
/* legacy text replay tests removed with the prompt replay implementation
#[test]
fn direct_history_prompt_replays_all_project_messages_in_order() {
let root = tempfile::tempdir().expect("temp dir");
init_local_game_project_at(root.path(), "history-project", "history").expect("init");
for (role, content, message_id) in [
("user", "hello", "direct-codex:old:user"),
(
"assistant",
"partial\nunexpected interrupt happened here",
"partial-id",
),
("tool", "file-read result", "tool-id"),
("user", "stored raw request", "direct-codex:new-turn:user"),
] {
append_local_conversation_message_for_session_idempotent_at(
root.path(),
None,
None,
LocalConversationMessage {
role: role.to_string(),
content: content.to_string(),
agent_id: None,
},
message_id,
)
.expect("append history");
}
let request = LlmRunRequest::single_turn("system", "new request")
.with_model("fixture-model")
.with_max_output_tokens(16_000);
let prompt = build_direct_codex_history_prompt(
root.path(),
"new-turn",
"new request",
&request,
&test_llm(),
)
.expect("build prompt");
assert_eq!(
prompt,
"user: hello\nassistant: partial\nunexpected interrupt happened here\ntool: file-read result\nuser: new request"
);
}
#[test]
fn direct_history_prompt_slides_old_prefix_when_budget_is_exceeded() {
let root = tempfile::tempdir().expect("temp dir");
init_local_game_project_at(root.path(), "window-project", "window").expect("init");
for (role, content, message_id) in [
("user", "old ".repeat(1_000), "old-user"),
("assistant", "middle ".repeat(100), "middle-assistant"),
("tool", "newest ".repeat(100), "newest-tool"),
] {
append_local_conversation_message_for_session_idempotent_at(
root.path(),
None,
None,
LocalConversationMessage {
role: role.to_string(),
content,
agent_id: None,
},
message_id,
)
.expect("append history");
}
let mut llm = test_llm();
llm.auto_compact_token_limit = 800;
let request = LlmRunRequest::single_turn("system", "new request")
.with_model("fixture-model")
.with_max_output_tokens(16_000);
let before =
std::fs::read_to_string(root.path().join(".agent/conversations/project.jsonl"))
.expect("read history");
let prompt = build_direct_codex_history_prompt(
root.path(),
"new-turn",
"new request",
&request,
&llm,
)
.expect("build prompt");
let after = std::fs::read_to_string(root.path().join(".agent/conversations/project.jsonl"))
.expect("read history");
assert!(prompt.starts_with(DIRECT_CODEX_REPLAY_OMISSION_MARKER));
assert!(prompt.contains("tool: newest"));
assert!(!prompt.contains("user: old"));
assert!(prompt.ends_with("user: new request"));
assert_eq!(before, after);
}
#[test]
fn direct_history_prompt_rejects_current_request_that_exceeds_context() {
let root = tempfile::tempdir().expect("temp dir");
init_local_game_project_at(root.path(), "oversized-project", "oversized").expect("init");
let mut llm = test_llm();
llm.context_window_tokens = 5_000;
llm.auto_compact_token_limit = 1_000;
let request = LlmRunRequest::single_turn("system", "new request")
.with_model("fixture-model")
.with_max_output_tokens(100);
let error = build_direct_codex_history_prompt(
root.path(),
"new-turn",
&"request ".repeat(10_000),
&request,
&llm,
)
.expect_err("oversized request should fail");
assert!(error.contains("Direct replay 请求"));
}
*/
#[test]
fn direct_item_activities_are_closed_safe_categories() {
let allowed = [
apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs
+14 -60
View File
@@ -1,56 +1,15 @@
use super::*;
use crate::project::{append_jsonl_line_unlocked, enforce_project_permission_policy, project_append_lock_for};
use crate::config::prepare_game_creator_private_path_for_read;
use crate::project::{
append_jsonl_line_unlocked, enforce_project_permission_policy, project_append_lock_for,
};
use crate::{LocalConversationMessageRecord, LocalConversationResult};
use serde_json::Value;
use std::collections::BTreeMap;
use std::fs::File;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
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")
}
@@ -74,9 +33,10 @@ pub(crate) fn append_direct_project_history_item_at(
let lock = project_append_lock_for(&path)?;
let _append_guard = lock.lock("DirectProject 历史追加写")?;
if let Some(item_id) = item.get("id").and_then(Value::as_str) {
if let Some(existing) = read_direct_project_history_items_at(root)?.into_iter().find(|existing| {
existing.get("id").and_then(Value::as_str) == Some(item_id)
}) {
if let Some(existing) = read_direct_project_history_items_at(root)?
.into_iter()
.find(|existing| existing.get("id").and_then(Value::as_str) == Some(item_id))
{
if &existing == item {
return Ok(());
}
@@ -119,9 +79,7 @@ pub(crate) fn direct_project_user_message_item(prompt: &str) -> Value {
})
}
pub(crate) fn read_direct_project_history_items_at(
root: &Path,
) -> Result<Vec<Value>, String> {
pub(crate) fn read_direct_project_history_items_at(root: &Path) -> Result<Vec<Value>, String> {
let path = history_path(root);
if !prepare_game_creator_private_path_for_read(&path, false, "DirectProject 历史")? {
return Ok(Vec::new());
@@ -153,8 +111,7 @@ pub(crate) fn read_direct_project_history_items_at(
));
}
};
if parsed.get("type").and_then(Value::as_str) != Some(DIRECT_PROJECT_HISTORY_RECORD_TYPE)
{
if parsed.get("type").and_then(Value::as_str) != Some(DIRECT_PROJECT_HISTORY_RECORD_TYPE) {
return Err(format!(
"DirectProject 历史记录类型无效：{}",
path.display()
@@ -197,10 +154,7 @@ pub(crate) fn read_direct_project_chat_history_at(
role: role.to_string(),
content,
agent_id: None,
message_id: item
.get("id")
.and_then(Value::as_str)
.map(str::to_string),
message_id: item.get("id").and_then(Value::as_str).map(str::to_string),
updated_at: 0,
})
})
apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_turn_history.rs
+48
View File
@@ -0,0 +1,48 @@
use super::direct_project_history::append_direct_project_history_item_at;
use serde_json::Value;
use std::collections::BTreeMap;
use std::path::Path;
#[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);
}
}
fn take_partial_items(&mut self) -> impl Iterator<Item = 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}],
})
})
}
}
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(())
}
apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs
+6 -6
View File
@@ -4327,8 +4327,8 @@ mod tests {
persist_direct_codex_assistant_reply_at(root.path(), turn_id, reply)
.expect("the App's same-id write converges idempotently");
let conversation = read_local_conversation_for_session_at(root.path(), None, None)
.expect("read project conversation");
let conversation =
read_direct_project_chat_history_at(root.path()).expect("read project conversation");
let message_id = format!("direct-codex:{turn_id}:assistant");
let persisted = conversation
.messages
@@ -4341,7 +4341,7 @@ mod tests {
assert!(
persist_direct_codex_assistant_reply_at(root.path(), turn_id, "不同回复")
.expect_err("same message identity cannot be rebound")
.contains("messageId 冲突")
.contains("item id 冲突")
);
}
@@ -4358,8 +4358,8 @@ mod tests {
persist_direct_codex_user_prompt_at(root.path(), turn_id, normalized_prompt)
.expect("retry reuses the same user message identity");
let conversation = read_local_conversation_for_session_at(root.path(), None, None)
.expect("read project conversation");
let conversation =
read_direct_project_chat_history_at(root.path()).expect("read project conversation");
let message_id = format!("direct-codex:{turn_id}:user");
let persisted = conversation
.messages
@@ -4372,7 +4372,7 @@ mod tests {
assert!(
persist_direct_codex_user_prompt_at(root.path(), turn_id, "不同的重试请求")
.expect_err("same turn identity cannot be rebound")
.contains("messageId 冲突")
.contains("item id 冲突")
);
}
apps/ai-game-creator-shell/src/App.tsx
+67 -205
View File
@@ -495,11 +495,6 @@ export function App({
const [directCodexTransientReply, setDirectCodexTransientReply] =
useState('');
const directCodexTransientReplyRef = useRef('');
const directCodexInterruptedPartialRef = useRef<{
projectPath: string;
text: string;
messageId: string;
} | null>(null);
const [
directCodexTransientReplyUpdatedAt,
setDirectCodexTransientReplyUpdatedAt,
@@ -510,10 +505,6 @@ export function App({
lastSequence: number;
receivedDirectUpdate: boolean;
} | null>(null);
const recoveredDirectCodexTurnClaimsRef = useRef(new Set<string>());
const directCodexClaimReleaseOnConversationWriteFailureRef = useRef(
new Map<string, string>(),
);
const directCodexConversationTurnSequenceRef = useRef(0);
const [projectSupervisorSessionId, setProjectSupervisorSessionId] = useState<
string | null
@@ -1675,6 +1666,12 @@ export function App({
projectConversationWriteConfirmedRef.current = null;
projectConversationWriteCancelledRef.current = null;
}
// DirectProject history is written by Rust from raw app-server items.
// The browser only renders that projection and must not append chat rows.
if (projectSupervisorOnly && directCodexProductRuntime) {
savedConversationCountRef.current = messages.length;
return;
}
const start = savedConversationCountRef.current;
const pendingMessages = messages.slice(start);
if (pendingMessages.length === 0) {
@@ -1745,7 +1742,6 @@ export function App({
return;
}
conversationWriteInFlightRef.current = true;
let failedDirectTerminalMessageCount: number | null = null;
void (async () => {
let wroteMessage = false;
for (const [index, message] of pendingMessages.entries()) {
@@ -1760,56 +1756,22 @@ export function App({
savedConversationCountRef.current = start + index + 1;
continue;
}
try {
await invoke<LocalConversationResult>(
'append_local_conversation_message',
{
projectPath: nextProjectPath,
await invoke<LocalConversationResult>(
'append_local_conversation_message',
{
projectPath: nextProjectPath,
agentId: null,
...(message.messageId ? { messageId: message.messageId } : {}),
message: {
role: message.role,
content: message.text,
agentId: null,
...(message.messageId ? { messageId: message.messageId } : {}),
message: {
role: message.role,
content: message.text,
agentId: null,
...(typeof message.updatedAt === 'number'
? { updatedAt: message.updatedAt }
: {}),
},
...(typeof message.updatedAt === 'number'
? { updatedAt: message.updatedAt }
: {}),
},
);
} catch (error) {
const claimKey = message.messageId
? directCodexClaimReleaseOnConversationWriteFailureRef.current.get(
message.messageId,
)
: undefined;
if (claimKey) {
directCodexClaimReleaseOnConversationWriteFailureRef.current.delete(
message.messageId!,
);
recoveredDirectCodexTurnClaimsRef.current.delete(claimKey);
failedDirectTerminalMessageCount = start + index + 1;
}
throw error;
}
if (message.messageId) {
const claimKey =
directCodexClaimReleaseOnConversationWriteFailureRef.current.get(
message.messageId,
);
if (claimKey) {
directCodexClaimReleaseOnConversationWriteFailureRef.current.delete(
message.messageId,
);
if (localProjectPathRef.current === nextProjectPath) {
// Any history read started before this terminal append may hold
// A stale history snapshot may still be missing this terminal
// append. Invalidate it before releasing the in-memory claim.
projectSupervisorHistoryLoadVersionRef.current += 1;
}
recoveredDirectCodexTurnClaimsRef.current.delete(claimKey);
}
}
},
);
wroteMessage = true;
savedConversationCountRef.current = start + index + 1;
}
@@ -1824,12 +1786,10 @@ export function App({
}
})()
.catch((error) => {
savedConversationCountRef.current = failedDirectTerminalMessageCount
? Math.max(
savedConversationCountRef.current,
failedDirectTerminalMessageCount,
)
: Math.min(savedConversationCountRef.current, start);
savedConversationCountRef.current = Math.min(
savedConversationCountRef.current,
start,
);
setWorkspaceStatus(
`项目对话保存失败：${
error instanceof Error ? error.message : String(error)
@@ -1859,6 +1819,7 @@ export function App({
conversationWriteVersion,
pendingUiConfirmation,
projectSupervisorOnly,
directCodexProductRuntime,
]);
function appendLocalPermissionLog(
@@ -5350,20 +5311,45 @@ export function App({
const appendDirectUserMessageIfMissing = (
current: ChatMessage[],
): ChatMessage[] => {
return current.some(
(message) => message.messageId === directUserMessageId,
)
? current
: [
...current,
{
role: 'user' as const,
text: prompt,
runtimeOwned: true,
messageId: directUserMessageId,
updatedAt: Date.now(),
},
];
if (
current.some((message) => message.messageId === directUserMessageId)
) {
return current;
}
let optimisticIndex = -1;
for (let index = current.length - 1; index >= 0; index -= 1) {
const message = current[index];
if (
message?.role === 'user' &&
message.text === prompt &&
!message.messageId
) {
optimisticIndex = index;
break;
}
}
if (optimisticIndex >= 0) {
return current.map((message, index) =>
index === optimisticIndex
? {
...message,
runtimeOwned: true,
messageId: directUserMessageId,
updatedAt: Date.now(),
}
: message,
);
}
return [
...current,
{
role: 'user' as const,
text: prompt,
runtimeOwned: true,
messageId: directUserMessageId,
updatedAt: Date.now(),
},
];
};
const appendDirectAssistantMessage = (
current: ChatMessage[],
@@ -5387,41 +5373,6 @@ export function App({
index === existingIndex ? nextMessage : message,
);
};
const persistDirectAssistantMessage = (text: string) =>
directInvoke<LocalConversationResult>(
'append_local_conversation_message',
{
projectPath: directProjectPath,
agentId: null,
messageId: directAssistantMessageId,
message: {
role: 'assistant',
content: text,
agentId: null,
},
},
);
const persistDirectPartialMessage = (messageId: string, text: string) =>
directInvoke<LocalConversationResult>(
'append_local_conversation_message',
{
projectPath: directProjectPath,
agentId: null,
messageId,
message: {
role: 'assistant',
// An interrupted partial is intentionally a normal assistant
// record so replay sees exactly what Codex emitted before the
// disconnect; the marker is product data, not UI metadata.
content: `${text.trim()}\nunexpected interrupt happened here`,
agentId: null,
},
},
);
const recoveredDirectCodexTurnClaimKey = `${directProjectPath}\u0000${clientTurnId}`;
recoveredDirectCodexTurnClaimsRef.current.add(
recoveredDirectCodexTurnClaimKey,
);
activeDirectCodexTurnRef.current = {
projectPath: directProjectPath,
turnId: clientTurnId,
@@ -5436,36 +5387,6 @@ export function App({
setDirectCodexTransientReplyUpdatedAt(null);
setProjectSupervisorRuntimeError('');
try {
const interruptedPartial = directCodexInterruptedPartialRef.current;
if (
interruptedPartial?.projectPath === directProjectPath &&
interruptedPartial.text.trim()
) {
await persistDirectPartialMessage(
interruptedPartial.messageId,
interruptedPartial.text,
);
directCodexInterruptedPartialRef.current = null;
}
// Rust owns the normalized user record for attachment turns so the
// durable message includes the same bounded project mapping that is
// sent to Codex. Plain turns keep the optimistic browser write; the
// Rust writer then converges on it through messageId idempotency.
if (!attachments?.length) {
await directInvoke<LocalConversationResult>(
'append_local_conversation_message',
{
projectPath: directProjectPath,
agentId: null,
messageId: directUserMessageId,
message: {
role: 'user',
content: prompt,
agentId: null,
},
},
);
}
const directTurnInput: {
projectPath: string;
prompt: string;
@@ -5487,28 +5408,11 @@ export function App({
'chat_with_game_creator_direct_codex',
directTurnInput,
);
try {
await persistDirectAssistantMessage(reply);
} catch (error) {
if (localProjectPathRef.current === directProjectPath) {
setProjectSupervisorRuntimeError(
`陶泥儿回复保存失败：${
error instanceof Error ? error.message : String(error)
}`,
);
}
}
// Rust persists a successful Direct reply before returning Ok. The
// browser append is redundant, so the in-memory turn claim can be
// released without reopening the Provider side effect. Invalidate
// any history snapshot captured before Rust committed the terminal
// reply first.
// Rust already persisted the complete raw response items. Invalidate
// any history snapshot captured before the turn completed.
if (localProjectPathRef.current === directProjectPath) {
projectSupervisorHistoryLoadVersionRef.current += 1;
}
recoveredDirectCodexTurnClaimsRef.current.delete(
recoveredDirectCodexTurnClaimKey,
);
if (localProjectPathRef.current === directProjectPath) {
clearDirectCodexTransientReply(directProjectPath, clientTurnId);
setMessages((current) =>
@@ -5520,9 +5424,6 @@ export function App({
}
} catch (error) {
if (isDirectCodexTurnAlreadyRunningError(error)) {
recoveredDirectCodexTurnClaimsRef.current.delete(
recoveredDirectCodexTurnClaimKey,
);
if (localProjectPathRef.current === directProjectPath) {
clearDirectCodexTransientReply(directProjectPath, clientTurnId);
setProjectSupervisorRuntimeError(
@@ -5542,46 +5443,7 @@ export function App({
'陶泥儿智能创作',
true,
);
const partial = directCodexTransientReplyRef.current.trim();
if (partial) {
const partialMessageId =
globalThis.crypto?.randomUUID?.() ||
`direct-partial-${Date.now().toString(36)}`;
directCodexInterruptedPartialRef.current = {
projectPath: directProjectPath,
text: partial,
messageId: partialMessageId,
};
try {
await persistDirectPartialMessage(partialMessageId, partial);
} catch {
// The next user send retries this idempotent append before
// constructing the replay prompt.
}
}
try {
await persistDirectAssistantMessage(visibleMessage);
if (localProjectPathRef.current === directProjectPath) {
projectSupervisorHistoryLoadVersionRef.current += 1;
}
recoveredDirectCodexTurnClaimsRef.current.delete(
recoveredDirectCodexTurnClaimKey,
);
} catch {
if (localProjectPathRef.current === directProjectPath) {
// Do not release the claim while the React conversation writer
// can still persist this terminal record. That writer releases
// the claim only after its exact append resolves or rejects.
directCodexClaimReleaseOnConversationWriteFailureRef.current.set(
directAssistantMessageId,
recoveredDirectCodexTurnClaimKey,
);
} else {
recoveredDirectCodexTurnClaimsRef.current.delete(
recoveredDirectCodexTurnClaimKey,
);
}
}
projectSupervisorHistoryLoadVersionRef.current += 1;
if (localProjectPathRef.current === directProjectPath) {
clearDirectCodexTransientReply(directProjectPath, clientTurnId);
setProjectSupervisorRuntimeError(visibleMessage);