From 1affb5eec3003e883a53227136d50656a1ad6c9e Mon Sep 17 00:00:00 2001 From: Suzumiya Date: Sat, 12 Sep 2026 00:07:04 +0800 Subject: [PATCH] =?UTF-8?q?=E9=94=81=E5=86=85=E5=8F=AA=E4=BF=9D=E7=95=99?= =?UTF-8?q?=20append=EF=BC=9ADirectProject=20=E5=8E=86=E5=8F=B2=E7=9A=84?= =?UTF-8?q?=E5=B9=82=E7=AD=89=E5=9B=9E=E6=89=AB=E7=A7=BB=E5=87=BA=E6=8C=81?= =?UTF-8?q?=E9=94=81=E7=AA=97=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - direct_project_history.rs：幂等回扫与序列化移到取锁之前，取锁后常态只做一次追加，不再在锁内读完并逐行解析整份历史 - direct_project_history.rs：回扫到取锁之间若历史文件变了（len/mtime 与回扫时不一致）才回到锁内重扫一次；同一 id 不写第二行的语义不放宽，只是常态不再为它读整份文件 - direct_project_history.rs：新增 direct_project_history_duplicate_at 三态判定（Absent/Identical/Conflict）与回扫状态快照，对外行为与原实现一致（幂等 no-op、id 冲突失败关闭） - direct_project_history.rs：新增回扫探针（两个时点回调）与三条用例：回扫在 append 锁外、锁外回扫后并发追加仍不重复写、id 冲突仍失败关闭 - project/agent_db.rs：新增测试专用探针 project_append_os_lock_is_held_for_test，用与生产同一套打开方式判断追加写目标的锁此刻是否被持有 - 变异验证：把回扫塞回锁内后 idempotency_reverse_scan_runs_outside_the_append_lock 变红（exit 101），报出「幂等回扫必须发生在 append 锁外」，验证后已还原 --- .../src/agent/direct_project_history.rs | 272 +++++++++++++++++- .../src-tauri/src/project/agent_db.rs | 20 ++ 2 files changed, 280 insertions(+), 12 deletions(-) diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs index a4f40e846..316118136 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_history.rs @@ -4,10 +4,15 @@ use crate::project::{ }; use crate::{LocalConversationMessageRecord, LocalConversationResult}; use serde_json::Value; -use std::fs::File; +use std::fs::{self, File}; use std::io::{BufRead, BufReader, Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; +/// 回扫探针时点。生产编译后 `fire_direct_project_history_scan_probe` 是空实现， +/// 只有测试会在这两个时点挂回调（见 `direct_project_history_scan_probe`）。 +const DIRECT_PROJECT_HISTORY_SCAN_PROBE_STARTED: usize = 0; +const DIRECT_PROJECT_HISTORY_SCAN_PROBE_FINISHED: usize = 1; + const DIRECT_PROJECT_HISTORY_RECORD_TYPE: &str = "response_item"; /// 格式切换到 `response_item`（#282）之前，DirectProject 主对话通过通用对话写入器 /// 落到同一份 `project.jsonl`，行形状是 `PersistedLocalConversationMessageRecord`。 @@ -96,6 +101,7 @@ fn find_direct_project_history_item_by_id_at( path: &Path, item_id: &str, ) -> Result, String> { + fire_direct_project_history_scan_probe(DIRECT_PROJECT_HISTORY_SCAN_PROBE_STARTED); let mut file = File::open(path) .map_err(|error| format!("打开 DirectProject 历史失败：{}: {error}", path.display()))?; let mut position = file @@ -246,6 +252,86 @@ fn direct_project_legacy_row(parsed: &Value) -> Option { )) } +/// 测试专用探针：在"锁外幂等回扫"这一段的两端回调。 +/// +/// 用来断言两件在别处无法观测的事实：① 回扫发生时 append 锁**不在**本进程手里； +/// ② 回扫结束到真正 append 之间若发生并发追加，代码会回到锁内重扫。 +#[cfg(test)] +mod direct_project_history_scan_probe { + use std::sync::Mutex; + + pub(super) const SCAN_STARTED: usize = super::DIRECT_PROJECT_HISTORY_SCAN_PROBE_STARTED; + pub(super) const SCAN_FINISHED: usize = super::DIRECT_PROJECT_HISTORY_SCAN_PROBE_FINISHED; + + static PROBES: Mutex<[Option>; 2]> = Mutex::new([None, None]); + + pub(super) fn install(slot: usize, probe: impl Fn() + Send + 'static) { + PROBES.lock().expect("history scan probe lock")[slot] = Some(Box::new(probe)); + } + + pub(super) fn fire(slot: usize) { + let probe = PROBES.lock().expect("history scan probe lock")[slot].take(); + if let Some(probe) = probe { + probe(); + } + } +} + +#[cfg(test)] +fn fire_direct_project_history_scan_probe(slot: usize) { + direct_project_history_scan_probe::fire(slot); +} + +#[cfg(not(test))] +fn fire_direct_project_history_scan_probe(_slot: usize) {} + +/// 回扫读到的历史文件状态。 +/// +/// 锁外回扫之后要复核"文件还是不是回扫时那一份"：只有一致才能沿用"这个 id 不在历史里" +/// 的结论；不一致就回到锁内重扫一次（罕见）。于是锁内常态只剩 append。 +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +struct DirectProjectHistoryScanState { + len: u64, + modified: Option, +} + +fn direct_project_history_scan_state_at( + path: &Path, +) -> Result { + let metadata = fs::metadata(path).map_err(|error| { + format!( + "读取 DirectProject 历史元数据失败：{}: {error}", + path.display() + ) + })?; + Ok(DirectProjectHistoryScanState { + len: metadata.len(), + modified: metadata.modified().ok(), + }) +} + +/// 幂等判定结果。`Absent` 表示历史里没有这个 id，可以继续追加。 +enum DirectProjectHistoryDuplicate { + Absent, + Identical, + Conflict, +} + +fn direct_project_history_duplicate_at( + path: &Path, + item: &Value, + item_id: &str, +) -> Result { + if !prepare_game_creator_private_path_for_read(path, false, "DirectProject 历史")? { + return Ok(DirectProjectHistoryDuplicate::Absent); + } + match find_direct_project_history_item_by_id_at(path, item_id)? { + None => Ok(DirectProjectHistoryDuplicate::Absent), + Some(existing) if &existing == item => Ok(DirectProjectHistoryDuplicate::Identical), + Some(_) => Ok(DirectProjectHistoryDuplicate::Conflict), + } +} + pub(crate) fn append_direct_project_history_item_at( root: &Path, item: &Value, @@ -275,23 +361,52 @@ fn append_direct_project_history_item_at_with_user_policy( if !allow_user_item && is_direct_project_codex_user_item(item) { return Ok(()); } - let _project_lock = crate::project::acquire_project_write_lock(root, "conversation.write")?; let path = history_path(root); let history_exists = prepare_game_creator_private_path_for_read(&path, false, "DirectProject 历史")?; + let item_id = item.get("id").and_then(Value::as_str); + // 幂等回扫与序列化都留在锁外。回扫（`find_direct_project_history_item_by_id_at`） + // 会从文件尾逐行解析整份历史，新条目永远命不中，所以它天然是 O(文件大小) 的重活； + // 历史到 MB 级时把它放进锁内，等于让另一个进程（`--agent-runner` 的 prompt 上下文、 + // MCP 宿主）等一次"读完整份历史"，而不是等一次追加。 + let scanned_state = if history_exists { + let state = direct_project_history_scan_state_at(&path)?; + if let Some(item_id) = item_id { + match direct_project_history_duplicate_at(&path, item, item_id)? { + DirectProjectHistoryDuplicate::Identical => return Ok(()), + DirectProjectHistoryDuplicate::Conflict => { + return Err(format!("DirectProject 历史 item id 冲突：{item_id}")) + } + DirectProjectHistoryDuplicate::Absent => {} + } + } + Some(state) + } else { + None + }; + fire_direct_project_history_scan_probe(DIRECT_PROJECT_HISTORY_SCAN_PROBE_FINISHED); + let line = record(item)?; + let _project_lock = crate::project::acquire_project_write_lock(root, "conversation.write")?; let lock = project_append_lock_for(&path)?; let _append_guard = lock.lock("DirectProject 历史追加写")?; - if history_exists { - if let Some(item_id) = item.get("id").and_then(Value::as_str) { - if let Some(existing) = find_direct_project_history_item_by_id_at(&path, item_id)? { - if &existing == item { - return Ok(()); + // 回扫到取锁之间可能有别的进程（或本项目另一条链）追加过：只有文件状态仍与回扫时一致 + // 才能沿用"这个 id 不在历史里"的结论，否则回到锁内重扫一次。 + // "同一个 id 不写第二行"的语义因此没有被放宽，只是常态不再为它读整份文件。 + if let Some(item_id) = item_id { + let scan_is_stale = match scanned_state { + Some(state) => direct_project_history_scan_state_at(&path)? != state, + None => path.exists(), + }; + if scan_is_stale { + match direct_project_history_duplicate_at(&path, item, item_id)? { + DirectProjectHistoryDuplicate::Identical => return Ok(()), + DirectProjectHistoryDuplicate::Conflict => { + return Err(format!("DirectProject 历史 item id 冲突：{item_id}")) } - return Err(format!("DirectProject 历史 item id 冲突：{item_id}")); + DirectProjectHistoryDuplicate::Absent => {} } } } - let line = record(item)?; append_jsonl_line_unlocked(&path, &line, "DirectProject 历史") } @@ -429,11 +544,13 @@ pub(crate) fn read_direct_project_chat_history_at( #[cfg(test)] mod tests { use super::{ - append_direct_project_history_item_at, append_direct_project_user_message_at, history_path, - is_direct_project_internal_context_item, read_direct_project_chat_history_at, - read_direct_project_history_items_at, + append_direct_project_history_item_at, append_direct_project_user_message_at, + direct_project_history_scan_probe, history_path, is_direct_project_internal_context_item, + read_direct_project_chat_history_at, read_direct_project_history_items_at, }; use serde_json::{json, Value}; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; fn init_history_project(name: &str) -> tempfile::TempDir { let root = tempfile::tempdir().expect("temp project"); @@ -452,6 +569,137 @@ mod tests { const LEGACY_ASSISTANT_ROW: &str = r#"{"schemaVersion":"game-creator-conversation.v1","role":"assistant","content":"已完成 第一行\n第二行 ","agentId":null,"updatedAt":1757000001}"#; const RESPONSE_ITEM_ROW: &str = r#"{"type":"response_item","payload":{"type":"message","role":"user","id":"codex-item-2","content":[{"type":"input_text","text":"再加一个按钮"}]}}"#; + /// 判据：幂等回扫发生时，append 锁**不能**在本进程手里。 + /// + /// 变异验证（把回扫挪回锁内）会把这条用例打成红的：探针在回扫时点用与生产同一套 + /// 打开方式（Windows 零共享 / Unix flock 非阻塞）去探同一把锁，持着就探不到。 + #[test] + fn idempotency_reverse_scan_runs_outside_the_append_lock() { + let root = init_history_project("scan-outside-lock"); + write_history_lines( + root.path(), + &[LEGACY_USER_ROW, RESPONSE_ITEM_ROW, LEGACY_ASSISTANT_ROW], + ); + let path = history_path(root.path()); + let fired = Arc::new(AtomicBool::new(false)); + let lock_held = Arc::new(AtomicBool::new(false)); + let fired_probe = Arc::clone(&fired); + let held_probe = Arc::clone(&lock_held); + let probe_path = path.clone(); + direct_project_history_scan_probe::install( + direct_project_history_scan_probe::SCAN_STARTED, + move || { + let held = crate::project::project_append_os_lock_is_held_for_test(&probe_path) + .expect("probe append lock state"); + fired_probe.store(true, Ordering::SeqCst); + held_probe.store(held, Ordering::SeqCst); + }, + ); + + append_direct_project_history_item_at( + root.path(), + &json!({ + "type": "message", + "role": "assistant", + "id": "scan-window-item", + "content": [{"type": "output_text", "text": "锁外回扫"}] + }), + ) + .expect("append item"); + + assert!( + fired.load(Ordering::SeqCst), + "幂等回扫必须真的发生，否则本用例只是空跑" + ); + assert!( + !lock_held.load(Ordering::SeqCst), + "幂等回扫必须发生在 append 锁外：持锁读整份历史会把持锁窗口从一次追加拉成 O(文件大小)" + ); + } + + /// 判据：锁外回扫之后、真正 append 之前发生的并发追加，必须由"锁内重扫"兜住， + /// 不能因为回扫结果过期就写下第二行同 id 记录。 + #[test] + fn concurrent_append_after_the_out_of_lock_scan_still_prevents_a_duplicate_row() { + let root = init_history_project("scan-stale-state"); + write_history_lines(root.path(), &[LEGACY_USER_ROW]); + let path = history_path(root.path()); + let item = json!({ + "type": "message", + "role": "assistant", + "id": "stale-scan-item", + "content": [{"type": "output_text", "text": "并发追加"}] + }); + let injected_path = path.clone(); + let injected_item = item.clone(); + direct_project_history_scan_probe::install( + direct_project_history_scan_probe::SCAN_FINISHED, + move || { + let line = serde_json::to_string(&json!({ + "type": "response_item", + "payload": injected_item, + })) + .expect("serialize injected history row"); + let mut existing = + std::fs::read_to_string(&injected_path).expect("read history before injection"); + existing.push_str(&line); + existing.push('\n'); + std::fs::write(&injected_path, existing).expect("inject concurrent history row"); + }, + ); + + append_direct_project_history_item_at(root.path(), &item).expect("idempotent append"); + + let items = read_direct_project_history_items_at(root.path()).expect("read history"); + assert_eq!( + items.iter().filter(|existing| *existing == &item).count(), + 1, + "并发追加后必须由锁内重扫判定幂等，不能写下第二行" + ); + } + + /// 判据：同一个 id 但内容不同，在"回扫移到锁外"之后依然失败关闭且不追加。 + #[test] + fn conflicting_item_id_still_fails_closed_with_the_scan_outside_the_lock() { + let root = init_history_project("id-conflict-outside-lock"); + write_history_lines(root.path(), &[LEGACY_USER_ROW]); + append_direct_project_history_item_at( + root.path(), + &json!({ + "type": "message", + "role": "assistant", + "id": "conflict-item", + "content": [{"type": "output_text", "text": "第一次"}] + }), + ) + .expect("append first item"); + + let error = append_direct_project_history_item_at( + root.path(), + &json!({ + "type": "message", + "role": "assistant", + "id": "conflict-item", + "content": [{"type": "output_text", "text": "同一个 id 但内容不同"}] + }), + ) + .expect_err("conflicting item id must fail closed"); + assert!( + error.starts_with("DirectProject 历史 item id 冲突"), + "{error}" + ); + + let items = read_direct_project_history_items_at(root.path()).expect("read history"); + assert_eq!( + items + .iter() + .filter(|item| item.get("id").and_then(Value::as_str) == Some("conflict-item")) + .count(), + 1, + "冲突条目不能追加第二行" + ); + } + #[test] fn filters_host_context_but_keeps_real_user_items() { assert!(is_direct_project_internal_context_item(&json!({ diff --git a/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs b/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs index 08691c487..bb3039585 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/project/agent_db.rs @@ -3623,6 +3623,26 @@ fn acquire_project_append_os_lock(path: &Path, error_label: &str) -> Result Result { + let lock_path = project_append_os_lock_path(path)?; + if !lock_path.exists() { + return Ok(false); + } + match try_open_project_append_os_lock(&lock_path, "追加写锁探针")? { + Some(file) => { + drop(file); + Ok(false) + } + None => Ok(true), + } +} + #[cfg(unix)] fn try_open_project_append_os_lock(path: &Path, error_label: &str) -> Result, String> { use std::os::fd::AsRawFd;