From a0efd921929db09e60d572e04697d705b5c939e5 Mon Sep 17 00:00:00 2001 From: Suzumiya Date: Tue, 15 Sep 2026 18:17:54 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=B7=A5=E5=85=B7=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E8=90=BD=E7=9B=98=E7=8A=B6=E6=80=81=E6=97=A0=E5=8D=95?= =?UTF-8?q?=E8=B0=83=E6=80=A7=EF=BC=88P1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - direct_tool_calls.rs：把 preserve_started_at（只管 startedAt）换成 merge_tool_call_snapshot， 同一 id 的多条快照按 updatedAt 单调合并：updatedAt 更旧的快照不得覆盖更新的 status 与 updatedAt， updatedAt 相同时终态优先，startedAt 仍取最早的非零值 - upsert_direct_tool_call_once（逐条快照落盘）与 persist_direct_tool_calls_at（回合末整批落盘） 两条竞争路径共用同一合并函数，后到的旧快照不再把 completed / failed 打回 running - normalize_tool_calls（回读归并）同 id 的重复行同样按 updatedAt 合并，磁盘上留下的旧快照不得回退状态 - 新增单测：tool_call_persist_keeps_newest_snapshot_per_item（单条 + 整批两条路径）、 tool_call_read_merges_duplicate_rows_monotonically（回读重复行） --- .../src-tauri/src/agent/direct_tool_calls.rs | 167 +++++++++++++++--- 1 file changed, 147 insertions(+), 20 deletions(-) diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs index 38cca0e22..7be534b10 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs @@ -346,11 +346,16 @@ fn read_tool_call_lines(path: &Path) -> Vec { calls } -/// 按 id 归并（后写覆盖先写），再按时间正序裁剪到最近 `DIRECT_TOOL_CALL_LIMIT` 条。 +/// 按 id 归并（同 id 按 `updatedAt` 单调合并），再按时间正序裁剪到最近 +/// `DIRECT_TOOL_CALL_LIMIT` 条。 fn normalize_tool_calls(calls: Vec) -> Vec { let mut by_id: BTreeMap = BTreeMap::new(); for call in calls { - by_id.insert(call.id.clone(), call); + let merged = match by_id.remove(&call.id) { + Some(previous) => merge_tool_call_snapshot(&previous, &call), + None => call, + }; + by_id.insert(merged.id.clone(), merged); } let mut normalized = by_id.into_values().collect::>(); normalized.sort_by(|left, right| { @@ -373,20 +378,43 @@ pub(crate) fn read_direct_tool_calls_at(root: &Path) -> Result) { - let Some(existing) = existing else { - return; - }; - if existing.started_at > 0 - && (incoming.started_at == 0 || existing.started_at < incoming.started_at) - { - incoming.started_at = existing.started_at; +/// 状态的「确定性」排序：终态（`completed` / `failed`）优先于 `running`。 +fn status_certainty(status: &str) -> u8 { + match status { + "completed" | "failed" => 1, + _ => 0, } } -/// 幂等 upsert：同 id 只保留一行，`completed` 覆盖 `started`。 +/// 同一 id 的两条快照按 `updatedAt` 做**单调合并**。 +/// +/// - `startedAt` 取最早的非零值：`item/completed` 事件不一定带 `startedAtMs`， +/// 不能让 completed 覆盖掉 started 记下的起点（卡片时间序依赖它）。 +/// - `updatedAt` 更旧的快照不得覆盖更新的状态与 `updatedAt`：`direct_runtime.rs` 里 +/// 「回合末整批落盘」与「逐条快照落盘（spawn_blocking）」两条路径竞争时，后到的 +/// 旧快照不能把已经 `completed` / `failed` 的卡片打回 `running`。 +/// - `updatedAt` 相同时终态优先，避免同一毫秒内的旧快照回退状态。 +fn merge_tool_call_snapshot( + existing: &DirectToolCall, + incoming: &DirectToolCall, +) -> DirectToolCall { + let take_incoming = incoming.updated_at > existing.updated_at + || (incoming.updated_at == existing.updated_at + && status_certainty(&incoming.status) > status_certainty(&existing.status)); + let mut merged = if take_incoming { + incoming.clone() + } else { + existing.clone() + }; + merged.started_at = [merged.started_at, existing.started_at, incoming.started_at] + .into_iter() + .filter(|started_at| *started_at > 0) + .min() + .unwrap_or_default(); + merged +} + +/// 幂等 upsert：同 id 只保留一行，快照按 `updatedAt` 单调合并（旧快照不得回退状态）。 /// /// 单次尝试的顺序是「取项目锁 + append 锁 → 锁内读 → 整文件原子替换」。 /// 工具调用是**追加 + 就地更新**混用的数据，没有纯追加的 JSONL 语义，所以只能整文件重写； @@ -404,9 +432,11 @@ fn upsert_direct_tool_call_once(root: &Path, call: &DirectToolCall) -> Result<() .iter() .find(|existing| existing.id == call.id) .cloned(); - let mut incoming = call.clone(); - preserve_started_at(&mut incoming, existing.as_ref()); - calls.retain(|existing| existing.id != call.id); + let incoming = match existing.as_ref() { + Some(existing) => merge_tool_call_snapshot(existing, call), + None => call.clone(), + }; + calls.retain(|existing| existing.id != incoming.id); calls.push(incoming); let normalized = normalize_tool_calls(calls); let mut body = String::new(); @@ -445,11 +475,13 @@ pub(crate) fn persist_direct_tool_calls_at( let mut existing = read_tool_call_lines(&path); let mut incoming = calls.to_vec(); for call in incoming.iter_mut() { - let previous = existing + let merged = existing .iter() - .find(|existing| existing.id == call.id) - .cloned(); - preserve_started_at(call, previous.as_ref()); + .find(|row| row.id == call.id) + .map(|previous| merge_tool_call_snapshot(previous, call)); + if let Some(merged) = merged { + *call = merged; + } } let ids = incoming .iter() @@ -484,6 +516,26 @@ mod tests { }; use serde_json::json; + /// 一行合法的落盘信封（回读用例的夹具）。 + fn tool_call_row(id: &str, started_at: u64, updated_at: u64) -> String { + serde_json::to_string(&json!({ + "type": "tool_call_item", + "payload": { + "schemaVersion": "agc-tool-call.v1", + "id": id, + "turnId": "turn-1", + "kind": "command", + "title": "执行命令", + "summary": "npm run build", + "status": "completed", + "detail": {"command": "npm run build"}, + "startedAt": started_at, + "updatedAt": updated_at + } + })) + .expect("serialize tool call row") + } + fn init_tool_call_project(name: &str) -> tempfile::TempDir { let root = tempfile::tempdir().expect("temp project"); crate::init_local_game_project_at(root.path(), name, "工具调用卡片测试") @@ -875,4 +927,79 @@ mod tests { assert_eq!(sanitize(&once), once, "脱敏必须幂等"); } + /// 判据：同 id 的快照按 `updatedAt` 单调合并——后到的旧快照不得把终态打回 `running`， + /// 也不得回退 `updatedAt`；`startedAt` 仍取最早。 + #[test] + fn tool_call_persist_keeps_newest_snapshot_per_item() { + let root = init_tool_call_project("tool-call-monotonic"); + let running = direct_tool_call_from_item( + root.path(), + &command_item("item-1", "npm run build"), + "turn-1", + false, + 1000, + ) + .expect("running tool call"); + assert_eq!(running.status, "running"); + let completed = direct_tool_call_from_item( + root.path(), + &json!({ + "id": "item-1", + "type": "commandExecution", + "command": "npm run build", + "exitCode": 0, + "completedAtMs": 2000, + }), + "turn-1", + true, + 2000, + ) + .expect("completed tool call"); + assert_eq!(completed.status, "completed"); + assert_eq!(completed.updated_at, 2000); + + persist_direct_tool_call_at(root.path(), &completed).expect("persist completed first"); + persist_direct_tool_call_at(root.path(), &running).expect("persist stale running"); + let calls = read_direct_tool_calls_at(root.path()).expect("read after stale single write"); + assert_eq!(calls.len(), 1, "同一 id 只能有一行"); + assert_eq!( + calls[0].status, "completed", + "后到的旧快照不得把 completed 打回 running" + ); + assert_eq!(calls[0].updated_at, 2000, "旧快照不得回退 updatedAt"); + assert_eq!(calls[0].started_at, 1000, "startedAt 仍取最早"); + + // 回合末整批落盘那条路径同样不得回退。 + persist_direct_tool_calls_at(root.path(), std::slice::from_ref(&running)) + .expect("persist stale running batch"); + let calls = read_direct_tool_calls_at(root.path()).expect("read after stale batch write"); + assert_eq!( + calls[0].status, "completed", + "整批落盘路径同样不得把 completed 打回 running" + ); + assert_eq!(calls[0].updated_at, 2000, "整批落盘不得回退 updatedAt"); + } + + /// 判据：读回时同 id 的重复行也按 `updatedAt` 单调合并（磁盘上留有旧快照不得回退状态）。 + #[test] + fn tool_call_read_merges_duplicate_rows_monotonically() { + let root = init_tool_call_project("tool-call-read-monotonic"); + let path = tool_calls_path(root.path()); + std::fs::create_dir_all(path.parent().expect("parent")).expect("create dir"); + let completed = tool_call_row("item-1", 1000, 2000); + let stale_running = tool_call_row("item-1", 1000, 1000) + .replace("\"status\":\"completed\"", "\"status\":\"running\""); + assert!(stale_running.contains("\"status\":\"running\"")); + std::fs::write(&path, format!("{completed}\n{stale_running}\n")).expect("write fixture"); + + let calls = read_direct_tool_calls_at(root.path()).expect("read duplicate rows"); + assert_eq!(calls.len(), 1, "同 id 归并成一条"); + assert_eq!( + calls[0].status, "completed", + "磁盘上更旧的快照不得把状态打回 running" + ); + assert_eq!(calls[0].updated_at, 2000, "归并保留更新的 updatedAt"); + assert_eq!(calls[0].started_at, 1000); + } + }