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

@@ -3,14 +3,15 @@
 //! 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;
 
@@ -28,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,
@@ -77,6 +78,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(())
 }
 
@@ -302,217 +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<Option<Vec<f32>>> {
-    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<u8>, 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<Option<Vec<f32>>> {
-    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<u8> {
     embedding.iter().flat_map(|f| f.to_le_bytes()).collect()
 }
-
-/// Decode a little-endian `f32` vector `BLOB` back into `Vec<f32>`, 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<Option<Vec<f32>>> {
-    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<f32> = 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<bool> {
-    let variants = signature_variants(model_signature);
-    let sig_clause = signature_in_clause(variants.len(), 1);
-    let params: Vec<&dyn rusqlite::ToSql> = variants
-        .iter()
-        .map(|variant| variant as &dyn rusqlite::ToSql)
-        .collect();
-    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}))
-               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<chunk_id, Vec<f32>>` 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<HashMap<String, Vec<f32>>> {
-    if chunk_ids.is_empty() {
-        return Ok(HashMap::new());
-    }
-    let variants = signature_variants(model_signature);
-    with_connection(config, |conn| {
-        let mut out: HashMap<String, Vec<f32>> = HashMap::with_capacity(chunk_ids.len());
-        for window in chunk_ids.chunks(MAX_EMBEDDING_BATCH) {
-            let placeholders = std::iter::repeat_n("?", window.len())
-                .collect::<Vec<_>>()
-                .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<u8>>(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<HashMap<String, Vec<f32>>> {
-    let signature = tree_active_signature(config);
-    get_chunk_embeddings_for_signature_batch(config, chunk_ids, &signature)
-}