审批判据改走全量扫描，不再拿有界尾窗证明「不存在」

上一版把尾窗的截断位改成 fail-closed，方向错了。这次连同它一起修掉根因。新增 read_agent_db_records_matching：只读流式全量扫描，上限与写侧幂等扫描完全一致（256 MiB / 100 万条）。读得到的范围从此等于写得进的范围，尾窗那道两个数量级的缺口在这两条判据上不复存在。 - approval_terminal_observation_exists_locked 改走全量，恢复返回 Ok(found)。它在 project_generic_submit_observation_locked 里是「这次是不是重放」的探针，首次审批时 false 就是正确答案，紧接着才会去写 terminal observation。上一版在那里报 Err，`?` 直接抛出，append_agent_db_terminal_observation_if_missing_for_action 永远到不了，调用方置 recoveryPending 后重试同构——审批永久停摆。而且 truncated 也会被崩溃残留的残缺末行置位，而有界读不修尾巴，所以任意大小的项目都能踩中。 - plan_gdd_decision_audit_state_locked 改走全量。它的过滤器只到 (gddId, version)，比写侧幂等键 (gddId, version, responseId) 少一格，是故意的：同版本不同 responseId 的两条 audit 写侧不拦、能共存，这里就是那个跨 responseId 的冲突检测器。上一版靠「命中即成立」跳过截断，等于让滑出窗口的旧冲突隐形，收束闸会错误放行。用例改名 approval_judgements_scan_the_whole_agent_db_not_just_the_tail_window，把冲突副本放在填充之前（窗口之外），并自检尾窗确实截断、副本确实在窗口外。两条改动分别退回旧行为都会让它失败。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-25 03:50:54 +00:00
parent 5c8d894c8a
commit 33ee790b92
2 changed files with 178 additions and 88 deletions
@@ -536,9 +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, scan_truncated) =
-        crate::project::read_agent_db_records_bounded(root, 16 * 1024 * 1024)?;
-    let found = 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)
@@ -557,25 +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())
-    });
-    if found {
-        return Ok(true);
-    }
-    // 尾窗只能证明「在」，不能证明「不在」：`read_agent_db_records_bounded` 保留的是
-    // 最新的一段，被丢掉的永远是更旧的记录。所以命中即成立，与截断无关；但一次没命
-    // 中的截断扫描是「没看见」，不是「没写过」。把它报成 Ok(false)，等于让调用方拿一
-    // 次视野缺失当审批从未被消费——已经收口的 run 会被当成恢复缺口反复重放。
-    //
-    // 不确定性必须交回调用方，而不是在这里替它猜：
-    // `project_generic_submit_observation_locked` 的 Err 分支会把它记成一条具名
-    // projection gap 并置 recoveryPending，收束闸的 Err 分支会原样带进 blocker detail。
-    if scan_truncated {
-        return Err(format!(
-            "Agent DB 尾窗扫描被截断，无法确认 Fast GDD terminal observation：runId={} actionId={}",
-            receipt.run_id, receipt.pending_action_id
-        ));
-    }
-    Ok(false)
+    })?;
+    Ok(!matches.is_empty())
 }

 fn project_generic_submit_runtime_observation_locked(
@@ -1616,19 +1603,21 @@ fn plan_gdd_decision_audit_state_locked(
    receipt: &PlanGddApprovalV1,
 ) -> Result<bool, String> {
    let expected = plan_gdd_decision_audit_value(receipt)?;
-    let (records, scan_truncated) =
-        crate::project::read_agent_db_records_bounded(root, 16 * 1024 * 1024)?;
+    // 过滤器只到 (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");
@@ -1639,28 +1628,7 @@ fn plan_gdd_decision_audit_state_locked(
        }
        found = true;
    }
-    if found {
-        return Ok(true);
-    }
-    // 这道检查原先放在遍历之前、且不看有没有命中就无条件 Err。agent.db 是纯 append
-    // JSONL，全仓没有任何 rotation/compaction，尾窗的记录数上限
-    // （`AGENT_DB_MAX_BOUNDED_RECORDS`）又远低于写侧的扫描上限，于是任何写够记录的项
-    // 目都会在这里一去不回：audit 明明就在窗口里，函数照样报错，收束闸把它翻成
-    // needs-reconciliation，文案还指向「identity 不一致」这个完全错误的方向。
-    //
-    // 截断掐的是头，而 audit 是决定那一刻刚落盘的最新记录——命中即成立。只有没命中
-    // 时截断才有意义：那时才真的分不清「没写过」和「没看见」。
-    //
-    // 代价是命中且截断时不再能断言唯一性（更旧的冲突副本可能已滑出窗口）。本函数的
-    // 契约是「audit 已落盘且看得见的都一致」，不是「全库唯一」；用一个永久停摆去换那
-    // 点额外强度不划算。
-    if scan_truncated {
-        return Err(format!(
-            "Agent DB 尾窗扫描被截断，无法确认 plan GDD decision audit：gddId={} version={}",
-            receipt.gdd_id, receipt.version
-        ));
-    }
-    Ok(false)
+    Ok(found)
 }

 fn plan_gdd_pending_identity_matches(
@@ -2328,24 +2296,28 @@ mod tests {
        let _ = fs::remove_dir_all(root);
    }

-    /// 尾窗扫描被截断，不得把「还在窗口里的记录」判成缺失。
+    /// 两条审批判据必须扫全量 agent.db，不能只看有界尾窗。
    ///
-    /// agent.db 是纯 append、全仓没有 rotation，而尾窗的记录数上限
-    /// （`AGENT_DB_MAX_BOUNDED_RECORDS`）比写侧的扫描上限低两个数量级：项目写够记录之
-    /// 后，每一次有界读都带着 `truncated`。截断掐掉的永远是更旧的记录，而审批 audit 和
-    /// terminal observation 都是决定那一刻刚落盘的最新记录——命中就必须成立。
+    /// 尾窗保留的是最新的一段，能证明「在」，证明不了「不在」，也看不见滑出窗口的更旧
+    /// 记录。两条判据都受不了这种半盲的答案，但受不了的方式相反：
    ///
-    /// 这两处判据原先一个在遍历前无条件 Err、一个直接把截断位丢给 `_`，于是写够记录的
-    /// 项目会在收束闸上永久停在 needs-reconciliation，文案还指向「identity 不一致」。
+    /// - terminal observation 判据在 `project_generic_submit_observation_locked` 里当
+    ///   「这次是不是重放」的探针用，首次审批时必须拿到可信的 `false`，紧接着才会去写
+    ///   observation。在那里 fail closed 会把首次审批拦在写之前，重试同构，永久停摆。
+    /// - decision audit 判据的过滤器只到 (gddId, version)，比写侧幂等键少一个
+    ///   responseId，就是为了抓同版本不同 responseId 的冲突副本；旧冲突一旦被挤出窗口，
+    ///   收束闸就会错误放行。
+    ///
+    /// 所以 fixture 把冲突副本放在填充**之前**（窗口之外），把正常记录放在填充之后。
    #[test]
-    fn truncated_tail_scan_still_confirms_records_inside_the_window() {
+    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-truncation-{}",
+            "genarrative-plan-approval-fullscan-{}",
            uuid::Uuid::new_v4().simple()
        ));
