lix/mnote
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(rag): harden post-LightRAG runtime

Browse Source 
Retire legacy OCR/media/evidence fallbacks, add local-folder event bus and Page Aggregate guards, and archive completed design checklists.

Validation: cargo test -p mnote-web -- --test-threads=1; cargo test --workspace -- --test-threads=1; git diff --check; codegraph sync .; codegraph_status.
This commit is contained in:
lix-2026
2026-06-07 10:35:21 +08:00
parent 22a92edcda
commit 9551d4c1dc
59 changed files with 4053 additions and 5249 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/crates/mnote-web/src/routes/knowledge_rag.rs
+274 -27
View File
@@ -24,6 +24,7 @@ const DEFAULT_LIGHTRAG_ENDPOINT: &str = "http://127.0.0.1:9621";
const DEFAULT_LIGHTRAG_INPUT_DIR: &str = "/mnt/Data1T/Mnote_data/lightrag/inputs";
const DEFAULT_LIGHTRAG_WORKING_DIR: &str = "/mnt/Data1T/Mnote_data/lightrag/rag_storage";
const MAX_INGEST_SOURCES_PER_REQUEST: usize = 200;
const SOURCE_SCOPE_MODE_POST_FILTER: &str = "post_filter_mapped_references";
const KNOWLEDGE_RAG_SOURCE_EXTENSIONS: &[&str] = &[
"md", "markdown", "txt", "pdf", "doc", "docx", "ppt", "pptx", "xls", "xlsx", "csv", "png",
"jpg", "jpeg", "webp", "gif", "bmp", "tif", "tiff",
@@ -154,6 +155,8 @@ pub(crate) fn knowledge_rag_source_statuses(
}
if provider_status == "failed" {
statuses.failed_paths.insert(path.to_string());
} else if provider_status == "delete_retry_required" {
statuses.failed_paths.insert(path.to_string());
} else if provider_status == "delete_submitted"
|| (entry.deleted_at_ms.is_some() && entry.light_rag_doc_id.is_some())
{
@@ -473,6 +476,8 @@ pub async fn query_rag(
"schema": "mnote.knowledge_rag.query_result.v1",
"provider": "lightrag",
"sourceScope": source_scope,
"sourceScopeMode": SOURCE_SCOPE_MODE_POST_FILTER,
"rawScopeFiltered": false,
"raw": raw,
"references": references,
})))
@@ -735,7 +740,13 @@ pub async fn prune_registry(
}
fn knowledge_rag_registry_entry_prunable(entry: &KnowledgeRagSourceRegistryEntry) -> bool {
if entry.light_rag_status.as_deref() == Some("delete_submitted") {
if matches!(
entry.light_rag_status.as_deref(),
Some("delete_submitted" | "delete_retry_required")
) {
return false;
}
if entry.light_rag_doc_id.is_some() && (entry.deleted_at_ms.is_some() || entry.stale) {
return false;
}
entry.deleted_at_ms.is_some()
@@ -746,6 +757,23 @@ fn knowledge_rag_registry_entry_prunable(entry: &KnowledgeRagSourceRegistryEntry
)
}
fn knowledge_rag_provider_delete_confirmed(entry: &KnowledgeRagSourceRegistryEntry) -> bool {
entry.light_rag_doc_id.is_some()
&& (entry.deleted_at_ms.is_some()
|| entry.stale
|| matches!(
entry.light_rag_status.as_deref(),
Some("delete_submitted" | "delete_retry_required")
))
}
fn mark_registry_entry_delete_completed(entry: &mut KnowledgeRagSourceRegistryEntry, now: u128) {
entry.light_rag_doc_id = None;
entry.indexed_at_ms = None;
entry.light_rag_status = Some("delete_completed".into());
entry.updated_at_ms = now;
}
async fn sync_registry_with_documents(
root_path: &Path,
registry: &mut KnowledgeRagSourceRegistry,
@@ -758,22 +786,28 @@ async fn sync_registry_with_documents(
let by_file_path = lightrag_documents_by_file_path(&docs);
let now = now_ms();
let mut changed = false;
let mut retry_doc_ids = Vec::new();
for entry in &mut registry.entries {
if let Some(doc) = document_for_registry_entry(&by_file_path, entry) {
if let Some(id) = doc.get("id").and_then(Value::as_str) {
entry.light_rag_doc_id = Some(id.to_string());
}
if let Some(status) = doc.get("status").and_then(Value::as_str) {
entry.light_rag_status = Some(
if entry.deleted_at_ms.is_some() {
"delete_submitted"
} else {
status
}
.to_string(),
);
let delete_pending = matches!(
entry.light_rag_status.as_deref(),
Some("delete_submitted" | "delete_retry_required")
) || entry.deleted_at_ms.is_some();
if delete_pending {
if let Some(doc_id) = entry.light_rag_doc_id.clone() {
retry_doc_ids.push(doc_id);
}
if entry.light_rag_status.is_none() {
entry.light_rag_status = Some("delete_submitted".into());
}
} else if let Some(status) = doc.get("status").and_then(Value::as_str) {
entry.light_rag_status = Some(status.to_string());
}
if entry.deleted_at_ms.is_none()
if !entry.stale
&& entry.deleted_at_ms.is_none()
&& doc.get("status").and_then(Value::as_str) == Some("processed")
{
entry.indexed_at_ms.get_or_insert(now);
@@ -781,11 +815,8 @@ async fn sync_registry_with_documents(
}
entry.updated_at_ms = now;
changed = true;
} else if entry.deleted_at_ms.is_some() && entry.light_rag_doc_id.is_some() {
entry.light_rag_doc_id = None;
entry.indexed_at_ms = None;
entry.light_rag_status = Some("delete_completed".into());
entry.updated_at_ms = now;
} else if knowledge_rag_provider_delete_confirmed(entry) {
mark_registry_entry_delete_completed(entry, now);
changed = true;
} else if entry.deleted_at_ms.is_some()
&& entry.light_rag_doc_id.is_none()
@@ -801,9 +832,12 @@ async fn sync_registry_with_documents(
changed = true;
}
}
let stale_doc_ids = sync_registry_source_state(registry, now)?;
let mut stale_doc_ids = sync_registry_source_state(registry, now)?;
stale_doc_ids.extend(retry_doc_ids);
stale_doc_ids.sort();
stale_doc_ids.dedup();
if !stale_doc_ids.is_empty() {
let _ = lightrag_json(
let delete_result = lightrag_json(
reqwest::Method::DELETE,
"/documents/delete_document",
Some(json!({
@@ -815,6 +849,23 @@ async fn sync_registry_with_documents(
context,
)
.await;
for entry in &mut registry.entries {
if entry
.light_rag_doc_id
.as_deref()
.is_some_and(|doc_id| stale_doc_ids.iter().any(|item| item == doc_id))
{
entry.light_rag_status = Some(
if delete_result.is_err() {
"delete_retry_required"
} else {
"delete_submitted"
}
.into(),
);
entry.updated_at_ms = now;
}
}
changed = true;
}
if changed {
@@ -897,8 +948,8 @@ fn sync_registry_source_state(
if !source_path.exists() {
entry.stale = true;
entry.deleted_at_ms = Some(now);
entry.light_rag_doc_id = None;
entry.indexed_at_ms = None;
entry.light_rag_status = Some("delete_submitted".into());
entry.updated_at_ms = now;
stale_doc_ids.push(doc_id);
continue;
@@ -907,8 +958,8 @@ fn sync_registry_source_state(
if current_hash != entry.source_hash {
entry.stale = true;
entry.source_hash = current_hash;
entry.light_rag_doc_id = None;
entry.indexed_at_ms = None;
entry.light_rag_status = Some("delete_submitted".into());
entry.updated_at_ms = now;
stale_doc_ids.push(doc_id);
}
@@ -989,6 +1040,10 @@ fn mapped_references(
.get("deleted")
.and_then(Value::as_bool)
.unwrap_or(false)
&& !reference
.get("unmapped")
.and_then(Value::as_bool)
.unwrap_or(false)
})
.collect()
}
@@ -1127,6 +1182,7 @@ fn map_reference_plan(
"sourceId": entry.map(|entry| entry.source_id.clone()),
"sourcePath": source_path,
"sourceRootRelativePath": source_root_relative_path,
"unmapped": entry.is_none(),
"stale": entry.is_some_and(|entry| entry.stale),
"deleted": entry.is_some_and(|entry| entry.deleted_at_ms.is_some()),
"locatorDegraded": locator_degraded,
@@ -1346,15 +1402,11 @@ fn find_lightrag_sidecar_block(
}
let path = sidecar_blocks_path(entry)?;
let content = fs::read_to_string(path).ok()?;
let mut first_positioned_block = None;
for line in content.lines() {
let block = serde_json::from_str::<Value>(line).ok()?;
if block.get("positions").and_then(Value::as_array).is_none() {
continue;
}
if first_positioned_block.is_none() {
first_positioned_block = Some(block.clone());
}
let block_text = block
.get("content")
.and_then(Value::as_str)
@@ -1369,7 +1421,7 @@ fn find_lightrag_sidecar_block(
return Some(block);
}
}
first_positioned_block
None
}
fn normalize_text_for_match(value: &str) -> String {
@@ -2234,17 +2286,85 @@ mod tests {
let stale_doc_ids = sync_registry_source_state(&mut registry, 42).expect("sync");
assert_eq!(stale_doc_ids, vec!["doc-changed", "doc-missing"]);
assert_eq!(registry.entries[0].deleted_at_ms, Some(42));
assert_eq!(registry.entries[0].light_rag_doc_id, None);
assert_eq!(
registry.entries[0].light_rag_doc_id.as_deref(),
Some("doc-missing")
);
assert_eq!(
registry.entries[0].light_rag_status.as_deref(),
Some("delete_submitted")
);
assert_eq!(registry.entries[0].indexed_at_ms, None);
assert!(registry.entries[0].stale);
assert_eq!(registry.entries[1].deleted_at_ms, None);
assert_eq!(registry.entries[1].light_rag_doc_id, None);
assert_eq!(
registry.entries[1].light_rag_doc_id.as_deref(),
Some("doc-changed")
);
assert_eq!(
registry.entries[1].light_rag_status.as_deref(),
Some("delete_submitted")
);
assert_eq!(registry.entries[1].indexed_at_ms, None);
assert!(registry.entries[1].stale);
assert_ne!(registry.entries[1].source_hash, "mnote-fnv64:old");
let _ = fs::remove_dir_all(root);
}
#[test]
fn provider_delete_confirmation_clears_doc_id_for_deleted_or_stale_entries() {
let root = temp_root("mnote-knowledge-rag-delete-confirmed");
let mut deleted = test_registry_entry(
&root,
"deleted.pdf",
Some("doc-deleted"),
Some("delete_submitted"),
Some(2),
Some(3),
true,
);
let mut changed = test_registry_entry(
&root,
"changed.pdf",
Some("doc-changed"),
Some("delete_submitted"),
Some(2),
None,
true,
);
let active = test_registry_entry(
&root,
"active.pdf",
Some("doc-active"),
Some("processed"),
Some(2),
None,
false,
);
assert!(knowledge_rag_provider_delete_confirmed(&deleted));
mark_registry_entry_delete_completed(&mut deleted, 42);
assert_eq!(deleted.light_rag_doc_id, None);
assert_eq!(deleted.indexed_at_ms, None);
assert_eq!(
deleted.light_rag_status.as_deref(),
Some("delete_completed")
);
assert!(knowledge_rag_provider_delete_confirmed(&changed));
mark_registry_entry_delete_completed(&mut changed, 43);
assert_eq!(changed.light_rag_doc_id, None);
assert_eq!(
changed.light_rag_status.as_deref(),
Some("delete_completed")
);
assert!(!knowledge_rag_provider_delete_confirmed(&active));
assert_eq!(active.light_rag_doc_id.as_deref(), Some("doc-active"));
let _ = fs::remove_dir_all(root);
}
#[test]
fn source_statuses_distinguish_indexed_processing_failed_and_deleted() {
let root = temp_root("mnote-knowledge-rag-source-statuses");
@@ -2372,9 +2492,29 @@ mod tests {
);
let failed =
test_registry_entry(&root, "failed.pdf", None, Some("failed"), None, None, false);
let retry = test_registry_entry(
&root,
"retry.pdf",
Some("doc-retry"),
Some("delete_retry_required"),
Some(2),
Some(3),
true,
);
let stale_with_doc = test_registry_entry(
&root,
"stale.pdf",
Some("doc-stale"),
Some("processed"),
Some(2),
None,
true,
);
assert!(!knowledge_rag_registry_entry_prunable(&active));
assert!(!knowledge_rag_registry_entry_prunable(&deleting));
assert!(!knowledge_rag_registry_entry_prunable(&retry));
assert!(!knowledge_rag_registry_entry_prunable(&stale_with_doc));
assert!(knowledge_rag_registry_entry_prunable(&removed));
assert!(knowledge_rag_registry_entry_prunable(&failed));
@@ -2466,6 +2606,42 @@ mod tests {
);
}
#[test]
fn mapped_references_filters_unmapped_provider_references() {
let registry = KnowledgeRagSourceRegistry {
schema: REGISTRY_SCHEMA.to_string(),
workspace_id: "ws".into(),
root_uri: "file:///tmp/root".into(),
updated_at_ms: 1,
entries: vec![],
};
let raw = json!({
"data": {
"references": [{"reference_id":"1","file_path":"orphan.pdf"}],
"chunks": [{
"reference_id":"1",
"chunk_id":"orphan-chunk",
"file_path":"orphan.pdf",
"content":"orphan provider chunk"
}]
}
});
let mapped = mapped_references(&raw, &registry, "file:///tmp/root", Path::new("/tmp/root"));
assert!(
mapped.is_empty(),
"unmapped provider references must not become MNote citations"
);
let plan = map_reference_plan(
&json!({"file_path":"orphan.pdf","chunk_id":"orphan-chunk"}),
&registry,
"file:///tmp/root",
Path::new("/tmp/root"),
);
assert_eq!(plan["unmapped"], true);
assert_eq!(plan["locatorDegraded"], true);
}
#[test]
fn query_ranking_prefers_exact_source_and_quote_match() {
let mut references = vec![
@@ -2597,6 +2773,77 @@ mod tests {
let _ = fs::remove_dir_all(root);
}
#[test]
fn locator_degrades_when_sidecar_quote_does_not_match() {
let _guard = env_lock().lock().expect("env lock");
let root = temp_root("mnote-knowledge-rag-sidecar-no-match");
fs::create_dir_all(root.join("docs")).expect("docs");
fs::write(root.join("docs").join("Host.md"), "# Host\n").expect("host");
fs::write(root.join("docs").join("scan.pdf"), b"pdf").expect("pdf");
let input_dir = root.join("inputs");
let parsed_dir = input_dir
.join("__parsed__")
.join("mnote-hash-scan.pdf.parsed");
fs::create_dir_all(&parsed_dir).expect("parsed dir");
fs::write(
parsed_dir.join("mnote-hash-scan.blocks.jsonl"),
[
r#"{"type":"meta","blocks":1}"#,
r##"{"type":"content","blockid":"block1","content":"This block is not the returned quote.","positions":[{"type":"bbox","anchor":"9","range":[1.0,2.0,3.0,4.0]}]}"##,
]
.join("\n"),
)
.expect("blocks");
std::env::set_var("MNOTE_LIGHTRAG_INPUT_DIR", &input_dir);
let registry = KnowledgeRagSourceRegistry {
schema: REGISTRY_SCHEMA.to_string(),
workspace_id: "ws".into(),
root_uri: "file:///tmp/root".into(),
updated_at_ms: 1,
entries: vec![KnowledgeRagSourceRegistryEntry {
source_id: "src1".into(),
workspace_id: "ws".into(),
root_uri: "file:///tmp/root".into(),
source_path: root.join("docs").join("scan.pdf").display().to_string(),
source_root_relative_path: "docs/scan.pdf".into(),
source_hash: "mnote-fnv64:1".into(),
light_rag_doc_id: Some("doc1".into()),
light_rag_status: Some("processed".into()),
light_rag_file_path: "mnote-hash-scan.pdf".into(),
symlink_path: input_dir.join("mnote-hash-scan.pdf").display().to_string(),
parser_hint: None,
indexed_at_ms: Some(2),
deleted_at_ms: None,
stale: false,
updated_at_ms: 2,
}],
};
let mapped = map_reference_plan(
&json!({
"file_path": "mnote-hash-scan.pdf",
"chunk_id": "doc1-chunk-000",
"chunks": [{"chunk_id": "doc1-chunk-000", "content": "A different quote should not get page or bbox."}]
}),
&registry,
"file:///tmp/root",
&root,
);
assert!(mapped["locator"].is_null());
assert_eq!(mapped["locatorDegraded"], true);
assert!(mapped["citationUrl"]
.as_str()
.is_some_and(|url| url.contains("resourceTab=")));
assert!(mapped["citationMarkdown"]
.as_str()
.unwrap()
.contains("来源定位降级"));
std::env::remove_var("MNOTE_LIGHTRAG_INPUT_DIR");
let _ = fs::remove_dir_all(root);
}
#[test]
fn lightrag_paths_prefer_source_env_over_legacy_process_env() {
let _guard = env_lock().lock().expect("env lock");