From a470d3e39baee9fab4ffa556a867ff116754383e Mon Sep 17 00:00:00 2001 From: hobertrand-hub Date: Sat, 25 Jul 2026 02:47:09 +0200 Subject: [PATCH 1/8] fix(memory): unblock legacy chunk reembedding Co-authored-by: Medulla --- src/memory/chunks/embeddings.rs | 11 ++++- src/memory/chunks/store_embed_tests.rs | 68 ++++++++++++++++++++++++-- src/memory/store/content/read.rs | 22 +++++++-- src/memory/store/content/read_tests.rs | 10 ++++ 4 files changed, 101 insertions(+), 10 deletions(-) diff --git a/src/memory/chunks/embeddings.rs b/src/memory/chunks/embeddings.rs index 3ef55616..1aa7306a 100644 --- a/src/memory/chunks/embeddings.rs +++ b/src/memory/chunks/embeddings.rs @@ -77,6 +77,11 @@ fn upsert_chunk_embedding_conn( created_at = excluded.created_at", rusqlite::params![chunk_id, model_signature, bytes, dim, created_at], )?; + conn.execute( + "DELETE FROM mem_tree_chunk_reembed_skipped + WHERE chunk_id = ?1 AND model_signature = ?2", + rusqlite::params![chunk_id, model_signature], + )?; Ok(()) } @@ -412,13 +417,15 @@ pub fn has_uncovered_reembed_work( .map(|variant| variant as &dyn rusqlite::ToSql) .collect(); conn.query_row( - &format!( +&format!( "SELECT EXISTS( SELECT 1 FROM mem_tree_chunks c WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e WHERE e.chunk_id = c.id AND e.model_signature {sig_clause}) AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk - WHERE sk.chunk_id = c.id AND sk.model_signature {sig_clause})) + WHERE sk.chunk_id = c.id AND sk.model_signature {sig_clause} + AND sk.reason NOT LIKE 'body read failed: no content pointer or raw refs for chunk %' + AND sk.reason NOT LIKE 'body read failed: empty content pointer and no raw refs for chunk %')) OR EXISTS( SELECT 1 FROM mem_tree_summaries s WHERE s.deleted = 0 diff --git a/src/memory/chunks/store_embed_tests.rs b/src/memory/chunks/store_embed_tests.rs index 5207db97..b8b40889 100644 --- a/src/memory/chunks/store_embed_tests.rs +++ b/src/memory/chunks/store_embed_tests.rs @@ -19,9 +19,10 @@ use super::{ clear_summary_reembed_skipped, content_root, count_chunks, db_path_for, delete_chunks_by_owner, delete_chunks_by_source, extraction_coverage, get_chunk, get_chunk_embedding, get_chunk_embedding_for_signature, get_chunk_embeddings_for_signature_batch, get_chunks_batch, - is_source_ingested, list_chunks, mark_chunk_reembed_skipped, mark_summary_reembed_skipped, - set_chunk_embedding, set_chunk_embedding_for_signature, tree_active_signature, upsert_chunks, - ListChunksQuery, DB_DIR, GLOBAL_TOPIC_PURGE_MIGRATION_VERSION, + has_uncovered_reembed_work, is_source_ingested, list_chunks, mark_chunk_reembed_skipped, + mark_summary_reembed_skipped, set_chunk_embedding, set_chunk_embedding_for_signature, + tree_active_signature, upsert_chunks, ListChunksQuery, DB_DIR, + GLOBAL_TOPIC_PURGE_MIGRATION_VERSION, }; use crate::memory::config::MemoryConfig; use crate::memory::tree::store::{ @@ -81,6 +82,28 @@ fn clear_chunk_reembed_skipped_is_idempotent() { assert_eq!(count, 0); } +#[test] +fn setting_chunk_embedding_clears_matching_skip_marker() { + let (_tmp, cfg) = test_config(); + let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000); + upsert_chunks(&cfg, std::slice::from_ref(&c)).unwrap(); + let sig = tree_active_signature(&cfg); + mark_chunk_reembed_skipped(&cfg, &c.id, &sig, "body read failed: no content pointer").unwrap(); + + set_chunk_embedding_for_signature(&cfg, &c.id, &sig, &[0.1, 0.2]).unwrap(); + + let count: i64 = with_connection(&cfg, |conn| { + Ok(conn.query_row( + "SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped + WHERE chunk_id = ?1 AND model_signature = ?2", + params![c.id, sig], + |r| r.get(0), + )?) + }) + .unwrap(); + assert_eq!(count, 0); +} + #[test] fn summary_reembed_tombstone_roundtrips_and_clears() { let (_tmp, cfg) = test_config(); @@ -318,6 +341,45 @@ fn batch_embedding_lookup_unknown_ids_absent_from_map() { assert_eq!(map.get(&c.id).cloned(), Some(vec![0.1])); } +#[test] +fn legacy_body_read_skip_does_not_hide_reembed_work() { + let (_tmp, cfg) = test_config(); + let c = sample_chunk("persona/communication", 0, 1_700_000_000_000); + upsert_chunks(&cfg, std::slice::from_ref(&c)).unwrap(); + let sig = "bge-m3@1024"; + mark_chunk_reembed_skipped( + &cfg, + &c.id, + sig, + &format!( + "body read failed: no content pointer or raw refs for chunk {}", + c.id + ), + ) + .unwrap(); + + with_connection(&cfg, |conn| { + assert!(has_uncovered_reembed_work(conn, sig)?); + Ok(()) + }) + .unwrap(); +} + +#[test] +fn non_legacy_skip_still_hides_reembed_work() { + let (_tmp, cfg) = test_config(); + let c = sample_chunk("persona/communication", 0, 1_700_000_000_000); + upsert_chunks(&cfg, std::slice::from_ref(&c)).unwrap(); + let sig = "bge-m3@1024"; + mark_chunk_reembed_skipped(&cfg, &c.id, sig, "embed failed: provider rejected input").unwrap(); + + with_connection(&cfg, |conn| { + assert!(!has_uncovered_reembed_work(conn, sig)?); + Ok(()) + }) + .unwrap(); +} + #[test] fn batch_embedding_lookup_splits_id_list_above_per_batch_threshold() { let (_tmp, cfg) = test_config(); diff --git a/src/memory/store/content/read.rs b/src/memory/store/content/read.rs index fa3babfc..fb09df75 100644 --- a/src/memory/store/content/read.rs +++ b/src/memory/store/content/read.rs @@ -8,8 +8,9 @@ use std::path::{Component, Path, PathBuf}; use super::atomic::sha256_hex; use super::compose::split_front_matter; use crate::memory::chunks::{ - content_root, get_chunk_content_pointers, get_chunk_raw_refs, get_summary_content_pointers, - update_chunk_content_sha256, update_summary_content_sha256, RawRef, + content_root, get_chunk, get_chunk_content_pointers, get_chunk_raw_refs, + get_summary_content_pointers, update_chunk_content_sha256, update_summary_content_sha256, + RawRef, }; use crate::memory::config::MemoryConfig; @@ -136,10 +137,11 @@ pub fn read_chunk_body(config: &MemoryConfig, chunk_id: &str) -> anyhow::Result< } } - let (rel_path, expected_sha256) = get_chunk_content_pointers(config, chunk_id)? - .ok_or_else(|| anyhow::anyhow!("no content pointer or raw refs for chunk {chunk_id}"))?; + let Some((rel_path, expected_sha256)) = get_chunk_content_pointers(config, chunk_id)? else { + return read_legacy_chunk_preview(config, chunk_id); + }; if rel_path.is_empty() { - anyhow::bail!("empty content pointer and no raw refs for chunk {chunk_id}"); + return read_legacy_chunk_preview(config, chunk_id); } let abs_path = resolve_within_content_root(&content_root(config), &rel_path)?; let result = read_chunk_file(&abs_path)?; @@ -154,6 +156,16 @@ pub fn read_chunk_body(config: &MemoryConfig, chunk_id: &str) -> anyhow::Result< Ok(result.body) } +fn read_legacy_chunk_preview(config: &MemoryConfig, chunk_id: &str) -> anyhow::Result { + let Some(chunk) = get_chunk(config, chunk_id)? else { + anyhow::bail!("no content pointer or raw refs for chunk {chunk_id}"); + }; + if chunk.content.is_empty() { + anyhow::bail!("empty content pointer and no raw refs for chunk {chunk_id}"); + } + Ok(chunk.content) +} + fn read_chunk_body_from_raw(config: &MemoryConfig, refs: &[RawRef]) -> anyhow::Result { let root = content_root(config); let mut parts = Vec::with_capacity(refs.len()); diff --git a/src/memory/store/content/read_tests.rs b/src/memory/store/content/read_tests.rs index b0b850d5..8c669171 100644 --- a/src/memory/store/content/read_tests.rs +++ b/src/memory/store/content/read_tests.rs @@ -199,6 +199,16 @@ fn high_level_chunk_reader_uses_custom_root_and_repairs_checksum() { assert_eq!(repaired, staged[0].content_sha256); } +#[test] +fn high_level_chunk_reader_falls_back_to_legacy_content_column() { + let dir = TempDir::new().unwrap(); + let config = crate::memory::MemoryConfig::new(dir.path()); + let chunk = sample_chunk(); + crate::memory::chunks::upsert_chunks(&config, std::slice::from_ref(&chunk)).unwrap(); + + assert_eq!(read_chunk_body(&config, &chunk.id).unwrap(), chunk.content); +} + #[test] fn high_level_chunk_reader_joins_clamped_raw_references() { let dir = TempDir::new().unwrap(); From f2f2ef73ba2b20b2a6bcf148b5e2b115494cb505 Mon Sep 17 00:00:00 2001 From: hobertrand-hub Date: Sat, 25 Jul 2026 03:38:55 +0200 Subject: [PATCH 2/8] fix(memory): keep empty legacy chunks terminal Co-authored-by: Medulla --- docs/openhuman-memory-migration.md | 32 +++++++------ docs/openhuman-memory/README.md | 6 ++- .../openhuman-memory/sources-registry-sync.md | 6 ++- docs/plan/05-openhuman-compat-matrix.md | 9 ++-- src/memory/chunks/store_embed_tests.rs | 48 +++++++++++++++++++ src/memory/store/content/read.rs | 2 +- src/memory/store/content/read_tests.rs | 16 +++++++ 7 files changed, 95 insertions(+), 24 deletions(-) diff --git a/docs/openhuman-memory-migration.md b/docs/openhuman-memory-migration.md index 7c076a7d..7c656d31 100644 --- a/docs/openhuman-memory-migration.md +++ b/docs/openhuman-memory-migration.md @@ -4,10 +4,10 @@ This repository now has a Rust crate rooted at the repository root. The first migration target is the memory core: stable contracts, storage primitives, and testable in-process behavior before API or UI integrations. -TinyCortex owns the generic sync engine and provider pipelines behind its -optional `sync` feature. OpenHuman retains scheduling, credentials, RPC, -source-scope/redaction policy, and event-bus publishing, and supplies those -product concerns through the sync adapter traits. +TinyCortex owns reusable provider fetch, pagination, and canonicalization +pipeline mechanics behind its optional `sync` feature and injected traits. +OpenHuman owns the live sync runner, credentials, scheduling, source policy, +callbacks, RPC, product events, and projections. ## Source Modules @@ -41,10 +41,10 @@ The memory engine now lives under `src/memory/` as cohesive modules: `conversations/`, `archivist/`: specialized memory surfaces. Future host adapters should keep OpenHuman's layer rule: orchestration depends -on storage, but storage does not depend upward on orchestration. Generic sync -fetch/pagination/canonical-record mechanics live in this crate behind injected -traits; OpenHuman retains scheduling, credentials, source policy, event-bus -translation, RPC, and product projections. +on storage, but storage does not depend upward on orchestration. Reusable sync +fetch, pagination, and canonical-record mechanics live in this crate behind +injected traits; OpenHuman retains the live runner, scheduling, credentials, +callbacks, source policy, event-bus translation, RPC, and product projections. ## Migration Order @@ -81,10 +81,12 @@ are the current validation gates). | `conversations` | `memory_conversations` | JSONL transcript store, inverted index, persistence bus. | | `archivist` | `memory_archivist` | Conversation turns → one tree leaf (tool-JSON stripped). Tree-leaf sink injected. | -Per the ownership boundary, the live sync scheduler, OAuth/webhook callbacks, -credentials, and real LLM/embedding/network backends remain host-owned -(OpenHuman) and are represented here as injectable traits. Generic provider -pipelines and workspace reconciliation are crate-owned. Known follow-ups: consolidate legacy -`score::store` entity-index helpers around `store::entity_index`; restore the -deferred peripheral surfaces (tree `health`/`nlp`, retrieval RPC/fast paths, -obsidian/wiki-git content, controller/tool registries) as host adapters land. +Per the ownership boundary, the live sync runner, OAuth/webhook callbacks, +credentials, scheduling, policy, RPC, product events, and real +LLM/embedding/network backends remain host-owned (OpenHuman) and are represented +here as injectable traits. Reusable provider fetch, pagination, and +canonicalization pipeline mechanics are crate-owned. Known follow-ups: +consolidate legacy `score::store` entity-index helpers around +`store::entity_index`; restore the deferred peripheral surfaces (tree +`health`/`nlp`, retrieval RPC/fast paths, obsidian/wiki-git content, +controller/tool registries) as host adapters land. diff --git a/docs/openhuman-memory/README.md b/docs/openhuman-memory/README.md index 4dd0c7b3..bcef5d2e 100644 --- a/docs/openhuman-memory/README.md +++ b/docs/openhuman-memory/README.md @@ -5,8 +5,10 @@ specifications for the TinyCortex migration. Each document captures the observed OpenHuman contract, the required data attributes, invariants, and the recommended TinyCortex landing area. -Boundary: TinyCortex does not own memory sync. The OpenHuman application -owns the sync module and decides when data is ingested on demand. These specs +Boundary: TinyCortex owns reusable provider fetch, pagination, and +canonicalization pipeline mechanics behind injected traits. OpenHuman owns the +live sync runner, credentials, scheduling, source policy, callbacks, RPC, +product events, and decides when data is ingested on demand. These specs describe the contracts TinyCortex exposes after OpenHuman supplies source data. Source checkout used for this pass: diff --git a/docs/openhuman-memory/sources-registry-sync.md b/docs/openhuman-memory/sources-registry-sync.md index d18e1b7b..f807fd07 100644 --- a/docs/openhuman-memory/sources-registry-sync.md +++ b/docs/openhuman-memory/sources-registry-sync.md @@ -191,5 +191,7 @@ src/memory/ingest/canonicalize/ Port order: source kind/types/validation, registry patch semantics, reader trait and static reader contracts, canonicalizer pure functions, then -OpenHuman-facing ingest and sync adapters. The provider pipeline is crate-owned; -the live scheduler, credentials, policy, and product events stay in OpenHuman. +OpenHuman-facing ingest and sync adapters. Reusable provider fetch, pagination, +and canonicalization pipeline mechanics are crate-owned; the live sync runner, +credentials, scheduling, callbacks, policy, RPC, and product events stay in +OpenHuman. diff --git a/docs/plan/05-openhuman-compat-matrix.md b/docs/plan/05-openhuman-compat-matrix.md index 59dfe755..933345dd 100644 --- a/docs/plan/05-openhuman-compat-matrix.md +++ b/docs/plan/05-openhuman-compat-matrix.md @@ -71,10 +71,11 @@ kind fields + sync budgets `types.rs:68-146`; discriminator validation reconciliation with `source_kind = raw_file` so interrupted syncs don't strand raw files (spec: sources-registry-sync.md §Raw Archive Coverage). -Generic Composio provider fetch/pagination pipelines are crate-owned behind -injected network and persistence traits. Provider credentials, the scheduler, -source policy, RPC, events, and non-Composio product integrations remain -host-owned. **Do not port the live sync scheduler.** +Generic provider fetch, pagination, and canonicalization pipeline mechanics are +crate-owned behind injected network and persistence traits. Provider +credentials, the live sync runner, scheduling, callbacks, source policy, RPC, +events, and non-Composio product integrations remain host-owned. **Do not port +the live sync scheduler.** --- diff --git a/src/memory/chunks/store_embed_tests.rs b/src/memory/chunks/store_embed_tests.rs index b8b40889..d4085c3e 100644 --- a/src/memory/chunks/store_embed_tests.rs +++ b/src/memory/chunks/store_embed_tests.rs @@ -365,6 +365,54 @@ fn legacy_body_read_skip_does_not_hide_reembed_work() { .unwrap(); } +#[test] +fn legacy_empty_pointer_skip_does_not_hide_reembed_work() { + let (_tmp, cfg) = test_config(); + let c = sample_chunk("persona/communication", 0, 1_700_000_000_000); + upsert_chunks(&cfg, std::slice::from_ref(&c)).unwrap(); + let sig = "bge-m3@1024"; + mark_chunk_reembed_skipped( + &cfg, + &c.id, + sig, + &format!( + "body read failed: empty content pointer and no raw refs for chunk {}", + c.id + ), + ) + .unwrap(); + + with_connection(&cfg, |conn| { + assert!(has_uncovered_reembed_work(conn, sig)?); + Ok(()) + }) + .unwrap(); +} + +#[test] +fn empty_legacy_content_skip_hides_reembed_work() { + let (_tmp, cfg) = test_config(); + let c = sample_chunk("persona/communication", 0, 1_700_000_000_000); + upsert_chunks(&cfg, std::slice::from_ref(&c)).unwrap(); + let sig = "bge-m3@1024"; + mark_chunk_reembed_skipped( + &cfg, + &c.id, + sig, + &format!( + "body read failed: legacy chunk content empty for chunk {}", + c.id + ), + ) + .unwrap(); + + with_connection(&cfg, |conn| { + assert!(!has_uncovered_reembed_work(conn, sig)?); + Ok(()) + }) + .unwrap(); +} + #[test] fn non_legacy_skip_still_hides_reembed_work() { let (_tmp, cfg) = test_config(); diff --git a/src/memory/store/content/read.rs b/src/memory/store/content/read.rs index fb09df75..9f91cdb6 100644 --- a/src/memory/store/content/read.rs +++ b/src/memory/store/content/read.rs @@ -161,7 +161,7 @@ fn read_legacy_chunk_preview(config: &MemoryConfig, chunk_id: &str) -> anyhow::R anyhow::bail!("no content pointer or raw refs for chunk {chunk_id}"); }; if chunk.content.is_empty() { - anyhow::bail!("empty content pointer and no raw refs for chunk {chunk_id}"); + anyhow::bail!("legacy chunk content empty for chunk {chunk_id}"); } Ok(chunk.content) } diff --git a/src/memory/store/content/read_tests.rs b/src/memory/store/content/read_tests.rs index 8c669171..332d4322 100644 --- a/src/memory/store/content/read_tests.rs +++ b/src/memory/store/content/read_tests.rs @@ -209,6 +209,22 @@ fn high_level_chunk_reader_falls_back_to_legacy_content_column() { assert_eq!(read_chunk_body(&config, &chunk.id).unwrap(), chunk.content); } +#[test] +fn high_level_chunk_reader_empty_legacy_content_uses_terminal_error() { + let dir = TempDir::new().unwrap(); + let config = crate::memory::MemoryConfig::new(dir.path()); + let mut chunk = sample_chunk(); + chunk.content.clear(); + crate::memory::chunks::upsert_chunks(&config, std::slice::from_ref(&chunk)).unwrap(); + + let err = read_chunk_body(&config, &chunk.id).unwrap_err(); + + assert_eq!( + err.to_string(), + format!("legacy chunk content empty for chunk {}", chunk.id) + ); +} + #[test] fn high_level_chunk_reader_joins_clamped_raw_references() { let dir = TempDir::new().unwrap(); From e94c673d15ac421566e040e902725ec6d9fc7b0b Mon Sep 17 00:00:00 2001 From: hobertrand-hub Date: Mon, 27 Jul 2026 00:23:14 +0200 Subject: [PATCH 3/8] Handle mem_src scopes in source retrieval Co-authored-by: Medulla --- src/memory/retrieval/fast.rs | 19 +++++++++++++++++-- src/memory/retrieval/source.rs | 3 +++ src/memory/retrieval/source_tests.rs | 4 ++++ 3 files changed, 24 insertions(+), 2 deletions(-) diff --git a/src/memory/retrieval/fast.rs b/src/memory/retrieval/fast.rs index 2322b9ad..318552e7 100644 --- a/src/memory/retrieval/fast.rs +++ b/src/memory/retrieval/fast.rs @@ -208,7 +208,7 @@ fn resolve_local( let mut hits = Vec::new(); for (id, _) in ordered { if let Some(mut hit) = by_id.remove(&id) { - if source_scope.is_some_and(|scope| !scope.contains(&hit.tree_scope)) { + if source_scope.is_some_and(|scope| !source_scope_allows(scope, &hit.tree_scope)) { continue; } hit.score = coverage.get(&id).copied().unwrap_or_default(); @@ -239,7 +239,9 @@ async fn dense( ) .await?; if let Some(scope) = source_scope { - response.hits.retain(|hit| scope.contains(&hit.tree_scope)); + response + .hits + .retain(|hit| source_scope_allows(scope, &hit.tree_scope)); } let total = response.hits.len(); response.hits.truncate(limit); @@ -282,6 +284,19 @@ fn dedup_ids(ids: impl Iterator) -> Vec { ids.filter(|id| seen.insert(id.clone())).collect() } +fn source_scope_allows(scope: &HashSet, tree_scope: &str) -> bool { + if scope.contains(tree_scope) { + return true; + } + extract_mem_src_id(tree_scope).is_some_and(|id| scope.contains(id)) +} + +fn extract_mem_src_id(value: &str) -> Option<&str> { + let rest = value.strip_prefix("mem_src:")?; + let (id, _) = rest.split_once(':')?; + (!id.is_empty()).then_some(id) +} + #[cfg(test)] #[path = "fast_tests.rs"] mod tests; diff --git a/src/memory/retrieval/source.rs b/src/memory/retrieval/source.rs index fffeb784..077f00f4 100644 --- a/src/memory/retrieval/source.rs +++ b/src/memory/retrieval/source.rs @@ -236,6 +236,9 @@ fn scope_matches_kind(scope: &str, kind_prefix: &str) -> bool { if lower.starts_with(&format!("{kind_prefix}:")) { return true; } + if kind_prefix == SourceKind::Document.as_str() && lower.starts_with("mem_src:") { + return true; + } PLATFORM_KINDS .iter() .any(|(platform, kind)| *kind == kind_prefix && lower.starts_with(&format!("{platform}:"))) diff --git a/src/memory/retrieval/source_tests.rs b/src/memory/retrieval/source_tests.rs index 343aded7..b8d0e3db 100644 --- a/src/memory/retrieval/source_tests.rs +++ b/src/memory/retrieval/source_tests.rs @@ -195,6 +195,10 @@ fn scope_prefix_matching_known_platforms() { assert!(scope_matches_kind("gmail:alice", "email")); assert!(scope_matches_kind("notion:page123", "document")); assert!(scope_matches_kind("linear:conn-1:issue-abc", "document")); + assert!(scope_matches_kind( + "mem_src:src-folder-9:Slides_Notes/example.md", + "document" + )); assert!(!scope_matches_kind("slack:#eng", "email")); assert!(scope_matches_kind("chat:custom", "chat")); } From 20bf04ae80c8a4c1510b809db0f029f5bf19b3c4 Mon Sep 17 00:00:00 2001 From: hobertrand-hub Date: Mon, 27 Jul 2026 04:26:20 +0200 Subject: [PATCH 4/8] Clarify mem_src source scope filtering Co-authored-by: Medulla --- src/memory/retrieval/fast.rs | 31 ++++++++++++++++++++++++++-- src/memory/retrieval/fast_tests.rs | 23 +++++++++++++++++++++ src/memory/retrieval/source.rs | 4 ++++ src/memory/retrieval/source_tests.rs | 4 ++++ 4 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/memory/retrieval/fast.rs b/src/memory/retrieval/fast.rs index 318552e7..000b5651 100644 --- a/src/memory/retrieval/fast.rs +++ b/src/memory/retrieval/fast.rs @@ -288,15 +288,42 @@ fn source_scope_allows(scope: &HashSet, tree_scope: &str) -> bool { if scope.contains(tree_scope) { return true; } - extract_mem_src_id(tree_scope).is_some_and(|id| scope.contains(id)) + // Caller scopes are source selectors: either the exact tree scope, a bare + // source id such as `src-folder-9`, or the collection prefix + // `mem_src:src-folder-9`. Per-file trees store the full + // `mem_src::` scope, so extract the collection source id + // before applying source-level filtering. + let Some(id) = extract_mem_src_id(tree_scope) else { + return false; + }; + scope.contains(id) + || scope + .iter() + .any(|allowed| mem_src_scope_selects_id(allowed, id)) } fn extract_mem_src_id(value: &str) -> Option<&str> { - let rest = value.strip_prefix("mem_src:")?; + let prefix = value.get(..8)?; + if !prefix.eq_ignore_ascii_case("mem_src:") { + return None; + } + let rest = &value[8..]; let (id, _) = rest.split_once(':')?; (!id.is_empty()).then_some(id) } +fn mem_src_scope_selects_id(value: &str, expected_id: &str) -> bool { + let Some(prefix) = value.get(..8) else { + return false; + }; + if !prefix.eq_ignore_ascii_case("mem_src:") { + return false; + } + let rest = &value[8..]; + let id = rest.split_once(':').map_or(rest, |(id, _)| id); + !id.is_empty() && id == expected_id +} + #[cfg(test)] #[path = "fast_tests.rs"] mod tests; diff --git a/src/memory/retrieval/fast_tests.rs b/src/memory/retrieval/fast_tests.rs index 2b38a438..fa909816 100644 --- a/src/memory/retrieval/fast_tests.rs +++ b/src/memory/retrieval/fast_tests.rs @@ -17,6 +17,29 @@ fn options_and_ids_are_bounded_deterministically() { ); } +#[test] +fn mem_src_scope_filter_accepts_bare_id_collection_prefix_and_exact_scope() { + let tree_scope = "Mem_Src:src-folder-9:Slides_Notes/example.md"; + + assert_eq!(extract_mem_src_id(tree_scope), Some("src-folder-9")); + assert!(source_scope_allows( + &HashSet::from(["src-folder-9".to_string()]), + tree_scope + )); + assert!(source_scope_allows( + &HashSet::from(["mem_src:src-folder-9".to_string()]), + tree_scope + )); + assert!(source_scope_allows( + &HashSet::from([tree_scope.to_string()]), + tree_scope + )); + assert!(!source_scope_allows( + &HashSet::from(["src-other".to_string()]), + tree_scope + )); +} + #[tokio::test] async fn blank_query_is_empty_without_opening_storage() { let (_temp, config) = test_config(); diff --git a/src/memory/retrieval/source.rs b/src/memory/retrieval/source.rs index 077f00f4..454ee934 100644 --- a/src/memory/retrieval/source.rs +++ b/src/memory/retrieval/source.rs @@ -236,6 +236,10 @@ fn scope_matches_kind(scope: &str, kind_prefix: &str) -> bool { if lower.starts_with(&format!("{kind_prefix}:")) { return true; } + // OpenHuman document sources encode per-file tree scopes as + // `mem_src::`, so kind filtering must classify the whole + // tree family as documents while exact source-id callers still use the + // source_id path above. if kind_prefix == SourceKind::Document.as_str() && lower.starts_with("mem_src:") { return true; } diff --git a/src/memory/retrieval/source_tests.rs b/src/memory/retrieval/source_tests.rs index b8d0e3db..deb3b7bc 100644 --- a/src/memory/retrieval/source_tests.rs +++ b/src/memory/retrieval/source_tests.rs @@ -199,6 +199,10 @@ fn scope_prefix_matching_known_platforms() { "mem_src:src-folder-9:Slides_Notes/example.md", "document" )); + assert!(scope_matches_kind( + "Mem_Src:src-folder-9:Slides_Notes/example.md", + "document" + )); assert!(!scope_matches_kind("slack:#eng", "email")); assert!(scope_matches_kind("chat:custom", "chat")); } From d92de6d58ba776d0c7b1793458325adab4ca6792 Mon Sep 17 00:00:00 2001 From: hobertrand-hub Date: Mon, 27 Jul 2026 04:30:08 +0200 Subject: [PATCH 5/8] Document legacy reembed skip reasons Co-authored-by: Medulla --- src/memory/chunks/embeddings.rs | 24 +++++++++++++++---- src/memory/chunks/store_embed_tests.rs | 10 +++++--- src/memory/store/content/read.rs | 32 ++++++++++++++++++++++---- src/memory/store/content/read_tests.rs | 2 +- 4 files changed, 55 insertions(+), 13 deletions(-) diff --git a/src/memory/chunks/embeddings.rs b/src/memory/chunks/embeddings.rs index 1aa7306a..f7504adf 100644 --- a/src/memory/chunks/embeddings.rs +++ b/src/memory/chunks/embeddings.rs @@ -13,6 +13,9 @@ use rusqlite::{Connection, OptionalExtension}; use std::collections::HashMap; use crate::memory::config::MemoryConfig; +use crate::memory::store::content::read::{ + LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX, LEGACY_NO_CONTENT_POINTER_REASON_PREFIX, +}; /// The active embedding vector dimension for `config`. Drives the legacy /// migration's dim-match decision. @@ -412,20 +415,33 @@ pub fn has_uncovered_reembed_work( ) -> rusqlite::Result { let variants = signature_variants(model_signature); let sig_clause = signature_in_clause(variants.len(), 1); - let params: Vec<&dyn rusqlite::ToSql> = variants + // Retryable skip reasons for legacy content-pointer rows; matched after the + // signature-variant params (which occupy `?1..=?variants.len()`). + let retry_idx_1 = variants.len() + 1; + let retry_idx_2 = variants.len() + 2; + let missing_pointer_retry_like = + format!("body read failed: {LEGACY_NO_CONTENT_POINTER_REASON_PREFIX}%"); + // Legacy-only: older readers emitted this for present-but-empty content + // pointers. Current readers filter those pointers before returning `Some`, + // but existing skip rows must still become retriable after the fallback fix. + let empty_pointer_retry_like = + format!("body read failed: {LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX}%"); + let mut params: Vec<&dyn rusqlite::ToSql> = variants .iter() .map(|variant| variant as &dyn rusqlite::ToSql) .collect(); + params.push(&missing_pointer_retry_like as &dyn rusqlite::ToSql); + params.push(&empty_pointer_retry_like as &dyn rusqlite::ToSql); conn.query_row( -&format!( + &format!( "SELECT EXISTS( SELECT 1 FROM mem_tree_chunks c WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e WHERE e.chunk_id = c.id AND e.model_signature {sig_clause}) AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk WHERE sk.chunk_id = c.id AND sk.model_signature {sig_clause} - AND sk.reason NOT LIKE 'body read failed: no content pointer or raw refs for chunk %' - AND sk.reason NOT LIKE 'body read failed: empty content pointer and no raw refs for chunk %')) + AND sk.reason NOT LIKE ?{retry_idx_1} + AND sk.reason NOT LIKE ?{retry_idx_2})) OR EXISTS( SELECT 1 FROM mem_tree_summaries s WHERE s.deleted = 0 diff --git a/src/memory/chunks/store_embed_tests.rs b/src/memory/chunks/store_embed_tests.rs index d4085c3e..a0d55bdf 100644 --- a/src/memory/chunks/store_embed_tests.rs +++ b/src/memory/chunks/store_embed_tests.rs @@ -25,6 +25,10 @@ use super::{ GLOBAL_TOPIC_PURGE_MIGRATION_VERSION, }; use crate::memory::config::MemoryConfig; +use crate::memory::store::content::read::{ + LEGACY_EMPTY_CHUNK_CONTENT_REASON_PREFIX, LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX, + LEGACY_NO_CONTENT_POINTER_REASON_PREFIX, +}; use crate::memory::tree::store::{ insert_summary_tx, insert_tree, SummaryNode, Tree, TreeKind, TreeStatus, }; @@ -352,7 +356,7 @@ fn legacy_body_read_skip_does_not_hide_reembed_work() { &c.id, sig, &format!( - "body read failed: no content pointer or raw refs for chunk {}", + "body read failed: {LEGACY_NO_CONTENT_POINTER_REASON_PREFIX}{}", c.id ), ) @@ -376,7 +380,7 @@ fn legacy_empty_pointer_skip_does_not_hide_reembed_work() { &c.id, sig, &format!( - "body read failed: empty content pointer and no raw refs for chunk {}", + "body read failed: {LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX}{}", c.id ), ) @@ -400,7 +404,7 @@ fn empty_legacy_content_skip_hides_reembed_work() { &c.id, sig, &format!( - "body read failed: legacy chunk content empty for chunk {}", + "body read failed: {LEGACY_EMPTY_CHUNK_CONTENT_REASON_PREFIX}{}", c.id ), ) diff --git a/src/memory/store/content/read.rs b/src/memory/store/content/read.rs index 9f91cdb6..5a27bda3 100644 --- a/src/memory/store/content/read.rs +++ b/src/memory/store/content/read.rs @@ -14,6 +14,24 @@ use crate::memory::chunks::{ }; use crate::memory::config::MemoryConfig; +/// Prefix for retryable legacy rows that were written before staged content +/// files/raw references existed. `has_uncovered_reembed_work` matches this text +/// inside `body read failed: ...` skip reasons to keep those chunks eligible for +/// backfill once a legacy `content` column fallback is available. +pub(crate) const LEGACY_NO_CONTENT_POINTER_REASON_PREFIX: &str = + "no content pointer or raw refs for chunk "; + +/// Prefix preserved for pre-fix skip rows written by older readers when the +/// staged content pointer was present-but-empty. Current reads no longer emit +/// this text because `get_chunk_content_pointers` filters empty paths before +/// returning `Some`. +pub(crate) const LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX: &str = + "empty content pointer and no raw refs for chunk "; + +/// Prefix for terminal legacy rows whose inline content column is empty. +pub(crate) const LEGACY_EMPTY_CHUNK_CONTENT_REASON_PREFIX: &str = + "legacy chunk content empty for chunk "; + /// Resolve a DB-stored relative forward-slash path against `content_root`, /// rejecting any traversal (`..`), absolute, or non-normal component. /// @@ -140,9 +158,6 @@ pub fn read_chunk_body(config: &MemoryConfig, chunk_id: &str) -> anyhow::Result< let Some((rel_path, expected_sha256)) = get_chunk_content_pointers(config, chunk_id)? else { return read_legacy_chunk_preview(config, chunk_id); }; - if rel_path.is_empty() { - return read_legacy_chunk_preview(config, chunk_id); - } let abs_path = resolve_within_content_root(&content_root(config), &rel_path)?; let result = read_chunk_file(&abs_path)?; if result.sha256 != expected_sha256 { @@ -156,12 +171,19 @@ pub fn read_chunk_body(config: &MemoryConfig, chunk_id: &str) -> anyhow::Result< Ok(result.body) } +/// Fallback reader for pre-content-store chunk rows. +/// +/// Its error text prefixes are part of the reembed-backfill contract: +/// `embeddings::has_uncovered_reembed_work` matches the wrapped +/// `body read failed: ...` skip reasons to decide which historical failures are +/// retryable after this fallback exists. Keep the constants above in sync with +/// that SQL before changing wording here. fn read_legacy_chunk_preview(config: &MemoryConfig, chunk_id: &str) -> anyhow::Result { let Some(chunk) = get_chunk(config, chunk_id)? else { - anyhow::bail!("no content pointer or raw refs for chunk {chunk_id}"); + anyhow::bail!("{LEGACY_NO_CONTENT_POINTER_REASON_PREFIX}{chunk_id}"); }; if chunk.content.is_empty() { - anyhow::bail!("legacy chunk content empty for chunk {chunk_id}"); + anyhow::bail!("{LEGACY_EMPTY_CHUNK_CONTENT_REASON_PREFIX}{chunk_id}"); } Ok(chunk.content) } diff --git a/src/memory/store/content/read_tests.rs b/src/memory/store/content/read_tests.rs index 332d4322..8927fcd6 100644 --- a/src/memory/store/content/read_tests.rs +++ b/src/memory/store/content/read_tests.rs @@ -221,7 +221,7 @@ fn high_level_chunk_reader_empty_legacy_content_uses_terminal_error() { assert_eq!( err.to_string(), - format!("legacy chunk content empty for chunk {}", chunk.id) + format!("{LEGACY_EMPTY_CHUNK_CONTENT_REASON_PREFIX}{}", chunk.id) ); } From 9333067a61e3e4dbe30a4b6a3b44d4882b725d01 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 10 Aug 2026 16:54:18 +0300 Subject: [PATCH 6/8] fix(docs): don't intra-doc-link a private signature helper signature_variants' doc comment linked to signature_in_clause, which is pub(crate); rustdoc rejects that link with -D rustdoc::private_intra_doc_links and the CI Document step fails. The error predates this PR (current main's own push run fails the same way); keep the reference as plain code text. Co-authored-by: Medulla --- src/memory/chunks/signature.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/memory/chunks/signature.rs b/src/memory/chunks/signature.rs index 0946f9e3..a83d2397 100644 --- a/src/memory/chunks/signature.rs +++ b/src/memory/chunks/signature.rs @@ -118,7 +118,7 @@ pub fn signatures_equivalent(a: &str, b: &str) -> bool { /// Every spelling of `signature` a stored row might carry, `signature` first. /// -/// Bind these into an `IN (…)` predicate (see [`signature_in_clause`]) instead +/// Bind these into an `IN (…)` predicate (see `signature_in_clause`) instead /// of `= ?`: that is what makes a read find rows written under the other /// convention without rewriting them. pub fn signature_variants(signature: &str) -> Vec { From 981a37426603acfd2a44b3da5db65994abb64588 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 10 Aug 2026 17:16:03 +0300 Subject: [PATCH 7/8] refactor(chunks): split signature-aware embedding reads into embeddings_query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Embeddings.rs had grown to 541 lines, over the repo's 500-line ceiling (as flagged in review). Move the read-side, signature-aware query functions — get_chunk_embedding_for_signature, get_chunk_embedding, get_chunk_embeddings_for_signature_batch, get_chunk_embeddings_batch and has_uncovered_reembed_work (plus the embedding_from_blob decoder and the MAX_EMBEDDING_BATCH cap) — into a focused embeddings_query sibling. embeddings.rs keeps the writers, upserts, and re-embed tombstones; both remain under 500 lines and the public chunks re-export surface is unchanged, so no caller or test file needs edits. Co-authored-by: Medulla --- src/memory/chunks/embeddings.rs | 239 +------------------------ src/memory/chunks/embeddings_query.rs | 247 ++++++++++++++++++++++++++ src/memory/chunks/mod.rs | 22 ++- 3 files changed, 267 insertions(+), 241 deletions(-) create mode 100644 src/memory/chunks/embeddings_query.rs diff --git a/src/memory/chunks/embeddings.rs b/src/memory/chunks/embeddings.rs index f7504adf..034b0c3b 100644 --- a/src/memory/chunks/embeddings.rs +++ b/src/memory/chunks/embeddings.rs @@ -3,19 +3,17 @@ //! Embeddings are stored in the `mem_tree_chunk_embeddings` sidecar table keyed //! by `(chunk_id, model_signature)` so multiple vector spaces can coexist. This //! module is pure storage: it does not compute embeddings (that backend is not -//! ported here) — callers pass vectors in. +//! ported here) — callers pass vectors in. The signature-aware *read* side +//! lives in the [`super::embeddings_query`] sibling, which this module's +//! re-exports also expose at the `chunks` level. use super::connection::with_connection; -use super::signature::{format_signature, signature_in_clause, signature_variants}; +use super::signature::format_signature; use anyhow::{Context, Result}; use chrono::Utc; -use rusqlite::{Connection, OptionalExtension}; -use std::collections::HashMap; +use rusqlite::Connection; use crate::memory::config::MemoryConfig; -use crate::memory::store::content::read::{ - LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX, LEGACY_NO_CONTENT_POINTER_REASON_PREFIX, -}; /// The active embedding vector dimension for `config`. Drives the legacy /// migration's dim-match decision. @@ -31,7 +29,7 @@ pub(crate) fn active_embedding_dims(config: &MemoryConfig) -> usize { /// This is the canonical `provider=…;model=…;dims=…` spelling, shared with the /// namespace store. Rows written under the tree's older `{model}@{dims}` /// spelling are still found, because per-signature reads match every variant -/// (see [`signature_variants`]) rather than one exact string. +/// (see [`super::signature::signature_variants`]) rather than one exact string. pub fn tree_active_signature(config: &MemoryConfig) -> String { format_signature( &config.embedding.provider, @@ -310,232 +308,7 @@ pub(crate) fn validate_reembed_skip_key<'a>(label: &str, value: &'a str) -> Resu Ok(trimmed) } -/// Fetch a chunk embedding for one provider/model/dimension signature — under -/// any of its spellings (see [`signature_variants`]). -/// -/// Returns `Ok(None)` when no row exists for `(chunk_id, model_signature)` — -/// absence is not an error. -/// -/// # Errors -/// Returns `Err` if the query fails, or if `embedding_from_blob` rejects -/// the stored blob (negative/zero-remainder-mismatched dim, or a blob length -/// not a multiple of 4 bytes — both indicate on-disk corruption of this row, -/// not a normal "no embedding" state). -pub fn get_chunk_embedding_for_signature( - config: &MemoryConfig, - chunk_id: &str, - model_signature: &str, -) -> Result>> { - let variants = signature_variants(model_signature); - with_connection(config, |conn| { - let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(variants.len() + 1); - params.push(&chunk_id as &dyn rusqlite::ToSql); - for variant in &variants { - params.push(variant as &dyn rusqlite::ToSql); - } - let row: Option<(Vec, i64)> = conn - .query_row( - &format!( - "SELECT vector, dim - FROM mem_tree_chunk_embeddings - WHERE chunk_id = ?1 AND model_signature {}", - signature_in_clause(variants.len(), 2) - ), - params.as_slice(), - |r| Ok((r.get(0)?, r.get(1)?)), - ) - .optional()?; - match row { - None => Ok(None), - Some((bytes, dim)) => embedding_from_blob(&bytes, dim, "chunk embedding"), - } - }) -} - -/// Fetch a chunk's embedding for the active model signature (see -/// [`tree_active_signature`]). See [`get_chunk_embedding_for_signature`] for -/// the return/error contract. -pub fn get_chunk_embedding(config: &MemoryConfig, chunk_id: &str) -> Result>> { - let signature = tree_active_signature(config); - get_chunk_embedding_for_signature(config, chunk_id, &signature) -} - /// Little-endian `f32` vector → `BLOB`. The inverse of `embedding_from_blob`. pub fn embedding_to_blob(embedding: &[f32]) -> Vec { embedding.iter().flat_map(|f| f.to_le_bytes()).collect() } - -/// Decode a little-endian `f32` vector `BLOB` back into `Vec`, validating -/// it against the DB's own recorded `dim` column. `label` only qualifies the -/// error message (e.g. `"chunk embedding"` vs. a future summary-embedding -/// caller). -/// -/// Always returns `Ok(Some(_))` on success — the `Option` in the return type -/// exists purely so callers can `?`-propagate this directly from inside a -/// `match row { None => Ok(None), Some(..) => embedding_from_blob(..) }` arm -/// without an extra `.map`. -/// -/// # Errors -/// Returns `Err` if `dim` is negative, if `bytes.len()` is not a multiple of -/// 4, or if the decoded float count does not equal `dim` — all three -/// indicate the stored row is internally inconsistent (corruption or a bug -/// upstream), not a normal "different embedding space" mismatch (that case is -/// handled by comparing signatures/dims *before* calling this, e.g. in -/// [`super::migrations::migrate_legacy_embeddings_to_sidecar`]). -fn embedding_from_blob(bytes: &[u8], dim: i64, label: &str) -> Result>> { - if dim < 0 { - anyhow::bail!("{label} has negative dimension {dim}"); - } - if !bytes.len().is_multiple_of(4) { - anyhow::bail!("{label} blob length {} not a multiple of 4", bytes.len()); - } - let floats: Vec = bytes - .chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) - .collect(); - if floats.len() != dim as usize { - anyhow::bail!( - "{label} dimension mismatch: dim column says {dim}, blob contains {} floats", - floats.len() - ); - } - Ok(Some(floats)) -} - -/// Whether any live chunk or summary lacks both an embedding and terminal -/// tombstone. -/// -/// Coverage counts a vector written under *any* spelling of the signature: a -/// row that only looks uncovered because the convention changed is not work, -/// and re-embedding it would spend provider quota to reproduce a vector that -/// is already on disk. -pub fn has_uncovered_reembed_work( - conn: &Connection, - model_signature: &str, -) -> rusqlite::Result { - let variants = signature_variants(model_signature); - let sig_clause = signature_in_clause(variants.len(), 1); - // Retryable skip reasons for legacy content-pointer rows; matched after the - // signature-variant params (which occupy `?1..=?variants.len()`). - let retry_idx_1 = variants.len() + 1; - let retry_idx_2 = variants.len() + 2; - let missing_pointer_retry_like = - format!("body read failed: {LEGACY_NO_CONTENT_POINTER_REASON_PREFIX}%"); - // Legacy-only: older readers emitted this for present-but-empty content - // pointers. Current readers filter those pointers before returning `Some`, - // but existing skip rows must still become retriable after the fallback fix. - let empty_pointer_retry_like = - format!("body read failed: {LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX}%"); - let mut params: Vec<&dyn rusqlite::ToSql> = variants - .iter() - .map(|variant| variant as &dyn rusqlite::ToSql) - .collect(); - params.push(&missing_pointer_retry_like as &dyn rusqlite::ToSql); - params.push(&empty_pointer_retry_like as &dyn rusqlite::ToSql); - conn.query_row( - &format!( - "SELECT EXISTS( - SELECT 1 FROM mem_tree_chunks c - WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e - WHERE e.chunk_id = c.id AND e.model_signature {sig_clause}) - AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk - WHERE sk.chunk_id = c.id AND sk.model_signature {sig_clause} - AND sk.reason NOT LIKE ?{retry_idx_1} - AND sk.reason NOT LIKE ?{retry_idx_2})) - OR EXISTS( - SELECT 1 FROM mem_tree_summaries s - WHERE s.deleted = 0 - AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e - WHERE e.summary_id = s.id AND e.model_signature {sig_clause}) - AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk - WHERE sk.summary_id = s.id AND sk.model_signature {sig_clause}))" - ), - params.as_slice(), - |row| row.get(0), - ) -} - -/// Defensive cap for batched `IN (?,?,…)` reads, well below SQLite's -/// `SQLITE_MAX_VARIABLE_NUMBER` (32 766). -const MAX_EMBEDDING_BATCH: usize = 500; - -/// Batched read of chunk embeddings under a single `model_signature`. -/// -/// Returns a `HashMap>` containing only the chunks that have -/// a vector under `model_signature`. Missing chunks are simply absent (callers -/// treat that the same as a `None` from the single-row helper). -/// -/// `chunk_ids` is split into windows of at most `MAX_EMBEDDING_BATCH` so a -/// single query never approaches SQLite's bound-parameter limit; each window -/// runs as its own `SELECT ... WHERE chunk_id IN (...)` inside the same -/// [`super::connection::with_connection`] call (not separately transacted — -/// reads only). -/// -/// # Errors -/// Returns `Err` if `chunk_ids` is non-empty and any window's query -/// preparation, execution, or blob decoding (`embedding_from_blob`) fails. -/// Returns `Ok(HashMap::new())` immediately (no DB access) when `chunk_ids` -/// is empty. -pub fn get_chunk_embeddings_for_signature_batch( - config: &MemoryConfig, - chunk_ids: &[String], - model_signature: &str, -) -> Result>> { - if chunk_ids.is_empty() { - return Ok(HashMap::new()); - } - let variants = signature_variants(model_signature); - with_connection(config, |conn| { - let mut out: HashMap> = HashMap::with_capacity(chunk_ids.len()); - for window in chunk_ids.chunks(MAX_EMBEDDING_BATCH) { - let placeholders = std::iter::repeat_n("?", window.len()) - .collect::>() - .join(","); - let sql = format!( - "SELECT chunk_id, vector, dim - FROM mem_tree_chunk_embeddings - WHERE chunk_id IN ({placeholders}) - AND model_signature {sig_clause}", - sig_clause = signature_in_clause(variants.len(), window.len() + 1), - ); - let mut stmt = conn - .prepare(&sql) - .context("prepare get_chunk_embeddings_for_signature_batch")?; - let mut params: Vec<&dyn rusqlite::ToSql> = - Vec::with_capacity(window.len() + variants.len()); - for id in window { - params.push(id as &dyn rusqlite::ToSql); - } - for variant in &variants { - params.push(variant as &dyn rusqlite::ToSql); - } - let rows = stmt - .query_map(params.as_slice(), |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, Vec>(1)?, - row.get::<_, i64>(2)?, - )) - }) - .context("query get_chunk_embeddings_for_signature_batch")?; - for row in rows { - let (chunk_id, bytes, dim) = row?; - if let Some(v) = embedding_from_blob(&bytes, dim, "chunk embedding")? { - out.insert(chunk_id, v); - } - } - } - Ok(out) - }) -} - -/// Batched read of chunk embeddings under the **active** model signature. See -/// [`get_chunk_embeddings_for_signature_batch`] for the batching and error -/// contract. -pub fn get_chunk_embeddings_batch( - config: &MemoryConfig, - chunk_ids: &[String], -) -> Result>> { - let signature = tree_active_signature(config); - get_chunk_embeddings_for_signature_batch(config, chunk_ids, &signature) -} diff --git a/src/memory/chunks/embeddings_query.rs b/src/memory/chunks/embeddings_query.rs new file mode 100644 index 00000000..09a3b8d2 --- /dev/null +++ b/src/memory/chunks/embeddings_query.rs @@ -0,0 +1,247 @@ +//! Signature-aware embedding **read** queries for the chunk store. +//! +//! Split out of `embeddings` so that module stays a focused writer-side +//! accessor and re-embed tombstone store, and this one owns the per-signature +//! read paths. Every read matches *any* spelling of a vector space (see +//! [`super::signature::signature_variants`]) so a store that changed +//! conventions can still see its own prior vectors. +//! +//! No embeddings are computed here — embeddings.rs takes vectors on the write +//! side; this module only reads them back. + +use super::connection::with_connection; +use super::embeddings::tree_active_signature; +use super::signature::{signature_in_clause, signature_variants}; +use anyhow::{Context, Result}; +use rusqlite::{Connection, OptionalExtension}; +use std::collections::HashMap; + +use crate::memory::config::MemoryConfig; +use crate::memory::store::content::read::{ + LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX, LEGACY_NO_CONTENT_POINTER_REASON_PREFIX, +}; + +/// Fetch a chunk embedding for one provider/model/dimension signature — under +/// any of its spellings (see [`signature_variants`]). +/// +/// Returns `Ok(None)` when no row exists for `(chunk_id, model_signature)` — +/// absence is not an error. +/// +/// # Errors +/// Returns `Err` if the query fails, or if `embedding_from_blob` rejects +/// the stored blob (negative/zero-remainder-mismatched dim, or a blob length +/// not a multiple of 4 bytes — both indicate on-disk corruption of this row, +/// not a normal "no embedding" state). +pub fn get_chunk_embedding_for_signature( + config: &MemoryConfig, + chunk_id: &str, + model_signature: &str, +) -> Result>> { + let variants = signature_variants(model_signature); + with_connection(config, |conn| { + let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(variants.len() + 1); + params.push(&chunk_id as &dyn rusqlite::ToSql); + for variant in &variants { + params.push(variant as &dyn rusqlite::ToSql); + } + let row: Option<(Vec, i64)> = conn + .query_row( + &format!( + "SELECT vector, dim + FROM mem_tree_chunk_embeddings + WHERE chunk_id = ?1 AND model_signature {}", + signature_in_clause(variants.len(), 2) + ), + params.as_slice(), + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional()?; + match row { + None => Ok(None), + Some((bytes, dim)) => embedding_from_blob(&bytes, dim, "chunk embedding"), + } + }) +} + +/// Fetch a chunk's embedding for the active model signature (see +/// [`tree_active_signature`]). See [`get_chunk_embedding_for_signature`] for +/// the return/error contract. +pub fn get_chunk_embedding(config: &MemoryConfig, chunk_id: &str) -> Result>> { + let signature = tree_active_signature(config); + get_chunk_embedding_for_signature(config, chunk_id, &signature) +} + +/// Decode a little-endian `f32` vector `BLOB` back into `Vec`, validating +/// it against the DB's own recorded `dim` column. `label` only qualifies the +/// error message (e.g. `"chunk embedding"` vs. a future summary-embedding +/// caller). +/// +/// Always returns `Ok(Some(_))` on success — the `Option` in the return type +/// exists purely so callers can `?`-propagate this directly from inside a +/// `match row { None => Ok(None), Some(..) => embedding_from_blob(..) }` arm +/// without an extra `.map`. +/// +/// # Errors +/// Returns `Err` if `dim` is negative, if `bytes.len()` is not a multiple of +/// 4, or if the decoded float count does not equal `dim` — all three +/// indicate the stored row is internally inconsistent (corruption or a bug +/// upstream), not a normal "different embedding space" mismatch (that case is +/// handled by comparing signatures/dims *before* calling this, e.g. in +/// [`super::migrations::migrate_legacy_embeddings_to_sidecar`]). +fn embedding_from_blob(bytes: &[u8], dim: i64, label: &str) -> Result>> { + if dim < 0 { + anyhow::bail!("{label} has negative dimension {dim}"); + } + if !bytes.len().is_multiple_of(4) { + anyhow::bail!("{label} blob length {} not a multiple of 4", bytes.len()); + } + let floats: Vec = bytes + .chunks_exact(4) + .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .collect(); + if floats.len() != dim as usize { + anyhow::bail!( + "{label} dimension mismatch: dim column says {dim}, blob contains {} floats", + floats.len() + ); + } + Ok(Some(floats)) +} + +/// Whether any live chunk or summary lacks both an embedding and terminal +/// tombstone. +/// +/// Coverage counts a vector written under *any* spelling of the signature: a +/// row that only looks uncovered because the convention changed is not work, +/// and re-embedding it would spend provider quota to reproduce a vector that +/// is already on disk. +pub fn has_uncovered_reembed_work( + conn: &Connection, + model_signature: &str, +) -> rusqlite::Result { + let variants = signature_variants(model_signature); + let sig_clause = signature_in_clause(variants.len(), 1); + // Retryable skip reasons for legacy content-pointer rows; matched after the + // signature-variant params (which occupy `?1..=?variants.len()`). + let retry_idx_1 = variants.len() + 1; + let retry_idx_2 = variants.len() + 2; + let missing_pointer_retry_like = + format!("body read failed: {LEGACY_NO_CONTENT_POINTER_REASON_PREFIX}%"); + // Legacy-only: older readers emitted this for present-but-empty content + // pointers. Current readers filter those pointers before returning `Some`, + // but existing skip rows must still become retriable after the fallback fix. + let empty_pointer_retry_like = + format!("body read failed: {LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX}%"); + let mut params: Vec<&dyn rusqlite::ToSql> = variants + .iter() + .map(|variant| variant as &dyn rusqlite::ToSql) + .collect(); + params.push(&missing_pointer_retry_like as &dyn rusqlite::ToSql); + params.push(&empty_pointer_retry_like as &dyn rusqlite::ToSql); + conn.query_row( + &format!( + "SELECT EXISTS( + SELECT 1 FROM mem_tree_chunks c + WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e + WHERE e.chunk_id = c.id AND e.model_signature {sig_clause}) + AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk + WHERE sk.chunk_id = c.id AND sk.model_signature {sig_clause} + AND sk.reason NOT LIKE ?{retry_idx_1} + AND sk.reason NOT LIKE ?{retry_idx_2})) + OR EXISTS( + SELECT 1 FROM mem_tree_summaries s + WHERE s.deleted = 0 + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature {sig_clause}) + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk + WHERE sk.summary_id = s.id AND sk.model_signature {sig_clause}))" + ), + params.as_slice(), + |row| row.get(0), + ) +} + +/// Defensive cap for batched `IN (?,?,…)` reads, well below SQLite's +/// `SQLITE_MAX_VARIABLE_NUMBER` (32 766). +const MAX_EMBEDDING_BATCH: usize = 500; + +/// Batched read of chunk embeddings under a single `model_signature`. +/// +/// Returns a `HashMap>` containing only the chunks that have +/// a vector under `model_signature`. Missing chunks are simply absent (callers +/// treat that the same as a `None` from the single-row helper). +/// +/// `chunk_ids` is split into windows of at most `MAX_EMBEDDING_BATCH` so a +/// single query never approaches SQLite's bound-parameter limit; each window +/// runs as its own `SELECT ... WHERE chunk_id IN (...)` inside the same +/// [`super::connection::with_connection`] call (not separately transacted — +/// reads only). +/// +/// # Errors +/// Returns `Err` if `chunk_ids` is non-empty and any window's query +/// preparation, execution, or blob decoding (`embedding_from_blob`) fails. +/// Returns `Ok(HashMap::new())` immediately (no DB access) when `chunk_ids` +/// is empty. +pub fn get_chunk_embeddings_for_signature_batch( + config: &MemoryConfig, + chunk_ids: &[String], + model_signature: &str, +) -> Result>> { + if chunk_ids.is_empty() { + return Ok(HashMap::new()); + } + let variants = signature_variants(model_signature); + with_connection(config, |conn| { + let mut out: HashMap> = HashMap::with_capacity(chunk_ids.len()); + for window in chunk_ids.chunks(MAX_EMBEDDING_BATCH) { + let placeholders = std::iter::repeat_n("?", window.len()) + .collect::>() + .join(","); + let sql = format!( + "SELECT chunk_id, vector, dim + FROM mem_tree_chunk_embeddings + WHERE chunk_id IN ({placeholders}) + AND model_signature {sig_clause}", + sig_clause = signature_in_clause(variants.len(), window.len() + 1), + ); + let mut stmt = conn + .prepare(&sql) + .context("prepare get_chunk_embeddings_for_signature_batch")?; + let mut params: Vec<&dyn rusqlite::ToSql> = + Vec::with_capacity(window.len() + variants.len()); + for id in window { + params.push(id as &dyn rusqlite::ToSql); + } + for variant in &variants { + params.push(variant as &dyn rusqlite::ToSql); + } + let rows = stmt + .query_map(params.as_slice(), |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Vec>(1)?, + row.get::<_, i64>(2)?, + )) + }) + .context("query get_chunk_embeddings_for_signature_batch")?; + for row in rows { + let (chunk_id, bytes, dim) = row?; + if let Some(v) = embedding_from_blob(&bytes, dim, "chunk embedding")? { + out.insert(chunk_id, v); + } + } + } + Ok(out) + }) +} + +/// Batched read of chunk embeddings under the **active** model signature. See +/// [`get_chunk_embeddings_for_signature_batch`] for the batching and error +/// contract. +pub fn get_chunk_embeddings_batch( + config: &MemoryConfig, + chunk_ids: &[String], +) -> Result>> { + let signature = tree_active_signature(config); + get_chunk_embeddings_for_signature_batch(config, chunk_ids, &signature) +} diff --git a/src/memory/chunks/mod.rs b/src/memory/chunks/mod.rs index 9a659138..819459f4 100644 --- a/src/memory/chunks/mod.rs +++ b/src/memory/chunks/mod.rs @@ -12,9 +12,11 @@ //! - `semantic` — heading- and paragraph-aware chunker used to split large //! documents into LLM-context-sized pieces while preserving heading context. //! Exported as [`chunk_semantic`]. -//! - `store` / `connection` / `migrations` / `raw_refs` / `embeddings` — -//! the SQLite-backed chunk store (the `mem_tree_chunks` table plus its -//! per-model embedding sidecars and source ingest gates). +//! - `store` / `connection` / `migrations` / `raw_refs` / `embeddings` / +//! `embeddings_query` — the SQLite-backed chunk store (the `mem_tree_chunks` +//! table plus its per-model embedding sidecars and source ingest gates). +//! `embeddings` owns the writes and re-embed tombstones; `embeddings_query` +//! owns the signature-aware reads and the re-embed coverage probe. //! //! ## Differences from OpenHuman //! @@ -41,6 +43,8 @@ mod recovery; #[path = "embeddings.rs"] mod embeddings; +#[path = "embeddings_query.rs"] +mod embeddings_query; #[path = "migrations.rs"] mod migrations; #[path = "produce.rs"] @@ -90,12 +94,14 @@ pub use types::{ pub use connection::{shared_connection, with_connection}; pub use embeddings::{ clear_chunk_reembed_skipped, clear_reembed_skipped_for_signature, - clear_summary_reembed_skipped, embedding_to_blob, get_chunk_embedding, - get_chunk_embedding_for_signature, get_chunk_embeddings_batch, + clear_summary_reembed_skipped, embedding_to_blob, mark_chunk_reembed_skipped, + mark_summary_reembed_skipped, set_chunk_embedding, set_chunk_embedding_for_signature, + set_chunk_embedding_for_signature_tx, set_summary_embedding_for_signature_tx, + tree_active_signature, REEMBED_SKIP_KEY_MAX_LEN, +}; +pub use embeddings_query::{ + get_chunk_embedding, get_chunk_embedding_for_signature, get_chunk_embeddings_batch, get_chunk_embeddings_for_signature_batch, has_uncovered_reembed_work, - mark_chunk_reembed_skipped, mark_summary_reembed_skipped, set_chunk_embedding, - set_chunk_embedding_for_signature, set_chunk_embedding_for_signature_tx, - set_summary_embedding_for_signature_tx, tree_active_signature, REEMBED_SKIP_KEY_MAX_LEN, }; pub use raw_refs::{ get_chunk_content_path, get_chunk_content_pointers, get_chunk_raw_refs, From e64b28a618ec46e92926bde36e8ab5401a9ba1cb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 10 Aug 2026 17:27:25 +0300 Subject: [PATCH 8/8] fix(queue): isolate drain tests from the shared LLM gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two drain tests (drain_terminates_on_a_deferred_job, drain_tolerates_retired_kind_rows) flaked in CI under the parallel all-features suite. The worker routes LLM-bound jobs to a process-wide gate keyed by queue.llm_permits, so every test config leaving it at the default of 1 shares a single slot; the worker gate-semantics tests can briefly hold the sole permit, and the drain's job then gets deferred with "llm concurrency gate busy", ending the drain before its handler ran. Give the drain tests their own gate by raising llm_permits in their test config (gates are keyed by permit count). Assertions are unchanged — the tests now simply never contend for the shared slot. No production code touched. Co-authored-by: Medulla --- src/memory/queue/testing_tests.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/memory/queue/testing_tests.rs b/src/memory/queue/testing_tests.rs index 546c2e5d..cd1bbfb3 100644 --- a/src/memory/queue/testing_tests.rs +++ b/src/memory/queue/testing_tests.rs @@ -10,7 +10,15 @@ use tempfile::TempDir; fn test_config() -> (TempDir, MemoryConfig) { let tmp = TempDir::new().unwrap(); - let cfg = MemoryConfig::new(tmp.path()); + let mut cfg = MemoryConfig::new(tmp.path()); + // Drain tests enqueue LLM-bound jobs (ExtractChunk / ReembedBackfill / + // Seal). `run_once` routes them to a process-wide gate keyed by + // `queue.llm_permits`, so the default of `1` would share a single slot + // with the worker gate-semantics tests, which can briefly hold the sole + // permit and defer this test's job (`llm concurrency gate busy`), ending + // the drain early. Use a distinct count so these tests get their own + // gate and never contend for the shared slot. + cfg.queue.llm_permits = crate::memory::queue::DEFAULT_LLM_PERMITS.saturating_mul(16); (tmp, cfg) }