Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 17 additions & 15 deletions docs/openhuman-memory-migration.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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.
6 changes: 4 additions & 2 deletions docs/openhuman-memory/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
6 changes: 4 additions & 2 deletions docs/openhuman-memory/sources-registry-sync.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
9 changes: 5 additions & 4 deletions docs/plan/05-openhuman-compat-matrix.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.**

---

Expand Down
9 changes: 8 additions & 1 deletion src/memory/chunks/embeddings.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
}

Expand Down Expand Up @@ -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
Expand Down
116 changes: 113 additions & 3 deletions src/memory/chunks/store_embed_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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() {
Comment thread
greptile-apps[bot] marked this conversation as resolved.
let (_tmp, cfg) = test_config();
Expand Down
22 changes: 17 additions & 5 deletions src/memory/store/content/read.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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)?;
Expand All @@ -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<String> {
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<String> {
let root = content_root(config);
let mut parts = Vec::with_capacity(refs.len());
Expand Down
26 changes: 26 additions & 0 deletions src/memory/store/content/read_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down