diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_protocol/planning_approval.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_protocol/planning_approval.rs index 2102a6279..27c300f3d 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_protocol/planning_approval.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_protocol/planning_approval.rs @@ -536,8 +536,13 @@ fn approval_terminal_observation_exists_locked( // deterministically from the immutable receipt and must match exactly // before any surviving submit anchor may be cleaned up. let expected_summary = approval_observation(receipt).summary; - let (records, _) = crate::project::read_agent_db_records_bounded(root, 16 * 1024 * 1024)?; - Ok(records.iter().any(|record| { + // 走全量扫描，不走有界尾窗。这条判据的两个调用点语义并不相同： + // `project_generic_submit_observation_locked` 在 pending 还在时用它当「这次是不是 + // 重放」的探针，首次审批时 `false` 就是正确答案，紧接着才会去写 terminal + // observation。尾窗在没命中时只能二选一，而两个都是错的——报 false 会让真正的重放 + // 被当成未消费，报 Err 会把首次审批拦在写 observation 之前，重试同构，永久停摆。 + // 所以答案必须精确：扫全量，`false` 就真的是没写过。 + let matches = crate::project::read_agent_db_records_matching(root, 8, |record| { record.get("recordType").and_then(serde_json::Value::as_str) == Some("agent.runtime.tool_observation") && record.get("agentId").and_then(serde_json::Value::as_str) @@ -556,7 +561,8 @@ fn approval_terminal_observation_exists_locked( && record.get("decision").and_then(serde_json::Value::as_str) == Some("approval") && record.get("summary").and_then(serde_json::Value::as_str) == Some(expected_summary.as_str()) - })) + })?; + Ok(!matches.is_empty()) } fn project_generic_submit_runtime_observation_locked( @@ -1597,22 +1603,21 @@ fn plan_gdd_decision_audit_state_locked( receipt: &PlanGddApprovalV1, ) -> Result { let expected = plan_gdd_decision_audit_value(receipt)?; - let (records, scan_truncated) = - crate::project::read_agent_db_records_bounded(root, 16 * 1024 * 1024)?; - if scan_truncated { - return Err("Agent DB 扫描被截断，无法确认 plan GDD decision audit".to_string()); - } + // 过滤器只到 (recordType, gddId, version)，比写侧的幂等键 + // (recordType, gddId, version, responseId) 少一格——这是故意的：同版本不同 + // responseId 的两条 audit 写侧根本不拦，能共存，而这里就是那个跨 responseId 的冲 + // 突检测器。所以扫描范围必须是全量：有界尾窗只要把更旧的那条冲突记录挤出去，收束 + // 闸就只看得见新的那条，然后放行。 + let records = crate::project::read_agent_db_records_matching(root, 64, |record| { + record.get("recordType").and_then(serde_json::Value::as_str) + == Some(PLAN_GDD_APPROVAL_DECISION_AUDIT_RECORD_TYPE) + && record.get("gddId").and_then(serde_json::Value::as_str) + == Some(receipt.gdd_id.as_str()) + && record.get("version").and_then(serde_json::Value::as_u64) + == Some(receipt.version as u64) + })?; let mut found = false; for record in records { - if record.get("recordType").and_then(serde_json::Value::as_str) - != Some(PLAN_GDD_APPROVAL_DECISION_AUDIT_RECORD_TYPE) - || record.get("gddId").and_then(serde_json::Value::as_str) - != Some(receipt.gdd_id.as_str()) - || record.get("version").and_then(serde_json::Value::as_u64) - != Some(receipt.version as u64) - { - continue; - } let mut comparable = record; if let Some(object) = comparable.as_object_mut() { object.remove("schemaVersion"); @@ -2185,9 +2190,12 @@ pub(crate) fn plan_gdd_typed_completion_blocker_at_locked( )); } Err(error) => { + // Err 现在有两种来源——identity 冲突，和尾窗截断导致的「无法确认」。标题 + // 只说无法确认，具体原因由 detail 里的 error 原文给出；写死成「identity + // 不一致」会把排查的人按到错误的方向上。 return Some(plan_gdd_completion_blocker( "needs-reconciliation", - "Fast GDD decision audit identity 不一致，不能收束任务", + "Fast GDD decision audit 无法确认，不能收束任务", error, )); } @@ -2288,6 +2296,140 @@ mod tests { let _ = fs::remove_dir_all(root); } + /// 两条审批判据必须扫全量 agent.db，不能只看有界尾窗。 + /// + /// 尾窗保留的是最新的一段，能证明「在」，证明不了「不在」，也看不见滑出窗口的更旧 + /// 记录。两条判据都受不了这种半盲的答案，但受不了的方式相反： + /// + /// - terminal observation 判据在 `project_generic_submit_observation_locked` 里当 + /// 「这次是不是重放」的探针用，首次审批时必须拿到可信的 `false`，紧接着才会去写 + /// observation。在那里 fail closed 会把首次审批拦在写之前，重试同构，永久停摆。 + /// - decision audit 判据的过滤器只到 (gddId, version)，比写侧幂等键少一个 + /// responseId，就是为了抓同版本不同 responseId 的冲突副本；旧冲突一旦被挤出窗口， + /// 收束闸就会错误放行。 + /// + /// 所以 fixture 把冲突副本放在填充**之前**（窗口之外），把正常记录放在填充之后。 + #[test] + fn approval_judgements_scan_the_whole_agent_db_not_just_the_tail_window() { + use std::io::Write; + + let root = std::env::temp_dir().join(format!( + "genarrative-plan-approval-fullscan-{}", + uuid::Uuid::new_v4().simple() + )); + init_local_game_project_at(&root, "project-test-approval", "approval full scan") + .expect("init project"); + let receipt = PlanGddApprovalV1 { + schema_version: PLAN_GDD_APPROVAL_SCHEMA_VERSION.to_string(), + project_id: "project-test-approval".to_string(), + gdd_id: "gdd-00000000-0000-4000-8000-000000000011".to_string(), + version: 1, + fingerprint: format!("sha256-serde-json-v2:{}", "6".repeat(64)), + pending_action_id: "action-0123456789abcdef01234568".to_string(), + action_fingerprint: "7".repeat(64), + approval_request_id: "gdd-approval-00000000-0000-4000-8000-000000000012".to_string(), + response_id: "gdd-response-00000000-0000-4000-8000-000000000013".to_string(), + decision_fingerprint: format!("sha256-serde-json-v2:{}", "8".repeat(64)), + source: PLAN_GDD_APPROVAL_SOURCE.to_string(), + run_profile: AGENT_RUNTIME_RUN_PROFILE_STANDARD.to_string(), + run_profile_binding_fingerprint: "9".repeat(64), + session_id: "session-approval-2".to_string(), + run_id: "run-planning-child-2".to_string(), + action: "approve".to_string(), + comment: None, + decided_at_utc: "2026-08-15T00:00:00.000Z".to_string(), + receipt_fingerprint: format!("sha256-serde-json-v2:{}", "a".repeat(64)), + }; + // 另一个 GDD：它下面会有两条 responseId 不同的 audit。写侧的幂等键带 responseId， + // 两条都能落盘；判据的过滤器不带，所以它必须把这一对认成冲突。 + let mut conflicted = receipt.clone(); + conflicted.gdd_id = "gdd-00000000-0000-4000-8000-000000000021".to_string(); + conflicted.response_id = "gdd-response-00000000-0000-4000-8000-000000000023".to_string(); + let mut stale = conflicted.clone(); + stale.response_id = "gdd-response-00000000-0000-4000-8000-000000000024".to_string(); + + crate::project::append_agent_db_plan_gdd_decision_if_missing( + &root, + plan_gdd_decision_audit_value(&stale).expect("build stale decision audit"), + ) + .expect("append stale decision audit"); + + // 填充把上面那条冲突副本挤出尾窗。走 append，不覆盖 init 已经写下的内容。 + let agent_db = root.join(".agent/agent.db"); + fs::create_dir_all(agent_db.parent().expect("agent db parent")).expect("agent db parent"); + let filler = (0..crate::project::AGENT_DB_MAX_BOUNDED_RECORDS + 64) + .map(|index| format!("{{\"recordType\":\"planning-fullscan-filler\",\"i\":{index}}}\n")) + .collect::(); + fs::OpenOptions::new() + .create(true) + .append(true) + .open(&agent_db) + .expect("open agent db fixture") + .write_all(filler.as_bytes()) + .expect("pad agent db past the bounded record window"); + + for decided in [&receipt, &conflicted] { + crate::project::append_agent_db_plan_gdd_decision_if_missing( + &root, + plan_gdd_decision_audit_value(decided).expect("build decision audit"), + ) + .expect("append decision audit"); + } + let observation = approval_observation(&receipt); + append_agent_db_terminal_observation_if_missing_for_action( + &root, + GAME_CREATOR_PROJECT_PLANNING_AGENT_ID, + &receipt.run_id, + &receipt.pending_action_id, + serde_json::json!({ + "recordType": "agent.runtime.tool_observation", + "agentId": GAME_CREATOR_PROJECT_PLANNING_AGENT_ID, + "taskId": "task-planning-1", + "runId": receipt.run_id, + "actionId": receipt.pending_action_id, + "actionFingerprint": receipt.action_fingerprint, + "tool": PLAN_GDD_APPROVAL_TOOL, + "status": "ok", + "summary": observation.summary, + "decision": "approval", + }), + ) + .expect("append terminal observation"); + + // fixture 自检：尾窗必须真的截断，且冲突副本必须真的在窗口之外。少了这两条， + // 下面的断言在旧实现上也能过，用例就什么都没证明。 + let (window, window_truncated) = + crate::project::read_agent_db_records_bounded(&root, 16 * 1024 * 1024) + .expect("bounded read"); + assert!(window_truncated, "fixture 必须真的把尾窗撑到截断"); + assert!( + !window.iter().any(|record| record + .get("responseId") + .and_then(serde_json::Value::as_str) + == Some(stale.response_id.as_str())), + "冲突副本必须落在尾窗之外，否则证明不了判据扫的是全量" + ); + + assert!(plan_gdd_decision_audit_state_locked(&root, &receipt) + .expect("无冲突的 decision audit 必须直接成立")); + assert!( + plan_gdd_decision_audit_state_locked(&root, &conflicted).is_err(), + "滑出尾窗的同版本冲突 audit 仍然必须被抓到" + ); + assert!(approval_terminal_observation_exists_locked(&root, &receipt) + .expect("已落盘的 terminal observation 必须认得出来")); + + // 没写过就是没写过：这里必须是可信的 Ok(false)，不能是 Err。首次审批走的正是 + // 这条路，在这里 fail closed 会把 append_agent_db_terminal_observation_... + // 拦在后面，重试同构，审批永久停在 recoveryPending。 + let mut absent = receipt.clone(); + absent.run_id = "run-planning-child-absent".to_string(); + assert!(!approval_terminal_observation_exists_locked(&root, &absent) + .expect("首次审批探针不得因为扫描范围报错")); + + let _ = fs::remove_dir_all(root); + } + /// 祖先绑定坏掉不能让一个跟立项策划无关的 supervisor run 被判成「策划根身份不明」。 /// /// `plan_root_completion_identity_at` 读绑定走的是会遍历完整父链的入口，而识别 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 7a26ef189..b9e2f985e 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 @@ -53,7 +53,12 @@ const AGENT_DB_MAX_ORDINARY_APPEND_RECORDS: usize = AGENT_DB_MAX_SCAN_RECORDS - AGENT_DB_PLAN_GDD_DECISION_RESERVE_RECORDS as usize; const AGENT_DB_FINALIZATION_CRITICAL_RECORDS_PER_SEQUENCE: usize = 7; const AGENT_DB_MAX_BOUNDED_READ_BYTES: u64 = 32 * 1024 * 1024; -const AGENT_DB_MAX_BOUNDED_RECORDS: usize = 16_384; +/// 尾窗一次最多返回的记录数。超出即 `truncated`，丢的永远是更旧的记录。 +/// +/// 这个上限比写侧的 `AGENT_DB_MAX_SCAN_RECORDS` 低两个数量级，所以「读得到」的窗口 +/// 远小于「写得进」的容量；任何拿尾窗证明「记录不存在」的判据都要自己处理这段差。 +/// `pub(crate)` 是给判据的回归测试用的——测试要能把 fixture 精确撑过这条线。 +pub(crate) const AGENT_DB_MAX_BOUNDED_RECORDS: usize = 16_384; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum AgentDbRecordAppendClass { @@ -3248,6 +3253,99 @@ pub(crate) fn read_agent_db_records_bounded( Ok((records.into_iter().collect(), truncated)) } +/// 全量扫描 Agent 本地索引，返回全部命中 `predicate` 的记录。 +/// +/// 有界尾窗（`read_agent_db_records_bounded`）保留的是最新的一段：它能证明「在」， +/// 证明不了「不在」，也看不见已经滑出窗口的更旧记录。凡是要拿扫描结果做 fail-closed +/// 判据的调用方——「这条回执消费过没有」「这个 (gddId, version) 下有没有第二条冲突 +/// audit」——都必须走这条路。用尾窗做这种判据只有两种输出，而两种都是错的：命不中就 +/// 报「不存在」会把视野缺失当成事实，命不中就报错会把首次写入拦在写之前。 +/// +/// 扫描上限与写侧的幂等扫描完全一致（`AGENT_DB_MAX_ACTION_RECEIPT_SCAN_BYTES` / +/// `AGENT_DB_MAX_SCAN_RECORDS`），所以只要追加还写得进去，这里就一定扫得完；不会出现 +/// 「写得进但读不到」的窗口——那正是尾窗留下的那道两个数量级的缺口。 +/// +/// `max_matches` 是命中数上限，超出报错而不是静默截断：判据宁可停，也不能拿一个不完 +/// 整的命中集合下结论。 +pub(crate) fn read_agent_db_records_matching( + root: &Path, + max_matches: usize, + predicate: impl Fn(&serde_json::Value) -> bool, +) -> Result, String> { + let path = root.join(".agent/agent.db"); + let Some(directory) = open_agent_db_directory(root, false)? else { + return Ok(Vec::new()); + }; + let append_lock = project_append_lock_for(&path)?; + let _append_guard = append_lock.lock_process("Agent 本地索引")?; + verify_agent_db_directory_current(&directory)?; + let Some(mut storage) = open_agent_db_storage(directory, false, false)? else { + return Ok(Vec::new()); + }; + let length = storage + .file + .metadata() + .map_err(|error| { + format!( + "读取 Agent 本地索引元数据失败：{}: {error}", + storage.path.display() + ) + })? + .len(); + if length > AGENT_DB_MAX_ACTION_RECEIPT_SCAN_BYTES { + return Err(format!( + "Agent 本地索引超过 {} 字节扫描上限：{}", + AGENT_DB_MAX_ACTION_RECEIPT_SCAN_BYTES, + storage.path.display() + )); + } + storage.file.seek(SeekFrom::Start(0)).map_err(|error| { + format!( + "定位 Agent 本地索引失败：{}: {error}", + storage.path.display() + ) + })?; + let mut reader = BufReader::new(&mut storage.file); + let mut matches = Vec::new(); + let mut record_count = 0usize; + while let Some(line) = read_agent_db_jsonl_line_bounded(&mut reader, &storage.path)? { + // 崩溃留下的残缺末行从未提交成功，写侧下一次追加会把它截掉。它不是记录，也不 + // 该让判据 fail closed——扫到这里停住就够了。 + if !line.complete { + break; + } + if line.content.iter().all(|byte| byte.is_ascii_whitespace()) { + continue; + } + record_count = record_count.saturating_add(1); + if record_count > AGENT_DB_MAX_SCAN_RECORDS { + return Err(format!( + "Agent 本地索引超过 {} 条记录扫描上限：{}", + AGENT_DB_MAX_SCAN_RECORDS, + storage.path.display() + )); + } + let record = + serde_json::from_slice::(&line.content).map_err(|error| { + format!( + "解析 Agent 本地索引失败：{}: {error}", + storage.path.display() + ) + })?; + if !predicate(&record) { + continue; + } + if matches.len() >= max_matches { + return Err(format!( + "Agent 本地索引命中记录超过 {max_matches} 条上限：{}", + storage.path.display() + )); + } + matches.push(record); + } + Ok(matches) +} + pub(crate) fn read_agent_db_action_receipts_by_identities( root: &Path, identities: &BTreeSet<(String, String, String)>,