Repair OpenHuman memory source retrieval paths

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.

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:

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.

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.**
 
 ---
 

src/memory/chunks/embeddings.rs

@@ -12,6 +12,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.
@@ -67,6 +70,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(())
 }
 
@@ -379,21 +387,34 @@ pub fn has_uncovered_reembed_work(
     conn: &Connection,
     model_signature: &str,
 ) -> rusqlite::Result<bool> {
+    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}%");
     conn.query_row(
         "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 = ?1)
                 AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk
-                                 WHERE sk.chunk_id = c.id AND sk.model_signature = ?1))
+                                 WHERE sk.chunk_id = c.id AND sk.model_signature = ?1
+                                   AND sk.reason NOT LIKE ?2
+                                   AND sk.reason NOT LIKE ?3))
            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 = ?1)
                 AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk
                                  WHERE sk.summary_id = s.id AND sk.model_signature = ?1))",
-        rusqlite::params![model_signature],
+        rusqlite::params![
+            model_signature,
+            missing_pointer_retry_like,
+            empty_pointer_retry_like
+        ],
         |row| row.get(0),
     )
 }

src/memory/chunks/store_embed_tests.rs

@@ -19,11 +19,16 @@ 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::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,
 };
@@ -81,6 +86,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 +345,93 @@ 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: {LEGACY_NO_CONTENT_POINTER_REASON_PREFIX}{}",
+            c.id
+        ),
+    )
+    .unwrap();
+
+    with_connection(&cfg, |conn| {
+        assert!(has_uncovered_reembed_work(conn, sig)?);
+        Ok(())
+    })
+    .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: {LEGACY_EMPTY_CONTENT_POINTER_REASON_PREFIX}{}",
+            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_EMPTY_CHUNK_CONTENT_REASON_PREFIX}{}",
+            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();

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,46 @@ fn dedup_ids(ids: impl Iterator<Item = String>) -> Vec<String> {
     ids.filter(|id| seen.insert(id.clone())).collect()
 }
 
+fn source_scope_allows(scope: &HashSet<String>, tree_scope: &str) -> bool {
+    if scope.contains(tree_scope) {
+        return true;
+    }
+    // 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:<source_id>:<path>` 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 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;

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();