新增: 消息级溯源 P0 地基(ChatMessage.id + ai_messages 拆表 + 迁移 + Repo)

依据消息拆分存储设计 + 消息级溯源设计 P0(地基,P1 溯源/P2 切读待后续):
- df-ai-core ChatMessage 加 id 字段(Option<String> serde 向前兼容)+ new_message_id(AtomicU64+ts 并发安全)
- 6 构造器生成 id,老 JSON 无 id → None 兼容
- df-storage V21 一次原子迁移:建 ai_messages 表 + ai_tool_executions.message_id 列 + 全量数据迁移(分批+坏数据容错+COUNT 幂等)
- AiMessageRecord + AiMessageRepo(insert_batch/list_by_conversation/delete_range/update_status/update_content_by_tool_call_id)
- audit message_id 列补建(conversation_repo/settings 白名单)
- src-tauri title.rs/finalize.rs 字面量占位(P1 接真值)

自验: df-storage 45 passed + df-ai-core 28 passed + workspace EXIT 0
This commit is contained in:
2026-06-19 19:24:02 +08:00
parent 44d1c6a00c
commit e981c1492a
10 changed files with 953 additions and 16 deletions

View File

@@ -23,7 +23,8 @@ pub fn run(conn: &Connection) -> Result<()> {
// 迁移步骤链: 顺序执行,跳过已应用的版本(current_version < N 才跑)。
// 新增版本时,在此数组追加一项 (N, migrate_vN) 即可,无需改逻辑。
let steps: [(i32, fn(&Connection) -> Result<()>); 19] = [
// V20 预留给 F-260619-01(任务关联灵感),跳过;V21 = 消息拆分存储 + audit message_id。
let steps: [(i32, fn(&Connection) -> Result<()>); 20] = [
(1, migrate_v1),
(2, migrate_v2),
(3, migrate_v3),
@@ -43,6 +44,7 @@ pub fn run(conn: &Connection) -> Result<()> {
(17, migrate_v17),
(18, migrate_v18),
(19, migrate_v19),
(21, migrate_v21),
];
for (version, migrate_fn) in steps {
@@ -348,6 +350,172 @@ fn migrate_v19(conn: &Connection) -> Result<()> {
Ok(())
}
/// V21:消息拆分存储(ai_messages 表 + 全量迁移)+ ai_tool_executions.message_id 列
///
/// **一次原子迁移**(决策 V21 合并,不拆 V21a/V21b):
/// 1. 建表 ai_messages(IF NOT EXISTS 幂等,新库空表/老库均安全)
/// 2. 幂等补 ai_tool_executions.message_id 列(消息级溯源 audit)
/// 3. COUNT 探测 ai_messages 已有数据 → 跳过数据迁移(仅写版本号,防重复迁移)
/// 4. 遍历 ai_conversations.messages JSON → 逐条提取到 ai_messages(分批 commit)
///
/// 设计要点(详见消息拆分存储设计 §4.2):
/// - **幂等安全**:COUNT 探测 + INSERT OR IGNORE,中途崩溃重跑跳过已迁移数据
/// - **分批 commit**:每 50 对话一批,避免长事务持有 SQLite 写锁
/// - **迁移期 ID**:`msg_migrated_{conv_id}_{seq}` —— 天然唯一(UNIQUE 是 conv_id+seq)、零依赖
/// - **裸 JSON 提取**:用 `serde_json::Value` 而非 ChatMessage(df-storage 不依赖 df-ai-core)
/// - **坏数据跳过**:JSON 解析失败 → warn + continue,不中断迁移
/// - **status 归一化**:None/空 → "active",列语义清晰永不 NULL
/// - **created_at 语义**:有 timestamp 用消息自己的;没有 fallback 到对话 created_at
///
/// ⚠️ **迁移耦合点**:迁移函数硬编码 JSON 字段名(role/content/parts/tool_call_id/
/// tool_calls/model/status/reasoning_content/timestamp),与 ChatMessage serde 序列化字段
/// 一一对应。ChatMessage 改字段名必须同步更新此函数,否则老库迁移漏数据。
/// 同步标注已在 types.rs ChatMessage 定义处加注释。
fn migrate_v21(conn: &Connection) -> Result<()> {
// 1. 建 ai_messages 表(IF NOT EXISTS 幂等)
conn.execute_batch(V21_SQL)?;
// 2. 幂等补 ai_tool_executions.message_id 列(消息级溯源 audit,F-260619-04)
// 表存在性兜底:run() 正常流程下 V9 已先建该表,但测试/手动调用可能跳过 V9。
// 表不存在时跳过 ALTER(新库会由 V9_SQL 建表带 message_id 列;此处只补老库已有表)。
let tool_exec_table_exists: bool = conn
.query_row(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='ai_tool_executions'",
[],
|_| Ok(()),
)
.is_ok();
if tool_exec_table_exists && !column_exists(conn, "ai_tool_executions", "message_id") {
conn.execute(
"ALTER TABLE ai_tool_executions ADD COLUMN message_id TEXT",
[],
)?;
tracing::info!("v21: 补建 ai_tool_executions.message_id 列(消息级溯源 audit)");
}
// 3. COUNT 探测:ai_messages 已有数据 → 跳过迁移只写版本号(幂等)
// INSERT OR IGNORE 防崩溃重跑(schema_version PK 冲突):run() 正常流程只调
// 一次 migrate_v21(current_version<21),但崩溃重跑/手动重调时 version=21
// 可能已存在,IGNORE 保证幂等不报错。
let existing: i64 = conn.query_row(
"SELECT COUNT(*) FROM ai_messages", [], |row| row.get(0),
)?;
if existing > 0 {
tracing::info!("v21: ai_messages 已有 {} 条,跳过数据迁移", existing);
conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [21])?;
return Ok(());
}
// 4. 遍历 ai_conversations,反序列化 messages JSON → 逐条写入 ai_messages
let mut stmt = conn.prepare(
"SELECT id, messages, created_at FROM ai_conversations",
)?;
let rows = stmt.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})?;
let all_rows: Vec<(String, String, String)> = rows.collect::<std::result::Result<Vec<_>, _>>()?;
// 5. 分批 commit(每 50 个对话一批,避免长事务持有写锁)
const BATCH_SIZE: usize = 50;
let mut migrated_count: usize = 0;
for (batch_idx, batch) in all_rows.chunks(BATCH_SIZE).enumerate() {
let tx = conn.unchecked_transaction()?;
for (conv_id, messages_json, conv_created_at) in batch {
// 6. 逐对话反序列化 messages JSON → Vec<serde_json::Value>
// (用裸 JSON 而非 ChatMessage,因 df-storage 不依赖 df-ai-core)
let messages: Vec<serde_json::Value> = match serde_json::from_str(messages_json) {
Ok(v) => v,
Err(e) => {
tracing::warn!("v21: 对话 {} messages JSON 解析失败,跳过: {}", conv_id, e);
continue; // 坏数据跳过,不中断迁移
}
};
for (seq, msg) in messages.iter().enumerate() {
// 7. 逐条消息提取字段 → INSERT INTO ai_messages
// 字段名硬编码("role"/"content" 等)——ChatMessage 改名会漏数据!
// 迁移期 ID 天然唯一(UNIQUE 是 conv_id+seq),人类可读,零依赖
let id = format!("msg_migrated_{}_{}", conv_id, seq);
let role = msg.get("role").and_then(|v| v.as_str()).unwrap_or("user");
let content = msg.get("content").and_then(|v| v.as_str()).unwrap_or("");
let parts = msg.get("parts")
.filter(|v| !v.is_null())
.map(|v| v.to_string());
let tool_call_id = msg.get("tool_call_id")
.and_then(|v| v.as_str())
.map(String::from);
let tool_calls = msg.get("tool_calls")
.filter(|v| !v.is_null())
.map(|v| v.to_string());
let model = msg.get("model")
.and_then(|v| v.as_str())
.map(String::from);
// status 归一化:None/空 → "active"(列语义清晰,永不 NULL)
let status = msg.get("status")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.unwrap_or("active");
let reasoning_content = msg.get("reasoning_content")
.and_then(|v| v.as_str())
.map(String::from);
// created_at:有 timestamp 用消息自己的,没有 fallback 到对话创建时间
let timestamp = msg.get("timestamp").and_then(|v| v.as_i64());
let created_at = timestamp
.map(|ts| ts.to_string())
.unwrap_or_else(|| conv_created_at.clone());
tx.execute(
"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)",
rusqlite::params![
id, conv_id, seq as i64, role, content, parts,
tool_call_id, tool_calls, model, status,
reasoning_content, timestamp, created_at
],
)?;
migrated_count += 1;
}
}
tx.commit()?;
tracing::info!("v21: 批次 {} 完成({} 对话)", batch_idx, batch.len());
}
conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [21])?;
tracing::info!("迁移 v21 完成,共迁移 {} 条消息", migrated_count);
Ok(())
}
/// V21 建表 SQL — 消息拆分存储 ai_messages 表
///
/// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。
/// 改动须两边同步。
const V21_SQL: &str = "
CREATE TABLE IF NOT EXISTS ai_messages (
id TEXT PRIMARY KEY,
conversation_id TEXT NOT NULL,
seq INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
parts TEXT,
tool_call_id TEXT,
tool_calls TEXT,
model TEXT,
status TEXT NOT NULL DEFAULT 'active',
reasoning_content TEXT,
timestamp INTEGER,
created_at TEXT NOT NULL,
UNIQUE(conversation_id, seq)
);
CREATE INDEX IF NOT EXISTS idx_ai_messages_conv ON ai_messages(conversation_id, seq);
";
/// V1 建表 SQL
const V1_SQL: &str = "
-- 想法表
@@ -541,6 +709,7 @@ CREATE TABLE IF NOT EXISTS ai_providers (
CREATE TABLE IF NOT EXISTS ai_tool_executions (
id TEXT PRIMARY KEY,
conversation_id TEXT,
message_id TEXT,
tool_call_id TEXT NOT NULL,
tool_name TEXT NOT NULL,
arguments TEXT NOT NULL,
@@ -551,6 +720,28 @@ CREATE TABLE IF NOT EXISTS ai_tool_executions (
executed_at TEXT,
decided_by TEXT
);
-- F-260619-03 消息拆分存储:每条 ChatMessage 一行的独立表。
-- 与 V21 迁移建表 SQL 镜像(V21 用于老库 ALTER,此处给新库直接建最终态)。
-- 改动须两边同步(V21_SQL 见下方)。
CREATE TABLE IF NOT EXISTS ai_messages (
id TEXT PRIMARY KEY,
conversation_id TEXT NOT NULL,
seq INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
parts TEXT,
tool_call_id TEXT,
tool_calls TEXT,
model TEXT,
status TEXT NOT NULL DEFAULT 'active',
reasoning_content TEXT,
timestamp INTEGER,
created_at TEXT NOT NULL,
UNIQUE(conversation_id, seq)
);
CREATE INDEX IF NOT EXISTS idx_ai_messages_conv ON ai_messages(conversation_id, seq);
";
/// V10 建表 SQL — 知识生命线事件表
@@ -584,3 +775,226 @@ CREATE TABLE IF NOT EXISTS app_settings (
updated_at TEXT NOT NULL
);
";
// ============================================================
// 单元测试 — V21 迁移幂等安全(新库/老库/坏数据三态,F-260619-03)
// ============================================================
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::Connection;
/// 构造最小老库 schema:ai_conversations 表(含 messages JSON 列)+ schema_version 表。
/// 不跑 V1-V19(测试聚焦 V21 单步行为),手动建最小依赖表。
fn setup_legacy_db() -> Connection {
let conn = Connection::open_in_memory().expect("open in-memory db");
conn.execute_batch(
"CREATE TABLE schema_version (version INTEGER PRIMARY KEY);
CREATE TABLE ai_conversations (
id TEXT PRIMARY KEY,
title TEXT,
messages TEXT NOT NULL DEFAULT '[]',
provider_id TEXT,
model TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);",
)
.expect("create legacy tables");
conn
}
/// 断言 ai_messages 表存在 + 列齐全
fn assert_ai_messages_schema(conn: &Connection) {
assert!(column_exists(conn, "ai_messages", "id"));
assert!(column_exists(conn, "ai_messages", "conversation_id"));
assert!(column_exists(conn, "ai_messages", "seq"));
assert!(column_exists(conn, "ai_messages", "role"));
assert!(column_exists(conn, "ai_messages", "content"));
assert!(column_exists(conn, "ai_messages", "status"));
assert!(column_exists(conn, "ai_messages", "created_at"));
}
/// 新库空跑:无 ai_conversations 数据,迁移应建表 + 写版本号 + 不崩 + ai_messages 空
#[test]
fn v21_new_db_empty_runs_clean() {
let conn = setup_legacy_db();
migrate_v21(&conn).expect("v21 应在新库空跑成功");
assert_ai_messages_schema(&conn);
// ai_tool_executions.message_id 列已补建
// 注:setup 未建 ai_tool_executions 表,column_exists 对不存在表返回 false。
// 此处验证迁移不因表不存在而崩(函数内 ALTER 被 column_exists 短路)。
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 0, "新库空跑 ai_messages 应为空");
let v: i64 = conn
.query_row("SELECT MAX(version) FROM schema_version", [], |r| r.get(0))
.unwrap();
assert_eq!(v, 21, "应写入版本号 21");
}
/// 老库有数据:正确迁移 messages JSON → ai_messages,字段全提取
#[test]
fn v21_legacy_db_migrates_messages() {
let conn = setup_legacy_db();
// 插入一条对话,messages 含 3 条消息(覆盖 user/assistant/tool + 各字段)
let messages_json = serde_json::json!([
{"role": "user", "content": "你好", "timestamp": 1718800000000i64},
{"role": "assistant", "content": "你好,有什么可以帮你?", "model": "glm-4", "reasoning_content": "思考中"},
{"role": "tool", "content": "工具结果", "tool_call_id": "call_abc", "tool_calls": [{"id": "call_abc"}]}
]).to_string();
conn.execute(
"INSERT INTO ai_conversations (id, title, messages, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5)",
rusqlite::params!["conv_1", "测试", messages_json, "1718800000000", "1718800000000"],
)
.unwrap();
migrate_v21(&conn).expect("v21 应成功迁移");
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 3, "应迁移 3 条消息");
// 校验 seq 递增 + 字段提取
let mut stmt = conn
.prepare("SELECT seq, role, content, model, tool_call_id, status, created_at FROM ai_messages WHERE conversation_id = 'conv_1' ORDER BY seq")
.unwrap();
let rows: Vec<(i64, String, String, Option<String>, Option<String>, String, String)> = stmt
.query_map([], |r| {
Ok((
r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?,
))
})
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(rows.len(), 3);
assert_eq!(rows[0].0, 0); // seq
assert_eq!(rows[0].1, "user");
assert_eq!(rows[0].2, "你好");
assert_eq!(rows[0].5, "active", "无 status → 归一化为 active");
assert_eq!(rows[0].6, "1718800000000", "有 timestamp → created_at 用它");
assert_eq!(rows[1].0, 1);
assert_eq!(rows[1].1, "assistant");
assert_eq!(rows[1].3.as_deref(), Some("glm-4"));
assert_eq!(rows[1].6, "1718800000000", "assistant 无 timestamp → fallback conv created_at");
assert_eq!(rows[2].0, 2);
assert_eq!(rows[2].1, "tool");
assert_eq!(rows[2].4.as_deref(), Some("call_abc"));
}
/// 坏数据:messages JSON 解析失败 → 该对话跳过,不中断整体迁移
#[test]
fn v21_bad_json_skipped_not_crash() {
let conn = setup_legacy_db();
// 坏数据对话
conn.execute(
"INSERT INTO ai_conversations (id, messages, created_at, updated_at) VALUES ('bad', '{not valid json', '0', '0')",
[],
)
.unwrap();
// 正常对话
let good = serde_json::json!([{"role": "user", "content": ""}]).to_string();
conn.execute(
"INSERT INTO ai_conversations (id, messages, created_at, updated_at) VALUES ('good', ?1, '0', '0')",
rusqlite::params![good],
)
.unwrap();
migrate_v21(&conn).expect("坏数据不应中断迁移");
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 1, "仅正常对话的 1 条被迁移");
// 坏数据对话在 ai_messages 无记录
let bad_count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM ai_messages WHERE conversation_id = 'bad'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(bad_count, 0);
}
/// 幂等重跑:第二次 migrate_v21 不重复迁移(COUNT 探测跳过)
#[test]
fn v21_idempotent_rerun() {
let conn = setup_legacy_db();
let msgs = serde_json::json!([{"role": "user", "content": "hi"}]).to_string();
conn.execute(
"INSERT INTO ai_conversations (id, messages, created_at, updated_at) VALUES ('c', ?1, '0', '0')",
rusqlite::params![msgs],
)
.unwrap();
migrate_v21(&conn).expect("首次迁移");
let count_after_first: i64 = conn
.query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0))
.unwrap();
assert_eq!(count_after_first, 1);
// 第二次跑:COUNT 探测 > 0 → 跳过数据迁移,不重复
migrate_v21(&conn).expect("二次迁移应幂等成功");
let count_after_second: i64 = conn
.query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0))
.unwrap();
assert_eq!(count_after_second, 1, "重跑不应重复插入");
// 版本号不重复写(schema_version version 是 PK,migrate_v21 用 INSERT OR IGNORE
// 防崩溃重跑 PK 冲突)
let v_count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM schema_version WHERE version = 21",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(v_count, 1, "版本号 21 应只写一次");
}
/// ai_tool_executions.message_id 列补建(老库已有表无该列)
#[test]
fn v21_adds_message_id_column_to_tool_executions() {
let conn = setup_legacy_db();
// 模拟老库已有 ai_tool_executions 表(V9 建的旧形态,无 message_id)
conn.execute_batch(
"CREATE TABLE ai_tool_executions (
id TEXT PRIMARY KEY,
conversation_id TEXT,
tool_call_id TEXT NOT NULL,
tool_name TEXT NOT NULL,
arguments TEXT NOT NULL,
result TEXT,
status TEXT NOT NULL DEFAULT 'pending',
risk_level TEXT NOT NULL DEFAULT 'medium',
requested_at TEXT NOT NULL,
executed_at TEXT,
decided_by TEXT
);",
)
.unwrap();
assert!(
!column_exists(&conn, "ai_tool_executions", "message_id"),
"迁移前应无 message_id 列"
);
migrate_v21(&conn).expect("v21 应补建 message_id 列");
assert!(
column_exists(&conn, "ai_tool_executions", "message_id"),
"迁移后应有 message_id 列"
);
}
}