feat(memory): port OpenHuman engine gaps

Cargo.toml

@@ -39,23 +39,31 @@ git-diff = ["dep:git2"]
 # gates the dependency.
 providers-http = ["dep:reqwest", "tokio"]
 
+# Live source synchronization (Composio HTTP pipelines and workspace scans).
+# Host schedulers, credentials, RPC, and event buses remain outside the crate.
+sync = ["dep:reqwest", "dep:tracing", "tokio"]
+
 # serde schema / envelope surface for the RPC boundary (`memory::rpc`). Reserved
 # for goal C5; gates the wire-facing surface without adding heavy dependencies
 # (serde is already a core dependency).
 rpc = []
 
 [dependencies]
 anyhow = "1"
+log = "0.4"
+futures = "0.3"
 async-trait = "0.1"
 chrono = { version = "0.4", features = ["serde"] }
 parking_lot = "0.12"
 rand = "0.10"
 regex = "1"
-rusqlite = { version = "0.39", features = ["bundled"] }
+rusqlite = { version = "0.40", features = ["bundled"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+schemars = "1"
 sha2 = "0.10"
 thiserror = "2"
+tinyagents = { git = "https://github.com/senamakel/tinyagents", rev = "8f75355" }
 toml = "0.8"
 uuid = { version = "1", features = ["serde", "v4"] }
 walkdir = "2"
@@ -73,6 +81,7 @@ reqwest = { version = "0.12", default-features = false, features = [
     "json",
     "rustls-tls",
 ], optional = true }
+tracing = { version = "0.1", optional = true }
 # tokio powers the optional background worker loops (`tokio` feature). It is
 # also listed under dev-dependencies so the async test suite always compiles.
 tokio = { version = "1", features = [
@@ -86,3 +95,12 @@ tokio = { version = "1", features = [
 [dev-dependencies]
 tempfile = "3"
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] }
+wiremock = "0.6"
+
+[[test]]
+name = "composio_sync_mock"
+required-features = ["sync"]
+
+[[test]]
+name = "composio_sync_live"
+required-features = ["sync"]

src/memory/chunks/connection.rs

@@ -219,7 +219,7 @@ fn apply_schema(conn: &Connection) -> Result<()> {
     // currently a no-op — a refusal is silently accepted here, and the
     // synchronous=FULL crash-safety assumption in `init_db` is only actually
     // valid when the mode really did become TRUNCATE. See audit finding SC-9.
-    if !journal_mode.eq_ignore_ascii_case("truncate") {}
+    let _ = journal_mode.eq_ignore_ascii_case("truncate");
     conn.execute_batch(SCHEMA)
         .context("Failed to initialize chunk DB schema")?;
     // Additive, idempotent migrations.
@@ -370,7 +370,7 @@ pub(crate) fn get_or_init_connection(config: &MemoryConfig) -> Result<Arc<PMutex
             // previously-tripped breaker (recovery signal); currently
             // discarded, so a breaker recovering from an open state is not
             // logged anywhere. See audit finding SC-9.
-            if breaker.record_success() {}
+            breaker.record_success();
             Ok(arc_conn)
         }
         Err(err) => {
@@ -384,7 +384,7 @@ pub(crate) fn get_or_init_connection(config: &MemoryConfig) -> Result<Arc<PMutex
             // NOTE: `record_failure` returns whether this call just tripped
             // the breaker; currently discarded, so the trip event itself is
             // not logged, only observable later via `is_open`. See SC-9.
-            if breaker.record_failure() {}
+            breaker.record_failure();
             Err(err)
         }
     }
@@ -445,13 +445,14 @@ pub(super) fn drop_cached_connection(config: &MemoryConfig) {
     conn_cache().breakers.lock().remove(&db_path);
 }
 
-/// Clear the entire connection cache (all workspace paths' connections,
-/// breakers, and init locks). For test isolation only — production code must
-/// never need to reset every path at once.
+/// Clear cached connections and init locks for test isolation.
+///
+/// Breakers are deliberately retained: tests use unique temporary workspace
+/// paths, and clearing the process-wide breaker map races with parallel tests
+/// that are proving threshold behavior for another path.
 #[cfg(test)]
 pub(crate) fn clear_connection_cache() {
     conn_cache().connections.lock().clear();
-    conn_cache().breakers.lock().clear();
     conn_cache().init_locks.lock().clear();
 }
 
@@ -482,6 +483,16 @@ pub fn with_connection<T>(
     f(&guard)
 }
 
+/// Return the initialized connection shared by chunk, tree, and auxiliary
+/// stores for this workspace.
+///
+/// Embedding applications may pass this handle to shared-connection stores
+/// such as `KvStore` and `EntityIndex`. Callers must not change connection
+/// pragmas or hold the mutex across an await point.
+pub fn shared_connection(config: &MemoryConfig) -> Result<Arc<PMutex<Connection>>> {
+    get_or_init_connection(config)
+}
+
 #[cfg(test)]
 #[path = "connection_tests.rs"]
 mod tests;

src/memory/chunks/embeddings.rs

@@ -120,7 +120,7 @@ pub fn set_chunk_embedding_for_signature(
 ///
 /// # Errors
 /// See [`upsert_chunk_embedding_conn`].
-pub(crate) fn set_chunk_embedding_for_signature_tx(
+pub fn set_chunk_embedding_for_signature_tx(
     tx: &rusqlite::Transaction<'_>,
     chunk_id: &str,
     model_signature: &str,
@@ -133,7 +133,7 @@ pub(crate) fn set_chunk_embedding_for_signature_tx(
 ///
 /// # Errors
 /// See [`upsert_summary_embedding_conn`].
-pub(crate) fn set_summary_embedding_for_signature_tx(
+pub fn set_summary_embedding_for_signature_tx(
     tx: &rusqlite::Transaction<'_>,
     summary_id: &str,
     model_signature: &str,
@@ -199,6 +199,49 @@ pub fn clear_chunk_reembed_skipped(
     })
 }
 
+pub fn mark_summary_reembed_skipped(
+    config: &MemoryConfig,
+    summary_id: &str,
+    model_signature: &str,
+    reason: &str,
+) -> Result<()> {
+    let summary_id = validate_reembed_skip_key("summary_id", summary_id)?;
+    let model_signature = validate_reembed_skip_key("model_signature", model_signature)?;
+    with_connection(config, |conn| {
+        conn.execute(
+            "INSERT INTO mem_tree_summary_reembed_skipped
+             (summary_id, model_signature, reason, skipped_at_ms)
+             VALUES (?1, ?2, ?3, ?4)
+             ON CONFLICT(summary_id, model_signature) DO UPDATE SET
+                reason = excluded.reason, skipped_at_ms = excluded.skipped_at_ms",
+            rusqlite::params![
+                summary_id,
+                model_signature,
+                reason,
+                Utc::now().timestamp_millis()
+            ],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn clear_summary_reembed_skipped(
+    config: &MemoryConfig,
+    summary_id: &str,
+    model_signature: &str,
+) -> Result<()> {
+    let summary_id = validate_reembed_skip_key("summary_id", summary_id)?;
+    let model_signature = validate_reembed_skip_key("model_signature", model_signature)?;
+    with_connection(config, |conn| {
+        conn.execute(
+            "DELETE FROM mem_tree_summary_reembed_skipped
+             WHERE summary_id = ?1 AND model_signature = ?2",
+            rusqlite::params![summary_id, model_signature],
+        )?;
+        Ok(())
+    })
+}
+
 /// Clear all chunk and summary tombstones for a model signature. Returns the
 /// total number of rows removed across both tombstone tables. Idempotent —
 /// calling again with nothing left to clear returns `Ok(0)`.
@@ -226,7 +269,7 @@ pub fn clear_reembed_skipped_for_signature(
 
 /// Bounds attacker-controlled ids/signatures passed to reembed-skipped admin
 /// helpers. Rejects NUL bytes so SQLite bindings cannot be truncated.
-pub(crate) const REEMBED_SKIP_KEY_MAX_LEN: usize = 2048;
+pub const REEMBED_SKIP_KEY_MAX_LEN: usize = 2048;
 
 /// Validate and trim a `chunk_id` / `model_signature` value before it reaches
 /// a reembed-skipped table query. `label` is only used to make the error
@@ -290,7 +333,7 @@ pub fn get_chunk_embedding(config: &MemoryConfig, chunk_id: &str) -> Result<Opti
 }
 
 /// Little-endian `f32` vector → `BLOB`. The inverse of `embedding_from_blob`.
-pub(crate) fn embedding_to_blob(embedding: &[f32]) -> Vec<u8> {
+pub fn embedding_to_blob(embedding: &[f32]) -> Vec<u8> {
     embedding.iter().flat_map(|f| f.to_le_bytes()).collect()
 }
 
@@ -331,6 +374,30 @@ fn embedding_from_blob(bytes: &[u8], dim: i64, label: &str) -> Result<Option<Vec
     Ok(Some(floats))
 }
 
+/// Whether any live chunk or summary lacks both an embedding and terminal tombstone.
+pub fn has_uncovered_reembed_work(
+    conn: &Connection,
+    model_signature: &str,
+) -> rusqlite::Result<bool> {
+    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))
+           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],
+        |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;

src/memory/chunks/mod.rs

@@ -4,15 +4,14 @@
 //! OpenHuman's `memory_store/chunks`:
 //!
 //! - [`types`]    — [`Chunk`], [`Metadata`], [`SourceKind`], [`DataSource`],
-//!                  [`SourceRef`], [`StagedChunk`] and the deterministic
-//!                  [`chunk_id`] / token-estimate functions. The persisted
-//!                  shape.
+//!   [`SourceRef`], [`StagedChunk`] and the deterministic
+//!   [`chunk_id`] / token-estimate functions. The persisted shape.
 //! - [`produce`]  — source-kind-dispatch chunker (chat / email / document).
-//!                  Splits canonical Markdown into bounded chunks with stable
-//!                  per-source sequence numbers.
+//!   Splits canonical Markdown into bounded chunks with stable per-source
+//!   sequence numbers.
 //! - [`semantic`] — heading- and paragraph-aware chunker used to split large
-//!                  documents into LLM-context-sized pieces while preserving
-//!                  heading context. Exported as [`chunk_semantic`].
+//!   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).
@@ -80,29 +79,39 @@ pub use types::{
     Chunk, DataSource, Metadata, SourceKind, SourceRef, StagedChunk,
 };
 
-pub use connection::with_connection;
+pub use connection::{shared_connection, with_connection};
 pub use embeddings::{
-    clear_chunk_reembed_skipped, clear_reembed_skipped_for_signature, get_chunk_embedding,
+    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,
-    get_chunk_embeddings_for_signature_batch, mark_chunk_reembed_skipped, set_chunk_embedding,
-    set_chunk_embedding_for_signature, tree_active_signature,
+    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,
     get_summary_content_pointers, list_chunk_raw_ref_paths_with_prefix,
     list_summaries_with_content_path, set_chunk_raw_refs, set_chunk_raw_refs_tx, RawRef,
 };
+pub use recovery::{
+    is_io_open_error, is_transient_cold_start, recover_corrupt_db, try_cleanup_stale_files,
+};
 pub use store::{
     claim_source_ingest_tx, count_chunks, count_chunks_by_lifecycle_status,
     count_raw_paths_ingested_with_prefix, delete_source_ingest, extraction_coverage,
     filter_raw_paths_not_ingested, get_chunk, get_chunk_lifecycle_status, get_chunks_batch,
-    is_source_ingested, list_chunks, mark_raw_paths_ingested, set_chunk_lifecycle_status,
-    upsert_chunks, ListChunksQuery, CHUNK_STATUS_ADMITTED, CHUNK_STATUS_BUFFERED,
-    CHUNK_STATUS_DROPPED, CHUNK_STATUS_PENDING_EXTRACTION, CHUNK_STATUS_SEALED, RAW_FILE_GATE_KIND,
+    is_source_ingested, list_chunks, list_source_ids_with_prefix, mark_raw_paths_ingested,
+    set_chunk_lifecycle_status, update_chunk_content_sha256, update_summary_content_sha256,
+    upsert_chunks, upsert_chunks_tx, upsert_staged_chunks_tx, ListChunksQuery,
+    CHUNK_STATUS_ADMITTED, CHUNK_STATUS_BUFFERED, CHUNK_STATUS_DROPPED,
+    CHUNK_STATUS_PENDING_EXTRACTION, CHUNK_STATUS_SEALED, RAW_FILE_GATE_KIND,
 };
 pub use store_delete::{
     delete_chunks_by_owner, delete_chunks_by_source, delete_chunks_by_source_prefix,
+    delete_orphaned_source_tree,
 };
+pub use store_sources::{get_chunk_lifecycle_status_tx, set_chunk_lifecycle_status_tx};
 
 // ── Shared internal constants / helpers ─────────────────────────────────────
 
@@ -128,7 +137,10 @@ pub(crate) fn db_path_for(config: &MemoryConfig) -> PathBuf {
 /// Root directory for on-disk chunk/summary content files associated with the
 /// chunk DB. Bodies that are too large for the SQLite preview column live here.
 pub(crate) fn content_root(config: &MemoryConfig) -> PathBuf {
-    config.workspace.join(DB_DIR).join("content")
+    config
+        .content_root
+        .clone()
+        .unwrap_or_else(|| config.workspace.join(DB_DIR).join("content"))
 }
 
 /// Redact a PII-bearing string for log output by hashing it to 8 hex chars.

src/memory/chunks/produce.rs

@@ -269,13 +269,13 @@ fn split_email_messages(md: &str) -> Vec<String> {
         if line == "---" {
             // Check if one of the next 8 lines starts with `From:`
             let window_end = (i + 9).min(n);
-            for j in (i + 1)..window_end {
-                if lines[j].starts_with("From:") {
+            for candidate in lines.iter().take(window_end).skip(i + 1) {
+                if candidate.starts_with("From:") {
                     split_positions.push(i);
                     break;
                 }
                 // Skip blank lines between `---` and `From:`
-                if !lines[j].trim().is_empty() {
+                if !candidate.trim().is_empty() {
                     break;
                 }
             }
@@ -298,7 +298,7 @@ fn split_email_messages(md: &str) -> Vec<String> {
         } else {
             n
         };
-        let piece_lines: Vec<&str> = lines[start..end].iter().copied().collect();
+        let piece_lines: Vec<&str> = lines[start..end].to_vec();
         let piece = piece_lines.join("\n").trim_end().to_string();
         if !piece.is_empty() {
             pieces.push(piece);

src/memory/chunks/produce_split.rs

@@ -128,12 +128,9 @@ fn split_on_sentences<'a>(text: &'a str, budget: u32, out: &mut Vec<&'a str>) ->
     let mut start = 0usize;
     for i in 0..bytes.len() {
         let c = bytes[i];
-        let boundary_end = if c == b'\n' {
-            Some(i + 1)
-        } else if (c == b'.' || c == b'!' || c == b'?')
-            && i + 1 < bytes.len()
-            && bytes[i + 1] == b' '
-        {
+        let sentence_end =
+            matches!(c, b'.' | b'!' | b'?') && i + 1 < bytes.len() && bytes[i + 1] == b' ';
+        let boundary_end = if c == b'\n' || sentence_end {
             Some(i + 1)
         } else {
             None

src/memory/chunks/raw_refs.rs

@@ -127,17 +127,12 @@ pub fn list_chunk_raw_ref_paths_with_prefix(
         let mut out: std::collections::HashSet<String> = std::collections::HashSet::new();
         for row in rows {
             let json = row?;
-            match serde_json::from_str::<Vec<RawRef>>(&json) {
-                Ok(refs) => {
-                    for raw_ref in refs {
-                        if raw_ref.path.starts_with(rel_prefix) {
-                            out.insert(raw_ref.path);
-                        }
+            if let Ok(refs) = serde_json::from_str::<Vec<RawRef>>(&json) {
+                for raw_ref in refs {
+                    if raw_ref.path.starts_with(rel_prefix) {
+                        out.insert(raw_ref.path);
                     }
                 }
-                // Tolerate individually-corrupt rows: skip rather than failing
-                // the whole coverage scan.
-                Err(_) => {}
             }
         }
         Ok(out)