feat(mcp): notes 档案分片双写 + backfill 自愈（可拆除性修复） #5
@ -28,7 +28,14 @@ fn validate_registered_path(raw: &str, field: &str) -> Result<PathBuf, String> {
|
||||
/// 解析项目根目录路径（WSL 项目自动转为 Windows UNC 路径）
|
||||
fn project_root(project_id: &str) -> Result<PathBuf, String> {
|
||||
let conn = db::pool().get().map_err(|e| e.to_string())?;
|
||||
project_root_with_conn(&conn, project_id)
|
||||
}
|
||||
|
||||
/// project_root 的依赖注入版：连接由调用方传入（业务层复用 + in-memory 测试）
|
||||
pub(crate) fn project_root_with_conn(
|
||||
conn: &rusqlite::Connection,
|
||||
project_id: &str,
|
||||
) -> Result<PathBuf, String> {
|
||||
let row: rusqlite::Result<(Option<String>, Option<String>, Option<String>)> = conn.query_row(
|
||||
"SELECT win_path, wsl_path, platform FROM project_workspaces WHERE id = ?1",
|
||||
[project_id],
|
||||
|
||||
@ -80,6 +80,11 @@ pub fn tools_list_result() -> Value {
|
||||
"required": ["project_id", "content"]
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "backfill_notes_archive",
|
||||
"description": "将 project_notes 全量重建到各项目 .blueprint/notes-archive.jsonl 分片档案（可拆除性修复：笔记是全系统唯一不在 git 的原创资产）。幂等，DB 为事实源整文件重写；append_project_note 双写失败后的自愈入口。无参数。",
|
||||
"inputSchema": { "type": "object", "properties": {}, "required": [] }
|
||||
},
|
||||
{
|
||||
"name": "get_blueprint_status",
|
||||
"description": "检测项目蓝图治理状态：CONVENTIONS 规则版本是否最新、蓝图内容是否可能落后于代码（synced/rules_outdated/content_stale/none）。等价于炼境 UI 项目卡片上的蓝图状态徽章。",
|
||||
@ -386,6 +391,10 @@ pub async fn tools_call(params: Option<&Value>) -> Value {
|
||||
let content = args.get("content").and_then(|v| v.as_str()).unwrap_or("");
|
||||
append_project_note(pid, content)
|
||||
}
|
||||
"backfill_notes_archive" => {
|
||||
let conn = db::pool().get().map_err(|e| e.to_string())?;
|
||||
backfill_notes_archive(&conn)
|
||||
}
|
||||
"get_project_snapshots" => {
|
||||
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
|
||||
let days = args.get("days").and_then(|v| v.as_u64()).unwrap_or(30) as u32;
|
||||
@ -1035,7 +1044,121 @@ fn append_project_note(project_id: &str, content: &str) -> Result<String, String
|
||||
rusqlite::params![project_id, content],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok("笔记已保存 ✅".to_string())
|
||||
// 双写到项目分片档案（可拆除性）；失败不回滚笔记——DB 是事实源，backfill 可自愈
|
||||
let note_id = conn.last_insert_rowid();
|
||||
if let Err(e) = archive_note_append(&conn, note_id) {
|
||||
db::log_event(
|
||||
"mcp_tools",
|
||||
"warn",
|
||||
&format!("notes 档案双写失败（DB 已保存，backfill_notes_archive 可自愈）: {e}"),
|
||||
None,
|
||||
);
|
||||
return Ok("笔记已保存 ✅（项目档案双写失败已记日志，可调 backfill_notes_archive 自愈）".to_string());
|
||||
}
|
||||
Ok("笔记已保存 ✅（已同步项目 .blueprint/notes-archive.jsonl）".to_string())
|
||||
}
|
||||
|
||||
// ── notes 档案分片（可拆除性修复）─────────────────────────────────────────────
|
||||
// project_notes 是全系统唯一不在 git 的原创资产；双写到各项目
|
||||
// .blueprint/notes-archive.jsonl（JSONL 机器可重放重建 DB，即「重生配方」）。
|
||||
// 归宿按项目分片而非炼境仓库集中存放：路径解析可靠（复用登记路径防护）、
|
||||
// 可拆除性按项目成立、子项目会话（含 DSH）可直读自家档案。
|
||||
|
||||
static NOTES_ARCHIVE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
|
||||
|
||||
/// 单条笔记 → JSONL 行（含 project_id，档案自含，脱离文件位置也可重建）
|
||||
fn note_jsonl_line(id: i64, project_id: &str, source: &str, content: &str, created_at: &str) -> String {
|
||||
json!({
|
||||
"id": id,
|
||||
"project_id": project_id,
|
||||
"source": source,
|
||||
"content": content,
|
||||
"created_at": created_at,
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
fn notes_archive_path(
|
||||
conn: &rusqlite::Connection,
|
||||
project_id: &str,
|
||||
) -> Result<std::path::PathBuf, String> {
|
||||
let root = crate::mcp_inject::project_root_with_conn(conn, project_id)?;
|
||||
Ok(root.join(".blueprint").join("notes-archive.jsonl"))
|
||||
}
|
||||
|
||||
/// 追加单条到项目分片档案。路径无效（死登记/空路径）返回 Err，由调用方降级处理
|
||||
fn archive_note_append(conn: &rusqlite::Connection, note_id: i64) -> Result<(), String> {
|
||||
let (pid, source, content, created_at): (String, String, String, String) = conn
|
||||
.query_row(
|
||||
"SELECT project_id, source, content, created_at FROM project_notes WHERE id = ?1",
|
||||
[note_id],
|
||||
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let path = notes_archive_path(conn, &pid)?;
|
||||
let _guard = NOTES_ARCHIVE_LOCK.lock().map_err(|e| e.to_string())?;
|
||||
if let Some(dir) = path.parent() {
|
||||
std::fs::create_dir_all(dir).map_err(|e| e.to_string())?;
|
||||
}
|
||||
use std::io::Write;
|
||||
let mut f = std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&path)
|
||||
.map_err(|e| e.to_string())?;
|
||||
writeln!(f, "{}", note_jsonl_line(note_id, &pid, &source, &content, &created_at))
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 全量重建所有项目的分片档案（幂等，DB 为事实源整文件重写）
|
||||
fn backfill_notes_archive(conn: &rusqlite::Connection) -> Result<String, String> {
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT id, project_id, source, content, created_at
|
||||
FROM project_notes ORDER BY project_id, id",
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let rows: Vec<(i64, String, String, String, String)> = stmt
|
||||
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)))
|
||||
.map_err(|e| e.to_string())?
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
|
||||
let mut by_project: std::collections::BTreeMap<String, Vec<String>> = std::collections::BTreeMap::new();
|
||||
for (id, pid, source, content, created_at) in &rows {
|
||||
by_project
|
||||
.entry(pid.clone())
|
||||
.or_default()
|
||||
.push(note_jsonl_line(*id, pid, source, content, created_at));
|
||||
}
|
||||
|
||||
let _guard = NOTES_ARCHIVE_LOCK.lock().map_err(|e| e.to_string())?;
|
||||
let mut report = String::from("## notes 档案回填（全量重建）\n\n");
|
||||
let (mut archived, mut skipped) = (0usize, 0usize);
|
||||
for (pid, lines) in &by_project {
|
||||
match notes_archive_path(conn, pid) {
|
||||
Ok(path) => {
|
||||
if let Some(dir) = path.parent() {
|
||||
std::fs::create_dir_all(dir).map_err(|e| e.to_string())?;
|
||||
}
|
||||
std::fs::write(&path, lines.join("\n") + "\n").map_err(|e| e.to_string())?;
|
||||
report.push_str(&format!("- ✅ {}：{} 条 → {}\n", pid, lines.len(), path.display()));
|
||||
archived += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
report.push_str(&format!("- ⚠️ {}：跳过（{}），{} 条仅存 DB\n", pid, e, lines.len()));
|
||||
skipped += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
report.push_str(&format!(
|
||||
"\n共 {} 条笔记；{} 个项目已归档，{} 个跳过（跳过项待修复登记路径后重跑即可）。",
|
||||
rows.len(),
|
||||
archived,
|
||||
skipped
|
||||
));
|
||||
Ok(report)
|
||||
}
|
||||
|
||||
fn get_project_snapshots(project_id: &str, days: u32) -> Result<String, String> {
|
||||
@ -1303,6 +1426,135 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// notes 档案测试用最小 schema（in-memory 或文件库共用）
|
||||
fn setup_notes_schema(conn: &rusqlite::Connection) {
|
||||
conn.execute_batch(
|
||||
"CREATE TABLE IF NOT EXISTS project_notes (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
project_id TEXT NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT 'mcp',
|
||||
content TEXT NOT NULL,
|
||||
created_at TEXT DEFAULT (datetime('now', 'localtime'))
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS project_workspaces (
|
||||
id TEXT PRIMARY KEY, profile_id TEXT,
|
||||
win_path TEXT, wsl_path TEXT, platform TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT);",
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn temp_base(tag: &str) -> std::path::PathBuf {
|
||||
let base = std::env::temp_dir().join(format!("lianjing_notes_{tag}_{}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&base);
|
||||
std::fs::create_dir_all(&base).unwrap();
|
||||
base
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn notes_archive_dual_write_line_matches_db() {
|
||||
let conn = rusqlite::Connection::open_in_memory().unwrap();
|
||||
setup_notes_schema(&conn);
|
||||
let base = temp_base("dual");
|
||||
conn.execute(
|
||||
"INSERT INTO project_workspaces (id, win_path) VALUES ('p1', ?1)",
|
||||
[base.to_str().unwrap()],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO project_notes (project_id, content) VALUES ('p1', '含换行\n与中文的内容')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
let note_id = conn.last_insert_rowid();
|
||||
archive_note_append(&conn, note_id).expect("双写应成功");
|
||||
|
||||
let file = base.join(".blueprint").join("notes-archive.jsonl");
|
||||
let text = std::fs::read_to_string(&file).unwrap();
|
||||
let lines: Vec<&str> = text.lines().collect();
|
||||
assert_eq!(lines.len(), 1, "应恰好一行");
|
||||
let v: Value = serde_json::from_str(lines[0]).expect("尾行应为合法 JSON");
|
||||
assert_eq!(v["content"], "含换行\n与中文的内容");
|
||||
assert_eq!(v["project_id"], "p1");
|
||||
assert_eq!(v["id"], note_id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn notes_backfill_rebuilds_and_skips_dead_registration() {
|
||||
let conn = rusqlite::Connection::open_in_memory().unwrap();
|
||||
setup_notes_schema(&conn);
|
||||
let base = temp_base("backfill");
|
||||
conn.execute(
|
||||
"INSERT INTO project_workspaces (id, win_path) VALUES ('good', ?1)",
|
||||
[base.to_str().unwrap()],
|
||||
)
|
||||
.unwrap();
|
||||
// 死登记：空路径（2026-07-15 空路径污染同款场景，必须跳过而非写进 cwd）
|
||||
conn.execute("INSERT INTO project_workspaces (id, win_path) VALUES ('dead', '')", []).unwrap();
|
||||
for i in 0..3 {
|
||||
conn.execute(
|
||||
"INSERT INTO project_notes (project_id, content) VALUES ('good', ?1)",
|
||||
[format!("note-{i}")],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
conn.execute("INSERT INTO project_notes (project_id, content) VALUES ('dead', 'x')", []).unwrap();
|
||||
|
||||
let report = backfill_notes_archive(&conn).expect("回填应成功");
|
||||
let file = base.join(".blueprint").join("notes-archive.jsonl");
|
||||
let count = std::fs::read_to_string(&file).unwrap().lines().count();
|
||||
assert_eq!(count, 3, "good 项目 JSONL 行数应等于其表行数");
|
||||
assert!(report.contains("⚠️ dead"), "死登记应被跳过并报告: {report}");
|
||||
|
||||
// 幂等：重跑后行数不变（整文件重写而非追加）
|
||||
backfill_notes_archive(&conn).unwrap();
|
||||
let count2 = std::fs::read_to_string(&file).unwrap().lines().count();
|
||||
assert_eq!(count2, 3, "重跑不应翻倍");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn notes_archive_concurrent_appends_no_interleave() {
|
||||
let base = temp_base("concurrent");
|
||||
let db_path = base.join("test.db");
|
||||
{
|
||||
let conn = rusqlite::Connection::open(&db_path).unwrap();
|
||||
setup_notes_schema(&conn);
|
||||
conn.execute(
|
||||
"INSERT INTO project_workspaces (id, win_path) VALUES ('p1', ?1)",
|
||||
[base.to_str().unwrap()],
|
||||
)
|
||||
.unwrap();
|
||||
for i in 0..8 {
|
||||
conn.execute(
|
||||
"INSERT INTO project_notes (project_id, content) VALUES ('p1', ?1)",
|
||||
[format!("并发内容较长以放大交错概率-{i}-{}", "x".repeat(500))],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
let handles: Vec<_> = (1..=8)
|
||||
.map(|note_id| {
|
||||
let p = db_path.clone();
|
||||
std::thread::spawn(move || {
|
||||
let conn = rusqlite::Connection::open(&p).unwrap();
|
||||
archive_note_append(&conn, note_id).expect("并发追加应成功");
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
for h in handles {
|
||||
h.join().unwrap();
|
||||
}
|
||||
let file = base.join(".blueprint").join("notes-archive.jsonl");
|
||||
let text = std::fs::read_to_string(&file).unwrap();
|
||||
let lines: Vec<&str> = text.lines().collect();
|
||||
assert_eq!(lines.len(), 8, "8 线程应产出 8 行");
|
||||
for l in &lines {
|
||||
let v: Value = serde_json::from_str(l).expect("每行都应是完整 JSON（无交错）");
|
||||
assert_eq!(v["project_id"], "p1");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn apply_onboarding_pack_schema_requires_project_id() {
|
||||
let result = tools_list_result();
|
||||
|
||||
Loading…
Reference in New Issue
Block a user