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 25d1591e4..fcfa250be 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 @@ -760,6 +760,17 @@ fn direct_thread_item_id(item: &serde_json::Value) -> Option { .map(str::to_string) } +/// 原始 response item 的调用 id:工具条目的 app-server `itemId` 就是这个值, +/// 所以它是两个 id 空间唯一的对齐点。 +fn direct_thread_item_call_id(item: &serde_json::Value) -> Option { + item.get("call_id") + .or_else(|| item.get("callId")) + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) +} + fn direct_codex_command_is_game_verification(command: &str) -> bool { let command = command.to_ascii_lowercase(); command.contains("game.static_smoke") @@ -2932,6 +2943,7 @@ impl CodexAppServerConnection { event_type: "turn.started".to_string(), turn_id: turn_id.clone(), item_id: None, + call_id: None, payload: serde_json::json!({ "threadId": thread_id, "turnId": turn_id, @@ -2996,6 +3008,7 @@ impl CodexAppServerConnection { event_type: "item.delta".to_string(), turn_id: turn_id.clone(), item_id: Some(item_id.clone()), + call_id: None, payload: serde_json::json!({ "delta": delta.clone() }), }, ); @@ -3060,6 +3073,7 @@ impl CodexAppServerConnection { event_type: "item.completed".to_string(), turn_id: turn_id.clone(), item_id, + call_id: direct_thread_item_call_id(&item), payload: serde_json::json!({}), }, ); @@ -3079,6 +3093,7 @@ impl CodexAppServerConnection { event_type: event_type.to_string(), turn_id: turn_id.clone(), item_id: None, + call_id: None, payload: request_id .map(|id| serde_json::json!({ "requestId": id })) .unwrap_or_else(|| serde_json::json!({})), @@ -3200,6 +3215,7 @@ impl CodexAppServerConnection { event_type: "item.started".to_string(), turn_id: turn_id.clone(), item_id, + call_id: None, payload: direct_thread_item_started_payload(item), }, ); @@ -3248,6 +3264,7 @@ impl CodexAppServerConnection { event_type: "turn.completed".to_string(), turn_id: turn_id.clone(), item_id: None, + call_id: None, payload: serde_json::json!({ "status": status }), }, ); 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 e3bc497ea..f72f37380 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 @@ -28,6 +28,10 @@ pub(crate) struct DirectThreadRawEvent { pub(crate) turn_id: String, #[serde(skip_serializing_if = "Option::is_none")] pub(crate) item_id: Option, + /// 合并身份的另一半:工具类条目的 `itemId` 是调用 id,原始 item 的 `id` 是 response item id, + /// 两者只在 `callId` 上对齐。前端按 `callId ?? itemId` 归并同一张卡片。 + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) call_id: Option, pub(crate) payload: Value, } @@ -36,9 +40,17 @@ pub(crate) struct DirectThreadRawEventDraft { pub(crate) event_type: String, pub(crate) turn_id: String, pub(crate) item_id: Option, + pub(crate) call_id: Option, pub(crate) payload: Value, } +impl DirectThreadRawEvent { + /// 条目合并身份:有 `callId` 就用它,否则用 `itemId`。 + fn item_identity(&self) -> Option<&str> { + self.call_id.as_deref().or(self.item_id.as_deref()) + } +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct DirectThreadSubscriptionBootstrap { @@ -81,6 +93,13 @@ struct ThreadState { total_bytes: usize, active_items: HashSet, unresolved_requests: HashSet, + /// 最新一条 `turn.started` / `turn.completed` 的独立拷贝。 + /// + /// TODO(thread-manager): 这里有意只保留"锚点",因为 replay 队列会回收可回收事件, + /// 队列本身不是完美事件日志——被回收的 `turn.started` / `turn.completed` 不会回放, + /// 只有这份拷贝保证新订阅仍能判定"最新回合是否还在跑"。若将来需要回放多个回合的 + /// 生命周期(回合账本、跨进程恢复、按回合统计),必须另建持久 ledger, + /// 不能靠扩大这份拷贝或放宽回收规则来模拟。 lifecycle_anchor: Option, last_completed_item_id: Option, subscribers: HashMap, @@ -140,6 +159,7 @@ impl DirectThreadManager { event_type: draft.event_type, turn_id: draft.turn_id, item_id: draft.item_id, + call_id: draft.call_id, payload: draft.payload, }; let cleanable = Self::observe_event(thread, &event); @@ -152,7 +172,7 @@ impl DirectThreadManager { bytes, cleanable, }); - Self::mark_item_events_cleanable(thread, event.item_id.as_deref()); + Self::mark_item_events_cleanable(thread, event.item_identity()); if matches!( event.event_type.as_str(), "approval.resolved" | "request.resolved" | "ask.resolved" @@ -256,19 +276,24 @@ impl DirectThreadManager { fn observe_event(thread: &mut ThreadState, event: &DirectThreadRawEvent) -> bool { match event.event_type.as_str() { "item.started" => { - if let Some(item_id) = event.item_id.as_deref() { - thread.active_items.insert(item_id.to_string()); + if let Some(identity) = event.item_identity() { + thread.active_items.insert(identity.to_string()); } false } "item.completed" => { + if let Some(identity) = event.item_identity() { + thread.active_items.remove(identity); + } + // 历史锚点必须是 project.jsonl 里的 response item id,不能用 call id。 if let Some(item_id) = event.item_id.as_deref() { - thread.active_items.remove(item_id); thread.last_completed_item_id = Some(item_id.to_string()); } true } "turn.started" | "turn.completed" => { + // 队列只保留最新一条生命周期事件,更早的可能已被回收;见 + // `ThreadState::lifecycle_anchor` 的 TODO:这不是完整事件日志。 thread.lifecycle_anchor = Some(event.clone()); true } @@ -301,8 +326,8 @@ impl DirectThreadManager { { return true; } - if let Some(item_id) = event.item_id.as_deref() { - return thread.active_items.contains(item_id); + if let Some(identity) = event.item_identity() { + return thread.active_items.contains(identity); } if let Some(request_id) = request_id(event) { return thread.unresolved_requests.contains(&request_id); @@ -476,10 +501,57 @@ mod tests { event_type: event_type.to_string(), turn_id: turn_id.to_string(), item_id: item_id.map(str::to_string), + call_id: None, payload: serde_json::json!({}), } } + /// 工具条目的两个 id 空间必须靠 `callId` 对齐:app-server `item.started` 的 itemId 是调用 id, + /// 原始 item 的 `item.completed` 的 itemId 是 response item id。 + fn draft_with_call_id( + event_type: &str, + turn_id: &str, + item_id: Option<&str>, + call_id: Option<&str>, + ) -> DirectThreadRawEventDraft { + DirectThreadRawEventDraft { + event_type: event_type.to_string(), + turn_id: turn_id.to_string(), + item_id: item_id.map(str::to_string), + call_id: call_id.map(str::to_string), + payload: serde_json::json!({}), + } + } + + #[test] + fn call_id_joins_started_and_completed_across_id_spaces() { + let mut manager = DirectThreadManager::with_limits(100, 100_000); + manager.append( + "thread-1", + draft_with_call_id("item.started", "turn-1", Some("call-1"), None), + ); + manager.append( + "thread-1", + draft_with_call_id("item.completed", "turn-1", Some("item-9"), Some("call-1")), + ); + // 活跃条目按合并身份清理,不留下永远收不到完成事件的幽灵条目。 + let bootstrap = manager.subscribe("thread-1"); + assert!( + bootstrap + .events + .iter() + .all(|event| event.event_type != "item.started"), + "已完成的条目不得再作为运行态事件回到 bootstrap" + ); + // 没有订阅者时,可回收事件全部被回收;历史锚点靠独立字段保留,不占队列。 + assert_eq!( + manager.thread_debug("thread-1").map(|debug| debug.0), + Some(0) + ); + // 历史锚点仍然是 response item id,前端才能拿它当分页锚点。 + assert_eq!(bootstrap.last_completed_item_id.as_deref(), Some("item-9")); + } + #[test] fn subscribers_have_independent_cursors_on_one_global_queue() { let mut manager = DirectThreadManager::with_limits(100, 100_000); @@ -582,6 +654,7 @@ mod tests { event_type: "approval.requested".to_string(), turn_id: "turn-1".to_string(), item_id: None, + call_id: None, payload: serde_json::json!({"requestId": "request-1"}), }, ); @@ -593,6 +666,7 @@ mod tests { event_type: "approval.resolved".to_string(), turn_id: "turn-1".to_string(), item_id: None, + call_id: None, payload: serde_json::json!({"requestId": "request-1"}), }, ); @@ -615,6 +689,7 @@ mod tests { event_type: "approval.requested".to_string(), turn_id: "turn-1".to_string(), item_id: None, + call_id: None, payload: serde_json::json!({"requestId": "request-1"}), }, ); @@ -625,6 +700,7 @@ mod tests { event_type: "approval.resolved".to_string(), turn_id: "turn-1".to_string(), item_id: None, + call_id: None, payload: serde_json::json!({"requestId": "request-1"}), }, );