新增: 消息级溯源 P2 切 ai_messages + AI Chat 跑题改进 P0-P2 + 标题诊断 + 苛刻测
消息级溯源 P2(一次性切读 b + 备份保留):
- 批次A 读路径切 ai_messages(state.rs AppState.ai_messages + 映射 ChatMessage↔AiMessageRecord +
switch/export load 切 + fallback 兜底)
- 批次B 写路径切 ai_messages(replace_conversation 单事务全量重写 + save/clear 切 + 旧 messages 备份)
- 摘要/压缩不改 updated_at(save_conversation touch_updated_at,compress false 防时间分组跳变)
AI Chat 跑题改进 P0-P2(5 改进,治跑题 5 根因,三原则优雅/可靠/易迭代):
- P0 系统提示聚焦(## 聚焦准则独立段)+ 意图接入 loop(filter_tool_defs 收敛工具 29→5-10,三重 fallback)
- P1 压缩增强(compress prompt 主题锚点 + 失败关键词兜底 extract_keyword_summary)+
工具结果压缩(should_summarize + extract_key_info view-only 不改持久化)
- P2 主题检测(TrackedMessage.topic + 双高置信保守标记 + tokenize 2-gram 修复中文锚点)
- title LLM complete 失败诊断日志(助定位返空根因)
- 跑题 P0 审查🟡修复(Debug 加 Data 域 + threshold 注释)
苛刻测 38 条(边界/对抗:全漂移/全停用词/全错误行/2KB 边界/连续主题切换/中文 2-gram/emoji)
df-ai 227 passed + workspace EXIT 0
This commit is contained in:
@@ -186,6 +186,61 @@ impl AiMessageRepo {
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 全量重写对话的消息(单事务 DELETE + INSERT OR IGNORE,原子)。
|
||||
///
|
||||
/// F-260619-03 批次 B(save_conversation 写路径切 ai_messages)的核心方法:
|
||||
/// 全量重写语义——以入参 records 为该对话的**唯一真相**,先删该 conv 全部旧行再批量插。
|
||||
/// 单事务保证「删 + 插」原子,无中间空窗(reload 不会读到半删半插的中间态)。
|
||||
///
|
||||
/// 设计权衡(非 dirty 增量,留 P2.1 优化):
|
||||
/// - 内存 ContextManager 是运行时真相源,save 是同步点,每轮 save 全量回写简单可靠;
|
||||
/// - INSERT OR IGNORE 幂等:records 内 id 重复或与残留行(理论不应有,事务已 DELETE)冲突跳过;
|
||||
/// - 入参 records 的 conversation_id 应一致(调用方 save_conversation 保证),本方法不校验。
|
||||
///
|
||||
/// 返回 Ok(()) —— 不返回受影响行数(DELETE + INSERT 两条计数语义混乱,调用方只关心成功)。
|
||||
pub async fn replace_conversation(
|
||||
&self,
|
||||
conv_id: &str,
|
||||
records: Vec<AiMessageRecord>,
|
||||
) -> Result<()> {
|
||||
let conn = self.conn.clone();
|
||||
let conv_id = conv_id.to_owned();
|
||||
tokio::task::spawn_blocking(move || -> Result<()> {
|
||||
let mut guard = conn.blocking_lock();
|
||||
let tx = guard.transaction().map_err(storage_err)?;
|
||||
{
|
||||
// 先删该 conv 全部旧行(全量重写语义)
|
||||
tx.execute(
|
||||
"DELETE FROM ai_messages WHERE conversation_id = ?1",
|
||||
params![conv_id],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
// 再批量插新行(INSERT OR IGNORE 幂等,id 冲突跳过)
|
||||
if !records.is_empty() {
|
||||
let mut stmt = tx.prepare(
|
||||
"INSERT OR IGNORE INTO ai_messages
|
||||
(id, conversation_id, seq, role, content, parts, tool_call_id,
|
||||
tool_calls, model, status, reasoning_content, timestamp, created_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
for rec in &records {
|
||||
stmt.execute(params![
|
||||
rec.id, rec.conversation_id, rec.seq, rec.role, rec.content,
|
||||
rec.parts, rec.tool_call_id, rec.tool_calls, rec.model, rec.status,
|
||||
rec.reasoning_content, rec.timestamp, rec.created_at
|
||||
])
|
||||
.map_err(storage_err)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
tx.commit().map_err(storage_err)?;
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 按 tool_call_id 定点更新消息 content(replace_tool_result_content 用)。
|
||||
///
|
||||
/// 返回是否实际更新(0 = 该 tool_call_id 在此对话无对应消息)。
|
||||
@@ -365,4 +420,162 @@ mod tests {
|
||||
let got = repo.list_by_conversation("c").await.expect("list");
|
||||
assert_eq!(got[0].content, "替换后的结果", "其他消息不应被改");
|
||||
}
|
||||
|
||||
// ---------- replace_conversation(F-260619-03 批次 B)----------
|
||||
|
||||
/// replace_conversation 全量重写:删旧 + 插新原子,list 一致
|
||||
#[tokio::test]
|
||||
async fn replace_conversation_full_rewrite() {
|
||||
let db = Database::open_in_memory().await.expect("open_in_memory");
|
||||
let repo = AiMessageRepo::new(&db);
|
||||
|
||||
// 预置旧数据(将被 replace 删除)
|
||||
repo.insert_batch(vec![
|
||||
mk_msg("old_0", "conv", 0, "user", "旧0"),
|
||||
mk_msg("old_1", "conv", 1, "user", "旧1"),
|
||||
])
|
||||
.await
|
||||
.expect("insert old");
|
||||
|
||||
// replace 成全新内容(完全不同的 id,旧 id 应被删)
|
||||
let now = now_millis_str();
|
||||
let records = vec![
|
||||
AiMessageRecord {
|
||||
id: "new_0".into(),
|
||||
conversation_id: "conv".into(),
|
||||
seq: 0,
|
||||
role: "user".into(),
|
||||
content: "新0".into(),
|
||||
parts: None,
|
||||
tool_call_id: None,
|
||||
tool_calls: None,
|
||||
model: Some("deepseek-chat".into()),
|
||||
status: "active".into(),
|
||||
reasoning_content: Some("思考".into()),
|
||||
timestamp: Some(1700000000000),
|
||||
created_at: now.clone(),
|
||||
},
|
||||
AiMessageRecord {
|
||||
id: "new_1".into(),
|
||||
conversation_id: "conv".into(),
|
||||
seq: 1,
|
||||
role: "assistant".into(),
|
||||
content: "新1".into(),
|
||||
parts: None,
|
||||
tool_call_id: None,
|
||||
tool_calls: Some(r#"[{"id":"c1","type":"function","function":{"name":"f","arguments":"{}"}}]"#.into()),
|
||||
model: None,
|
||||
status: "active".into(),
|
||||
reasoning_content: None,
|
||||
timestamp: None,
|
||||
created_at: now,
|
||||
},
|
||||
];
|
||||
repo.replace_conversation("conv", records).await.expect("replace");
|
||||
|
||||
let got = repo.list_by_conversation("conv").await.expect("list");
|
||||
assert_eq!(got.len(), 2, "旧 2 条应被删,新 2 条入");
|
||||
assert_eq!(got[0].id, "new_0");
|
||||
assert_eq!(got[0].content, "新0");
|
||||
assert_eq!(got[0].model.as_deref(), Some("deepseek-chat"));
|
||||
assert_eq!(got[0].reasoning_content.as_deref(), Some("思考"));
|
||||
assert_eq!(got[1].id, "new_1");
|
||||
assert!(got[1].tool_calls.is_some());
|
||||
|
||||
// 确认旧 id 已删
|
||||
let ids: Vec<&str> = got.iter().map(|r| r.id.as_str()).collect();
|
||||
assert!(!ids.contains(&"old_0") && !ids.contains(&"old_1"));
|
||||
}
|
||||
|
||||
/// replace_conversation 空列表 = 清空该对话消息(单事务 DELETE 不插)
|
||||
#[tokio::test]
|
||||
async fn replace_conversation_empty_clears() {
|
||||
let db = Database::open_in_memory().await.expect("open_in_memory");
|
||||
let repo = AiMessageRepo::new(&db);
|
||||
repo.insert_batch(vec![
|
||||
mk_msg("m0", "c", 0, "user", "0"),
|
||||
mk_msg("m1", "c", 1, "user", "1"),
|
||||
])
|
||||
.await
|
||||
.expect("insert");
|
||||
|
||||
repo.replace_conversation("c", vec![]).await.expect("replace empty");
|
||||
let got = repo.list_by_conversation("c").await.expect("list");
|
||||
assert!(got.is_empty(), "空 records 应清空对话");
|
||||
}
|
||||
|
||||
/// replace_conversation 对话隔离:两 conv 并发写不串
|
||||
#[tokio::test]
|
||||
async fn replace_conversation_isolation_between_convs() {
|
||||
let db = Database::open_in_memory().await.expect("open_in_memory");
|
||||
let repo = AiMessageRepo::new(&db);
|
||||
|
||||
// 预置 conv_a 与 conv_b
|
||||
repo.insert_batch(vec![mk_msg("a0", "conv_a", 0, "user", "a0")])
|
||||
.await
|
||||
.expect("insert a");
|
||||
repo.insert_batch(vec![mk_msg("b0", "conv_b", 0, "user", "b0")])
|
||||
.await
|
||||
.expect("insert b");
|
||||
|
||||
// 只 replace conv_a,conv_b 不应被影响
|
||||
let now = now_millis_str();
|
||||
repo.replace_conversation(
|
||||
"conv_a",
|
||||
vec![AiMessageRecord {
|
||||
id: "a_new".into(),
|
||||
conversation_id: "conv_a".into(),
|
||||
seq: 0,
|
||||
role: "user".into(),
|
||||
content: "a_new".into(),
|
||||
parts: None,
|
||||
tool_call_id: None,
|
||||
tool_calls: None,
|
||||
model: None,
|
||||
status: "active".into(),
|
||||
reasoning_content: None,
|
||||
timestamp: None,
|
||||
created_at: now,
|
||||
}],
|
||||
)
|
||||
.await
|
||||
.expect("replace a");
|
||||
|
||||
let got_a = repo.list_by_conversation("conv_a").await.expect("list a");
|
||||
assert_eq!(got_a.len(), 1);
|
||||
assert_eq!(got_a[0].id, "a_new", "conv_a 应被全量重写");
|
||||
|
||||
let got_b = repo.list_by_conversation("conv_b").await.expect("list b");
|
||||
assert_eq!(got_b.len(), 1, "conv_b 不应被影响");
|
||||
assert_eq!(got_b[0].id, "b0");
|
||||
assert_eq!(got_b[0].content, "b0");
|
||||
}
|
||||
|
||||
/// replace_conversation 幂等性:INSERT OR IGNORE 在同事务内 DELETE 后无残留,
|
||||
/// 重复 replace 同 id 不报错(DELETE 后表内该 conv 空,INSERT 必然成功)
|
||||
#[tokio::test]
|
||||
async fn replace_conversation_idempotent_rerun() {
|
||||
let db = Database::open_in_memory().await.expect("open_in_memory");
|
||||
let repo = AiMessageRepo::new(&db);
|
||||
let now = now_millis_str();
|
||||
let rec = || AiMessageRecord {
|
||||
id: "m0".into(),
|
||||
conversation_id: "c".into(),
|
||||
seq: 0,
|
||||
role: "user".into(),
|
||||
content: "0".into(),
|
||||
parts: None,
|
||||
tool_call_id: None,
|
||||
tool_calls: None,
|
||||
model: None,
|
||||
status: "active".into(),
|
||||
reasoning_content: None,
|
||||
timestamp: None,
|
||||
created_at: now.clone(),
|
||||
};
|
||||
repo.replace_conversation("c", vec![rec()]).await.expect("1st");
|
||||
repo.replace_conversation("c", vec![rec()]).await.expect("2nd");
|
||||
let got = repo.list_by_conversation("c").await.expect("list");
|
||||
assert_eq!(got.len(), 1, "重复 replace 不应叠加");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user