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

宿主：开口用户条目的发点提前到接单之后，失败说明才能挂回自己那一轮

Browse Source 
- 本轮开口用户条目（item_completed，direct-codex:{clientTurnId}:user）原来在 app-server turn/start 应答之后才下发；接单到 turn/start 之间的失败（连不上 app-server、执行器未通过验收、历史注入失败）走不到那一步，事件流里只有逻辑回合的一对事件，没有开口条目
- 把那段内联下发抽成 emit_direct_thread_user_item，发点提前到「接单成立、用户条目落盘成功、起 codex 之前」（direct_runtime/user_input.rs 的命令主体），并删掉 turn/start 之后那一处：线上仍只有一处下发，不变式变成「接单 → 开口用户条目 → 整轮里其余一切」
- 新增回归用例 the_opening_user_item_is_emitted_before_anything_that_can_fail_in_the_turn：断言行首两条事件是带身份的 turn.started 与开口用户条目，终态只能在它们之后
- 回显过滤用例补上同一发点的模拟步骤（生产入口的两个动作：落盘 + 下发）
This commit is contained in:
王德宇
2026-09-24 18:02:26 +08:00
parent cd196cf61c
commit 2487a2c8e6
2 changed files with 109 additions and 15 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
+34 -15
View File
@@ -831,6 +831,30 @@ fn direct_thread_visible_item(
direct_thread_event_item(root, item)
}
/// 下发本轮的开口用户条目：接单之后、起 codex 之前的第一条运行态条目。
///
/// **顺序是这条通道的全部意义**：整轮里任何失败说明都靠"属于哪一轮"归位，而归属只认这一轮的
/// 开口用户条目。发点在 `turn/start` 之后时，"接单到 `turn/start` 之间"的失败（连不上
/// app-server、执行器未通过验收、历史注入失败）没有用户条目可以挂，说明会按位置落进**上一轮**
/// 的分区里：界面显示成"错误在用户消息上面"，上一轮还顶替本轮显示耗时，本轮的用户气泡再自成
/// 一个 0.0 秒的假回合。
///
/// 调用点必须是"用户条目落盘成功之后"（`agent/direct_runtime/user_input.rs` 的命令主体）：
/// 条目身份取自落盘的那条条目，不在这里重造。投影不出条目时返回 `None`，不下发半条。
pub(crate) fn emit_direct_thread_user_item(
root: &std::path::Path,
item: &serde_json::Value,
) -> Option<DirectThreadItem> {
let entry_item = direct_thread_event_item(root, item)?;
// 条目时间是落盘 / 观测时间；前端按同一个身份保留更早的真实发送时间，不用此时间覆盖它。
let at = entry_item.at();
append_direct_thread_event(
&direct_thread_id_for_project(root),
DirectThreadEvent::item_completed(entry_item.clone(), at),
);
Some(entry_item)
}
/// AGC 预写的 canonical 用户条目 id：`direct-codex:{clientTurnId}:user`。
///
/// 与 `direct_project_history::is_direct_project_codex_user_item` 的判据同一份口径（前缀 +
@@ -3545,27 +3569,14 @@ impl CodexAppServerConnection {
.as_ref()
.and_then(direct_thread_item_identity);
// 逻辑回合的**边界**不在这里：开始事件由接单动作发出、兜底由接单占用对象持有
// （`direct_turn_accept.rs`）。这里只把本轮的用户条目作为第一条运行态条目下发，
// 于是顺序天然是"逻辑回合开始 → 用户消息 → 起 codex"。
// （`direct_turn_accept.rs`）。本轮的用户条目也不在这里下发——发点在接单之后、
// 起 codex 之前（`emit_direct_thread_user_item`），见那条注释。
//
// 下面这个毫秒钟与逻辑回合无关，只服务模型终态的**完成时刻**：上游 Turn 的
// `startedAt` / `completedAt` 只有秒级，秒级截断撑不起前端 0.1 秒粒度的展示，也可能
// 让完成时刻落进该轮用户消息的同一秒。因此这里在进入模型往返前取一次宿主毫秒钟，与
// `durationMs` 相加得到终态时刻；拿不到 `durationMs` 时退回观察时刻。
let direct_turn_started_at_ms = direct_tool_call_now_ms();
if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject {
if let Some(user_item) = direct_turn_user_item.as_ref() {
if let Some(entry_item) = direct_thread_event_item(history_root, user_item) {
// 这里的条目时间可能是启动应答后的观测时间；前端按同一用户条目身份
// 保留更早的真实发送时间，不用此事件时间覆盖它。
let user_item_at = entry_item.at();
append_direct_thread_event(
&direct_thread_id,
DirectThreadEvent::item_completed(entry_item, user_item_at),
);
}
}
}
let mut receiver = self.register_turn(&turn_id).await;
let mut direct_project_history = DirectProjectHistoryAccumulator::default();
let mut guard = CodexTurnGuard {
@@ -8044,6 +8055,10 @@ done
.expect("accept logical turn");
crate::agent::append_direct_project_user_message_at(&project, &user_item)
.expect("persist opener user item");
// 生产入口在落盘成功、起 codex 之前就把本轮的用户条目下发（`emit_direct_thread_user_item`）：
// 这里补上同一步，于是"开口用户条目一定在整轮里最先到"这条不变式在用例里也成立。
crate::agent::codex_app_server::emit_direct_thread_user_item(&project, &user_item)
.expect("emit opener user item");
let execution = super::super::direct_execution::open_at(
&temp.path().join("host"),
&project,
@@ -8207,6 +8222,10 @@ done
.expect("accept logical turn");
crate::agent::append_direct_project_user_message_at(&project, &user_item)
.expect("persist opener user item");
// 生产入口在落盘成功、起 codex 之前就把本轮的用户条目下发（`emit_direct_thread_user_item`）：
// 这里补上同一步，于是"开口用户条目一定在整轮里最先到"这条不变式在用例里也成立。
crate::agent::codex_app_server::emit_direct_thread_user_item(&project, &user_item)
.expect("emit opener user item");
let execution = super::super::direct_execution::open_at(
&temp.path().join("host"),
&project,
apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime/user_input.rs
+75
View File
@@ -129,6 +129,10 @@ async fn chat_with_game_creator_direct_codex_typed(
reservation.finish_if_unfinished(DirectTurnTerminal::failed(root, &failure));
return Ok(());
}
// 用户条目落盘成功即下发：这一轮从"接单"到"起 codex"之间的一切失败（连不上
// app-server、执行器未通过验收、历史注入失败）都靠它把失败说明挂回自己那一轮；晚到
// `turn/start` 之后才发，这些失败就没有用户条目可挂，界面会把说明显示在用户消息上面。
crate::agent::codex_app_server::emit_direct_thread_user_item(root, &canonical_user_item);
let capture = crate::analytics::gui::capture_writer_context();
let root = root.to_path_buf();
tauri::async_runtime::spawn(async move {
@@ -263,6 +267,77 @@ mod tests {
assert_eq!(user_item_id.as_deref(), Some("direct-codex:turn-1:user"));
}
/// 本轮的开口用户条目必须先于整轮里任何可能失败的东西下发。
///
/// 现场（用户可见的坏体验）：命令接单、用户条目落盘之后，整轮在 `turn/start` 之前就失败
/// （连不上 app-server 一类）。这时如果用户条目还没下发，界面就只剩一条失败说明——它按位置
/// 落进**上一轮**的分区里，于是"错误显示在用户消息上面"、上一轮顶替本轮显示耗时，本轮的用户
/// 气泡再自成一个 0.0 秒的假回合。
///
/// 判据取事件流的前两条：命令体是顺序执行的，后台整轮是它之后才起的，所以"开始 → 用户条目"
/// 一定在最前面，之后才可能有失败终态。
#[tokio::test]
async fn the_opening_user_item_is_emitted_before_anything_that_can_fail_in_the_turn() {
let temp = tempfile::tempdir().expect("temp dir");
let root = temp.path().join("direct-user-item-first");
crate::init_local_game_project_at(&root, "direct-user-item-first", "用户条目先下发")
.expect("init project");
let thread_id = direct_thread_id_for_project(&root);
let subscription = subscribe_direct_thread(&thread_id);
let _ = consume_direct_thread(&subscription.subscription_id);
let user_item: DirectCodexUserItem = serde_json::from_value(serde_json::json!({
"type": "message",
"role": "user",
"id": "direct-codex:turn-1:user",
"content": [{ "type": "input_text", "text": "hello" }],
}))
.expect("canonical user item");
chat_with_game_creator_direct_codex_typed(
&root,
user_item,
None,
Some("turn-1".to_string()),
None,
)
.await
.expect("接单成立：命令只回报接单");
let events = consume_direct_thread(&subscription.subscription_id)
.expect("consume logical turn")
.events;
assert!(
matches!(
events.first(),
Some(DirectThreadEvent::TurnStarted { user_item_id, .. })
if user_item_id.as_deref() == Some("direct-codex:turn-1:user")
),
"第一条必须是带身份的回合开始：{events:?}"
);
assert!(
matches!(
events.get(1),
Some(DirectThreadEvent::ItemCompleted { item, .. })
if item.item_id() == "direct-codex:turn-1:user"
),
"第二条必须是本轮的开口用户条目：{events:?}"
);
let terminal = events
.iter()
.position(|event| matches!(event, DirectThreadEvent::TurnCompleted { .. }));
assert!(
terminal.is_none_or(|index| index > 1),
"终态只能在用户条目之后：{events:?}"
);
// 落盘与下发同一份身份：历史里的条目 id 就是事件里的 itemId。
let persisted = std::fs::read_to_string(root.join(".agent/conversations/project.jsonl"))
.expect("read project history");
assert!(
persisted.contains("direct-codex:turn-1:user"),
"用户条目必须已经落盘：{persisted}"
);
}
/// 接单之后的落盘失败：**只走占用对象的失败终态**，命令返回 `Ok`。
///
/// 这条路径的 `turn.started` 已经发过，命令再回一个 `Err` 就等于同一个失败下发两次（事件一条