admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(memory): W3 first sub-store flip — chunks raw_refs delegates to TinyCortex (#4538)

Browse Source
This commit is contained in:
Steven Enamakel
2026-07-04 22:47:59 -07:00
committed by GitHub
parent f0290a6d6d
commit 5a44150e25
1 changed files with 37 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/memory_store/chunks/raw_refs.rs
+37 -162
View File
@@ -4,206 +4,81 @@
//! archives under `<content_root>/raw/` while storing only a ≤500-char
//! preview in the SQLite `content` column. Retrieval reads the archive
//! directly instead of going through the SQL preview path.
//!
//! **W3 sub-store flip:** these operations now delegate to
//! [`tinycortex::memory::chunks`] (ported from this exact module — identical SQL
//! against the same `mem_tree_chunks` / `mem_tree_summaries` tables in the shared
//! `chunks.db` the crate now owns). The host signatures are preserved so the ~4
//! external callers (`content::read`, memory_sync gmail/slack ingest, rebuild)
//! are untouched; only `&Config` is mapped to the crate's `MemoryConfig`.
use anyhow::{Context, Result};
use rusqlite::{params, OptionalExtension, Transaction};
use anyhow::Result;
use rusqlite::Transaction;
use super::with_connection;
use crate::openhuman::config::Config;
use crate::openhuman::tinycortex::memory_config_from;
/// One pointer into the raw archive. A chunk's body is reconstructed by
/// reading each [`RawRef`] in order and joining with `"\n\n"`.
///
/// `start` / `end` are byte offsets into the raw `.md` file. `end =
/// None` means "read to end of file". Both default to "the whole
/// file" (`start = 0`, `end = None`) for the common one-message-one-chunk
/// path; oversize-message chunks get explicit ranges so each chunk
/// reconstructs its sub-slice.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct RawRef {
/// Forward-slash relative path under `<content_root>/`,
/// e.g. `"raw/gmail-stevent95-at-gmail-dot-com/1700000_msg-id.md"`.
pub path: String,
#[serde(default)]
pub start: usize,
#[serde(default)]
pub end: Option<usize>,
// `RawRef` is re-exported from the crate (identical fields + serde derives), so
// every `chunks::RawRef { path, start, end }` construction site keeps compiling.
pub use tinycortex::memory::chunks::RawRef;
/// Map the host `Config` to the engine `MemoryConfig` addressing the same
/// `<workspace_dir>/memory_tree/chunks.db` (only `workspace` is load-bearing for
/// these DB ops).
fn engine_config(config: &Config) -> tinycortex::memory::MemoryConfig {
memory_config_from(config, config.workspace_dir.clone())
}
/// Stash a list of [`RawRef`] entries on a chunk row. Replaces any
/// previous value. Used by ingest pipelines that mirror their bytes
/// into `<content_root>/raw/...` so reads can skip the SQL preview
/// path and pull the full body straight from the archive.
/// Stash a list of [`RawRef`] entries on a chunk row. Replaces any previous
/// value.
pub fn set_chunk_raw_refs(config: &Config, chunk_id: &str, refs: &[RawRef]) -> Result<()> {
let json = serde_json::to_string(refs).context("serialize raw_refs")?;
with_connection(config, |conn| {
conn.execute(
"UPDATE mem_tree_chunks SET raw_refs_json = ?1 WHERE id = ?2",
params![json, chunk_id],
)?;
Ok(())
})
tinycortex::memory::chunks::set_chunk_raw_refs(&engine_config(config), chunk_id, refs)
}
/// Stash raw archive pointers on a chunk row inside a caller-owned
/// transaction. Used by ingest producers so raw-backed chunks are readable
/// before their `extract_chunk` jobs are committed.
/// Stash raw archive pointers on a chunk row inside a caller-owned transaction.
pub fn set_chunk_raw_refs_tx(tx: &Transaction<'_>, chunk_id: &str, refs: &[RawRef]) -> Result<()> {
let json = serde_json::to_string(refs).context("serialize raw_refs")?;
tx.execute(
"UPDATE mem_tree_chunks SET raw_refs_json = ?1 WHERE id = ?2",
params![json, chunk_id],
)?;
Ok(())
tinycortex::memory::chunks::set_chunk_raw_refs_tx(tx, chunk_id, refs)
}
/// Return the raw-archive pointers stored in SQLite for `chunk_id`,
/// or `None` if no `raw_refs_json` was recorded.
/// Return the raw-archive pointers stored in SQLite for `chunk_id`, or `None`.
pub fn get_chunk_raw_refs(config: &Config, chunk_id: &str) -> Result<Option<Vec<RawRef>>> {
with_connection(config, |conn| {
let row = conn
.query_row(
"SELECT raw_refs_json FROM mem_tree_chunks WHERE id = ?1",
params![chunk_id],
|r| r.get::<_, Option<String>>(0),
)
.optional()?
.flatten();
match row {
Some(json) if !json.is_empty() => {
let refs: Vec<RawRef> =
serde_json::from_str(&json).context("deserialize raw_refs_json")?;
Ok(Some(refs))
}
_ => Ok(None),
}
})
tinycortex::memory::chunks::get_chunk_raw_refs(&engine_config(config), chunk_id)
}
/// Collect every raw-archive path referenced by ANY chunk row whose
/// `raw_refs_json` is set, restricted to paths under `rel_prefix` (e.g.
/// `"raw/gmail-acct/"`). Rust-side prefix filter so `_` / `%` in slugs
/// are treated literally.
///
/// Used by the raw-coverage reconcile: a raw file referenced by a
/// persisted chunk is already ingested through the chunk pipeline and
/// must not be re-summarised by the rebuild path.
/// Collect every raw-archive path referenced by any chunk row, restricted to
/// paths under `rel_prefix`.
pub fn list_chunk_raw_ref_paths_with_prefix(
config: &Config,
rel_prefix: &str,
) -> Result<std::collections::HashSet<String>> {
with_connection(config, |conn| {
let mut stmt = conn.prepare(
"SELECT raw_refs_json FROM mem_tree_chunks \
WHERE raw_refs_json IS NOT NULL AND raw_refs_json != ''",
)?;
let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
let mut out: std::collections::HashSet<String> = std::collections::HashSet::new();
for row in rows {
let json = row?;
// Tolerate individually-corrupt rows: skip with a warning
// rather than failing the whole coverage scan.
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);
}
}
}
Err(e) => {
log::warn!(
"[memory::chunk_store] skipping unparseable raw_refs_json during \
coverage scan: {e}"
);
}
}
}
Ok(out)
})
tinycortex::memory::chunks::list_chunk_raw_ref_paths_with_prefix(
&engine_config(config),
rel_prefix,
)
}
/// Return both `content_path` and `content_sha256` stored in SQLite for `chunk_id`.
///
/// Returns `Ok(None)` if the chunk does not exist or has no content_path recorded yet.
pub fn get_chunk_content_pointers(
config: &Config,
chunk_id: &str,
) -> Result<Option<(String, String)>> {
with_connection(config, |conn| {
let row = conn
.query_row(
"SELECT content_path, content_sha256 FROM mem_tree_chunks WHERE id = ?1",
params![chunk_id],
|r| {
let path: Option<String> = r.get(0)?;
let sha: Option<String> = r.get(1)?;
Ok((path, sha))
},
)
.optional()?;
Ok(row.and_then(|(p, s)| p.zip(s)))
})
tinycortex::memory::chunks::get_chunk_content_pointers(&engine_config(config), chunk_id)
}
/// Return the `content_path` stored in SQLite for `chunk_id`, if any.
pub fn get_chunk_content_path(config: &Config, chunk_id: &str) -> Result<Option<String>> {
with_connection(config, |conn| {
let row = conn
.query_row(
"SELECT content_path FROM mem_tree_chunks WHERE id = ?1",
params![chunk_id],
|r| r.get::<_, Option<String>>(0),
)
.optional()?
.flatten();
Ok(row)
})
tinycortex::memory::chunks::get_chunk_content_path(&engine_config(config), chunk_id)
}
/// Return both `content_path` and `content_sha256` stored in SQLite for `summary_id`.
///
/// Returns `Ok(None)` if the summary does not exist or has no content_path recorded yet
/// (legacy rows pre-MD-content migration).
pub fn get_summary_content_pointers(
config: &Config,
summary_id: &str,
) -> Result<Option<(String, String)>> {
with_connection(config, |conn| {
let row = conn
.query_row(
"SELECT content_path, content_sha256 FROM mem_tree_summaries WHERE id = ?1",
params![summary_id],
|r| {
let path: Option<String> = r.get(0)?;
let sha: Option<String> = r.get(1)?;
Ok((path, sha))
},
)
.optional()?;
Ok(row.and_then(|(p, s)| p.zip(s)))
})
tinycortex::memory::chunks::get_summary_content_pointers(&engine_config(config), summary_id)
}
/// List all summary rows that have a non-NULL `content_path`. Used by the
/// bin integrity checker.
/// List all summary rows that have a non-NULL `content_path`.
pub fn list_summaries_with_content_path(config: &Config) -> Result<Vec<(String, String, String)>> {
with_connection(config, |conn| {
let mut stmt = conn.prepare(
"SELECT id, content_path, content_sha256
FROM mem_tree_summaries
WHERE content_path IS NOT NULL AND content_sha256 IS NOT NULL
AND deleted = 0",
)?;
let rows = stmt
.query_map([], |r| {
let id: String = r.get(0)?;
let path: String = r.get(1)?;
let sha: String = r.get(2)?;
Ok((id, path, sha))
})?
.collect::<rusqlite::Result<Vec<_>>>()
.context("Failed to list summaries with content_path")?;
Ok(rows)
})
tinycortex::memory::chunks::list_summaries_with_content_path(&engine_config(config))
}