-        init_local_game_project_at(&root, "project-test-approval", "approval truncation")
+        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(),
@@ -2368,16 +2340,25 @@ mod tests {
            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();

-        // 先把尾窗撑过截断线，再落审批记录：这样待证记录一定还在窗口里，而扫描一定是
-        // 截断的——正是「无法证明不存在」被误当成「不存在」的那个状态。填充走 append，
-        // 不覆盖 init 已经写下的内容。
+        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)
-            .map(|index| {
-                format!("{{\"recordType\":\"planning-truncation-filler\",\"i\":{index}}}\n")
-            })
+        let filler = (0..crate::project::AGENT_DB_MAX_BOUNDED_RECORDS + 64)
+            .map(|index| format!("{{\"recordType\":\"planning-fullscan-filler\",\"i\":{index}}}\n"))
            .collect::<String>();
        fs::OpenOptions::new()
            .create(true)
@@ -2387,11 +2368,13 @@ mod tests {
            .write_all(filler.as_bytes())
            .expect("pad agent db past the bounded record window");

-        crate::project::append_agent_db_plan_gdd_decision_if_missing(
-            &root,
-            plan_gdd_decision_audit_value(&receipt).expect("build decision audit"),
-        )
-        .expect("append decision audit");
+        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,
@@ -2413,22 +2396,36 @@ mod tests {
        )
        .expect("append terminal observation");

-        assert!(
+        // fixture 自检：尾窗必须真的截断，且冲突副本必须真的在窗口之外。少了这两条，
+        // 下面的断言在旧实现上也能过，用例就什么都没证明。
+        let (window, window_truncated) =
            crate::project::read_agent_db_records_bounded(&root, 16 * 1024 * 1024)
-                .expect("bounded read")
-                .1,
-            "fixture 必须真的把尾窗撑到截断，否则这条用例什么都没测"
+                .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!(approval_terminal_observation_exists_locked(&root, &receipt)
-            .expect("terminal observation 命中时不得因截断报错"));

-        // 真的没命中时，截断扫描只能报「无法确认」。报成 Ok(false) 会让调用方拿一次视
-        // 野缺失当恢复缺口，反复重放一个其实已经收口的审批。
+        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).is_err());
+        assert!(!approval_terminal_observation_exists_locked(&root, &absent)
+            .expect("首次审批探针不得因为扫描范围报错"));

        let _ = fs::remove_dir_all(root);
    }
@@ -3253,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<Vec<serde_json::Value>, 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::<serde_json::Value>(&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)>,