diff --git a/docs/openhuman-memory-migration.md b/docs/openhuman-memory-migration.md index 7c076a7..7c656d3 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 4dd0c7b..bcef5d2 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 d18e1b7..f807fd0 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 59dfe75..933345d 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/embeddings.rs b/src/memory/chunks/embeddings.rs index 0e07e25..ff04b42 100644 --- a/src/memory/chunks/embeddings.rs +++ b/src/memory/chunks/embeddings.rs @@ -67,6 +67,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(()) } @@ -385,7 +390,9 @@ pub fn has_uncovered_reembed_work( 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 '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 4d364a8..a25ba4f 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,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: 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 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(); + 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 fa3babf..9f91cdb 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!("legacy chunk content empty 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 b0b850d..332d432 100644 --- a/src/memory/store/content/read_tests.rs +++ b/src/memory/store/content/read_tests.rs @@ -199,6 +199,32 @@ 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_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();