修复残留问题 (#192)
agent.db 是纯 append、全仓没有 rotation，而有界尾窗的记录数上限（16384）比写侧 的扫描上限（100 万）低两个数量级。截断掐掉的永远是更旧的记录，所以「命中」与截 断无关；只有「没命中」时才真的分不清「没写过」和「没看见」。 - plan_gdd_decision_audit_state_locked：截断判断从遍历前的无条件 Err 挪到遍历后、 且只在没命中时报。原先任何写够记录的项目都会永久停在 needs-reconciliation—— audit 明明就在窗口里，函数照样报错，文案还指向「identity 不一致」。 - approval_terminal_observation_exists_locked：不再把截断位丢给 `_`。没命中且截 断时返回 Err 而不是 Ok(false)，交给调用方既有的具名 projection gap 分支，不再 让一次视野缺失冒充恢复缺口被反复重放。 - 收束闸 Err 分支标题改为「Fast GDD decision audit 无法确认」。Err 现在有 identity 冲突和尾窗截断两种来源，具体原因由 detail 里的 error 原文给出。 - AGENT_DB_MAX_BOUNDED_RECORDS 改 pub(crate)，让回归测试能精确把 fixture 撑过截 断线，不必在测试里埋魔数。 新增 truncated_tail_scan_still_confirms_records_inside_the_window 钉住这两处：把 任一处改回旧行为都会让它失败。 recovery_scan.rs 的七槽审计是同样的形状，但它断言的是精确基数而非存在性，截断确 实证不了基数，另行处理。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: 段舒康 <kdletters@qq.com> Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/192 Co-authored-by: 孔令弘 <ink29535@proton.me> Co-committed-by: 孔令弘 <ink29535@proton.me>
This commit was merged in pull request #192.
This commit is contained in:
+160
-18
@@ -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<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)?;
|
||||
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::<String>();
|
||||
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` 读绑定走的是会遍历完整父链的入口，而识别
|
||||
|
||||
@@ -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<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)>,
|
||||
|
||||
Reference in New Issue
Block a user