diff --git a/src/openhuman/memory_store/chunks/store.rs b/src/openhuman/memory_store/chunks/store.rs
index 61b4d12fb..7237cd6e9 100644
@@ -1,122 +1,120 @@
//! SQLite-backed persistence for ingested chunks (Phase 1 / issue #707).
//!
//! The store lives at `<workspace>/memory_tree/chunks.db`. Schema is applied
//! lazily on first access via `with_connection`, so the DB is created on
//! demand without an explicit migration step.
//!
//! Upsert semantics: writes are idempotent on `chunk.id` so re-ingesting the
//! same raw source yields no duplicates.
//!
//! ## Connection cache (#2206)
//!
//! `with_connection()` previously opened a new SQLite connection and re-ran
//! the full schema init (8 tables, 15+ indexes, 8+ migrations) on **every**
//! call. With 4 workers polling every 5 s this amounted to ~69K connection
//! opens/day, and a family of WAL/SHM cold-start I/O codes (1546
//! IOERR_TRUNCATE, 4618 IOERR_SHMOPEN, 4874 IOERR_SHMSIZE, 14 CANTOPEN)
//! flooded Sentry with ~19K events in 4 days.
//!
//! Fix: a process-level `ConnectionCache` keyed by DB path. Each entry holds
//! one `parking_lot::Mutex<Connection>` that is initialised once (schema +
//! migrations + legacy-embedding migration) and then reused for all subsequent
//! calls. A per-entry `CircuitBreaker` stops retrying after 3 consecutive
//! init failures for 30 s so a broken install does not busy-loop.
use anyhow::{Context, Result};
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::Utc;
use rusqlite::{params, Connection, OptionalExtension, Transaction};
use std::collections::{HashMap, HashSet};
#[cfg(test)]
use std::sync::Arc;
use std::time::Duration;
use crate::openhuman::config::Config;
use crate::openhuman::memory::util::redact::{self, redact as redact_value};
-use crate::openhuman::memory_store::chunks::types::{Chunk, Metadata, SourceKind, SourceRef};
+use crate::openhuman::memory_store::chunks::types::{Chunk, SourceKind};
use crate::openhuman::memory_store::content::StagedChunk;
use crate::openhuman::tinycortex::memory_config_from;
const DB_DIR: &str = "memory_tree";
const DB_FILE: &str = "chunks.db";
-const DEFAULT_LIST_LIMIT: usize = 100;
-const MAX_LIST_LIMIT: usize = 10_000;
// 15s gives the busy-handler enough headroom that transient write-lock
// contention (4 job workers + scheduler + ingest producers all writing the
// same `memory_tree/chunks.db`) is absorbed inside rusqlite instead of
// surfacing as `SQLITE_BUSY` to callers. Workers still treat busy as a
// soft signal (see `memory_tree::jobs::worker`) so even if this is
// exceeded, the only effect is a one-poll backoff — but 15s is
// comfortably above realistic peer-write durations and shrinks the rate
// at which we have to fall back to that path. The previous 5s was tight
// enough on contended Windows hosts that we were observing avoidable
// busy returns (see OPENHUMAN-TAURI-BP).
const SQLITE_BUSY_TIMEOUT: Duration = Duration::from_secs(15);
/// Chunk lifecycle: freshly persisted, awaiting the async extract job.
pub const CHUNK_STATUS_PENDING_EXTRACTION: &str = "pending_extraction";
/// Chunk lifecycle: extract ran and the chunk passed admission.
pub const CHUNK_STATUS_ADMITTED: &str = "admitted";
/// Chunk lifecycle: appended to the L0 buffer of its source tree.
pub const CHUNK_STATUS_BUFFERED: &str = "buffered";
/// Chunk lifecycle: rolled into a sealed L1 summary.
pub const CHUNK_STATUS_SEALED: &str = "sealed";
/// Chunk lifecycle: rejected by the admission gate (too low signal).
pub const CHUNK_STATUS_DROPPED: &str = "dropped";
// `PRAGMA foreign_keys = ON` is intentionally NOT in SCHEMA — it is
// a connection-local pragma that resets to off on every new
// `Connection::open`. SCHEMA only runs once per DB path (first-init);
// applying foreign_keys here would leak FK-off into every later
// `with_connection()` call that hits the fast path. The pragma is
// set per-connection in `open_connection()` instead.
/// `PRAGMA user_version` value once the one-shot legacy→sidecar embedding
/// migration (#1574 §7) has run. `0` (fresh/legacy DB) triggers the copy on
/// next open; `>= 1` skips it. Bump only for a new one-shot data migration.
const TREE_EMBEDDING_MIGRATION_VERSION: i64 = 1;
/// `PRAGMA user_version` value once the global/topic-tree purge has run.
/// The global (time-axis) and topic (subject-axis) trees were removed; this
/// one-shot migration deletes their rows + on-disk summary folders. `< 2`
/// triggers the purge on next open; `>= 2` skips it.
const GLOBAL_TOPIC_PURGE_MIGRATION_VERSION: i64 = 2;
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS mem_tree_chunks (
id TEXT PRIMARY KEY,
source_kind TEXT NOT NULL,
source_id TEXT NOT NULL,
path_scope TEXT,
source_ref TEXT,
owner TEXT NOT NULL,
timestamp_ms INTEGER NOT NULL,
time_range_start_ms INTEGER NOT NULL,
time_range_end_ms INTEGER NOT NULL,
tags_json TEXT NOT NULL DEFAULT '[]',
content TEXT NOT NULL,
token_count INTEGER NOT NULL,
seq_in_source INTEGER NOT NULL,
created_at_ms INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_source
ON mem_tree_chunks(source_kind, source_id);
CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_timestamp
ON mem_tree_chunks(timestamp_ms);
CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_owner
ON mem_tree_chunks(owner);
CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_source_seq
ON mem_tree_chunks(source_kind, source_id, seq_in_source);
-- Per-(chunk, embedding model) vectors (#1574). The legacy
-- mem_tree_chunks.embedding column remains in place during the dual-write
-- migration; this table lets multiple vector spaces coexist safely.
CREATE TABLE IF NOT EXISTS mem_tree_chunk_embeddings (
chunk_id TEXT NOT NULL REFERENCES mem_tree_chunks(id) ON DELETE CASCADE,
model_signature TEXT NOT NULL,
vector BLOB NOT NULL,
dim INTEGER NOT NULL,
created_at REAL NOT NULL,
PRIMARY KEY (chunk_id, model_signature)
);
@@ -492,334 +490,177 @@ pub(crate) fn upsert_staged_chunks_tx(
path_scope = excluded.path_scope,
source_ref = excluded.source_ref,
owner = excluded.owner,
timestamp_ms = excluded.timestamp_ms,
time_range_start_ms = excluded.time_range_start_ms,
time_range_end_ms = excluded.time_range_end_ms,
tags_json = excluded.tags_json,
content = excluded.content,
token_count = excluded.token_count,
seq_in_source = excluded.seq_in_source,
created_at_ms = excluded.created_at_ms,
content_path = excluded.content_path,
content_sha256 = excluded.content_sha256",
)?;
for s in staged {
let chunk = &s.chunk;
// SQL `content` column always carries a ≤500-char preview now
// — the full body either lives at `content_path` (chat /
// document) or is reconstructed from `raw_refs_json` byte
// ranges in the raw archive (email). See `read_chunk_body`.
let preview: String = chunk.content.chars().take(500).collect();
stmt.execute(params![
chunk.id,
chunk.metadata.source_kind.as_str(),
chunk.metadata.source_id,
chunk.metadata.path_scope,
chunk.metadata.source_ref.as_ref().map(|r| r.value.as_str()),
chunk.metadata.owner,
chunk.metadata.timestamp.timestamp_millis(),
chunk.metadata.time_range.0.timestamp_millis(),
chunk.metadata.time_range.1.timestamp_millis(),
serde_json::to_string(&chunk.metadata.tags)?,
preview,
chunk.token_count,
chunk.seq_in_source,
chunk.created_at.timestamp_millis(),
s.content_path,
s.content_sha256,
])?;
}
Ok(staged.len())
}
fn upsert_chunks_with_statement(
stmt: &mut rusqlite::Statement<'_>,
chunks: &[Chunk],
) -> Result<()> {
for chunk in chunks {
stmt.execute(params![
chunk.id,
chunk.metadata.source_kind.as_str(),
chunk.metadata.source_id,
chunk.metadata.path_scope,
chunk.metadata.source_ref.as_ref().map(|r| r.value.as_str()),
chunk.metadata.owner,
chunk.metadata.timestamp.timestamp_millis(),
chunk.metadata.time_range.0.timestamp_millis(),
chunk.metadata.time_range.1.timestamp_millis(),
serde_json::to_string(&chunk.metadata.tags)?,
chunk.content,
chunk.token_count,
chunk.seq_in_source,
chunk.created_at.timestamp_millis(),
])?;
}
Ok(())
}
/// Fetch one chunk by its id.
/// Map the host `Config` to the engine `MemoryConfig` addressing the same
/// `<workspace_dir>/memory_tree/chunks.db` (only `workspace` is load-bearing for
/// these delegating DB reads). W3 store-op flip.
fn engine_config(config: &Config) -> tinycortex::memory::MemoryConfig {
memory_config_from(config, config.workspace_dir.clone())
}
pub fn get_chunk(config: &Config, id: &str) -> Result<Option<Chunk>> {
tinycortex::memory::chunks::get_chunk(&engine_config(config), id)
}
-/// Defensive cap for batched `IN (?,?,…)` reads.
-///
-/// SQLite's compile-time limit on bound parameters in a single statement
-/// (`SQLITE_MAX_VARIABLE_NUMBER`) has been **32 766** since 3.32 (2020),
-/// so 500 leaves a ~65× safety margin. The current call-site
-/// (`memory_tree::retrieval::fetch::fetch_leaves`) is capped at 20 ids,
-/// so the chunked loop runs exactly once today. The window exists so
-/// future call-sites passing larger id lists do not blow up against a
-/// host with a lower compile-time SQLite cap (older builds, custom
-/// embeddings, etc.).
-///
-/// Volume is **not** reduced: all input ids in → all matching rows out.
-/// The loop only splits the SQL; the merged `HashMap` is byte-identical
-/// to what one giant query would return.
-const MAX_FETCH_BATCH: usize = 500;
-
-/// Batched read of full chunk rows by id.
-///
-/// Contract mirror of looping [`get_chunk`] per id, but in
-/// `O(ceil(n / MAX_FETCH_BATCH))` SQLite round-trips instead of `O(n)`.
-/// The returned map contains only ids that exist in `mem_tree_chunks`;
-/// missing ids are silently absent (same as `get_chunk` returning
-/// `Ok(None)`). Callers that depend on input order must iterate their
-/// own id slice and look each id up in the map.
-///
-/// Reuses [`row_to_chunk`] so decoding stays bit-identical to the
-/// per-row helper — no risk of decoder drift.
+/// Batched read of full chunk rows by id — delegates to the crate.
pub fn get_chunks_batch(config: &Config, chunk_ids: &[String]) -> Result<HashMap<String, Chunk>> {
- if chunk_ids.is_empty() {
- return Ok(HashMap::new());
- }
- log::debug!(
- "[memory::chunk_store] get_chunks_batch: n={} windows={}",
- chunk_ids.len(),
- chunk_ids.len().div_ceil(MAX_FETCH_BATCH)
- );
- with_connection(config, |conn| {
- let mut out: HashMap<String, Chunk> = HashMap::with_capacity(chunk_ids.len());
- for window in chunk_ids.chunks(MAX_FETCH_BATCH) {
- // Build the placeholder list `?1, ?2, …, ?n` matching the
- // window length; rusqlite assigns positional binds 1..n in
- // the order the values are passed.
- let placeholders = (1..=window.len())
- .map(|i| format!("?{i}"))
- .collect::<Vec<_>>()
- .join(",");
- let sql = format!(
- "SELECT id, source_kind, source_id, path_scope, source_ref, owner,
- timestamp_ms, time_range_start_ms, time_range_end_ms,
- tags_json, content, token_count, seq_in_source, created_at_ms
- FROM mem_tree_chunks WHERE id IN ({placeholders})"
- );
- let mut stmt = conn.prepare(&sql).context("prepare get_chunks_batch")?;
- let params: Vec<&dyn rusqlite::ToSql> =
- window.iter().map(|id| id as &dyn rusqlite::ToSql).collect();
- let rows = stmt
- .query_map(params.as_slice(), row_to_chunk)
- .context("query get_chunks_batch")?;
- for row in rows {
- let chunk = row.context("decode get_chunks_batch row")?;
- out.insert(chunk.id.clone(), chunk);
- }
- }
- log::debug!(
- "[memory::chunk_store] get_chunks_batch: matched {}/{} ids",
- out.len(),
- chunk_ids.len()
- );
- Ok(out)
- })
+ tinycortex::memory::chunks::get_chunks_batch(&engine_config(config), chunk_ids)
}
-/// Query parameters for [`list_chunks`]. All fields are optional filters —
-/// callers pass `ListChunksQuery::default()` to get recent-across-everything.
-#[derive(Debug, Default, Clone)]
-pub struct ListChunksQuery {
- pub source_kind: Option<SourceKind>,
- pub source_id: Option<String>,
- pub owner: Option<String>,
- /// Inclusive lower bound on `timestamp` (milliseconds since epoch).
- pub since_ms: Option<i64>,
- /// Inclusive upper bound on `timestamp` (milliseconds since epoch).
- pub until_ms: Option<i64>,
- /// Max rows to return (default 100 when `None`).
- pub limit: Option<usize>,
- /// Per-profile memory-source allowlist. When `Some`, memory-source chunks
- /// (those tagged `memory_sources`) whose source identifier is not in the set
- /// are dropped *before* the row limit is applied, so a disallowed-source
- /// prefix can't starve permitted rows. Non-source chunks always pass. `None`
- /// = unrestricted (the default for every non-agent caller).
- pub source_scope: Option<std::collections::HashSet<String>>,
- /// When `true`, rows the admission gate rejected (`lifecycle_status =
- /// 'dropped'`) are excluded. Default `false` preserves the all-rows
- /// behaviour every existing caller relies on; retrieval paths that must not
- /// surface filtered-out junk (e.g. `cover_window`) opt in.
- pub exclude_dropped: bool,
-}
+/// Query parameters for [`list_chunks`], re-exported from the crate (identical
+/// fields incl. the `source_scope` allowlist + `exclude_dropped`).
+pub use tinycortex::memory::chunks::ListChunksQuery;
/// List chunks matching the provided filters, ordered by `timestamp` DESC.
+///
+/// Delegates to the crate, which preserves the `source_scope` allowlist gate
+/// (byte-identical `chunk_source_allowed_in` + `extract_mem_src_id`) — the
+/// security-critical per-profile enforcement, pinned by
+/// `store_tests::list_chunks_source_scope_filters_before_limit`.
pub fn list_chunks(config: &Config, query: &ListChunksQuery) -> Result<Vec<Chunk>> {
- with_connection(config, |conn| {
- let mut sql = String::from(
- "SELECT id, source_kind, source_id, path_scope, source_ref, owner,
- timestamp_ms, time_range_start_ms, time_range_end_ms,
- tags_json, content, token_count, seq_in_source, created_at_ms
- FROM mem_tree_chunks WHERE 1=1",
- );
- let mut bound: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
-
- if let Some(kind) = query.source_kind {
- sql.push_str(" AND source_kind = ?");
- bound.push(Box::new(kind.as_str().to_string()));
- }
- if let Some(ref source_id) = query.source_id {
- sql.push_str(" AND source_id = ?");
- bound.push(Box::new(source_id.clone()));
- }
- if let Some(ref owner) = query.owner {
- sql.push_str(" AND owner = ?");
- bound.push(Box::new(owner.clone()));
- }
- if let Some(since_ms) = query.since_ms {
- sql.push_str(" AND timestamp_ms >= ?");
- bound.push(Box::new(since_ms));
- }
- if let Some(until_ms) = query.until_ms {
- sql.push_str(" AND timestamp_ms <= ?");
- bound.push(Box::new(until_ms));
- }
- if query.exclude_dropped {
- sql.push_str(" AND lifecycle_status != ?");
- bound.push(Box::new(CHUNK_STATUS_DROPPED.to_string()));
- }
- let requested_limit = normalized_limit(query.limit);
- // When a profile source-scope is active, fetch a wider candidate set and
- // apply the gate in Rust *before* truncating, so a disallowed-source
- // prefix can't push permitted rows past the requested limit. Otherwise
- // the SQL LIMIT alone is correct and cheap.
- let sql_limit = if query.source_scope.is_some() {
- MAX_LIST_LIMIT as i64
- } else {
- requested_limit
- };
- sql.push_str(" ORDER BY timestamp_ms DESC, seq_in_source ASC LIMIT ?");
- bound.push(Box::new(sql_limit));
-
- let mut stmt = conn.prepare(&sql)?;
- let param_refs: Vec<&dyn rusqlite::ToSql> = bound
- .iter()
- .map(|b| b.as_ref() as &dyn rusqlite::ToSql)
- .collect();
- let mut rows = stmt
- .query_map(param_refs.as_slice(), row_to_chunk)?
- .collect::<rusqlite::Result<Vec<_>>>()
- .context("Failed to collect chunks")?;
- if let Some(ref allowed) = query.source_scope {
- let before = rows.len();
- rows.retain(|c| {
- crate::openhuman::memory::source_scope::chunk_source_allowed_in(
- allowed,
- &c.metadata.tags,
- &c.metadata.source_id,
- )
- });
- if rows.len() != before {
- log::debug!(
- "[profiles] list_chunks source-scope filter: {before} -> {} row(s)",
- rows.len()
- );
- }
- rows.truncate(requested_limit as usize);
- }
- Ok(rows)
- })
+ tinycortex::memory::chunks::list_chunks(&engine_config(config), query)
}
/// Count total chunks in the store (useful for tests / diagnostics).
pub fn count_chunks(config: &Config) -> Result<u64> {
tinycortex::memory::chunks::count_chunks(&engine_config(config))
}
/// #002 (FR-010 / US5): extraction coverage — the fraction of chunks that have
/// at least one indexed entity in `mem_tree_entity_index`, in `[0.0, 1.0]`.
///
/// Turns "wiki built / not built" into a quality signal: a value near 0 with a
/// non-zero chunk count means extraction is producing nothing (the model is
/// timing out / failing), even though chunks exist — the "empty-but-built
/// wiki" symptom. Joins the entity index against `mem_tree_chunks.id` so the
/// numerator is node-kind-agnostic (we only count entity rows whose `node_id`
/// is an actual chunk). Returns `0.0` when there are no chunks.
pub fn extraction_coverage(config: &Config) -> Result<f32> {
tinycortex::memory::chunks::extraction_coverage(&engine_config(config))
}
/// Set the lifecycle status column for `chunk_id`. See `CHUNK_STATUS_*`.
pub fn set_chunk_lifecycle_status(config: &Config, chunk_id: &str, status: &str) -> Result<()> {
with_connection(config, |conn| {
set_chunk_lifecycle_status_conn(conn, chunk_id, status)
})
}
pub(crate) fn set_chunk_lifecycle_status_tx(
tx: &Transaction<'_>,
chunk_id: &str,
status: &str,
) -> Result<()> {
set_chunk_lifecycle_status_conn(tx, chunk_id, status)
}
/// Read the lifecycle status column for `chunk_id`, or `None` if the row is absent.
pub fn get_chunk_lifecycle_status(config: &Config, chunk_id: &str) -> Result<Option<String>> {
with_connection(config, |conn| {
get_chunk_lifecycle_status_conn(conn, chunk_id)
})
}
pub(crate) fn get_chunk_lifecycle_status_tx(
tx: &Transaction<'_>,
chunk_id: &str,
) -> Result<Option<String>> {
get_chunk_lifecycle_status_conn(tx, chunk_id)
}
fn get_chunk_lifecycle_status_conn(conn: &Connection, chunk_id: &str) -> Result<Option<String>> {
let row = conn
.query_row(
"SELECT lifecycle_status FROM mem_tree_chunks WHERE id = ?1",
params![chunk_id],
|r| r.get::<_, String>(0),
)
.optional()?;
Ok(row)
}
/// Count chunks currently sitting at a given lifecycle status (test/diagnostic helper).
pub fn count_chunks_by_lifecycle_status(config: &Config, status: &str) -> Result<u64> {
with_connection(config, |conn| {
let n: i64 = conn.query_row(
"SELECT COUNT(*) FROM mem_tree_chunks WHERE lifecycle_status = ?1",
params![status],
|r| r.get(0),
)?;
Ok(n.max(0) as u64)
})
}
fn set_chunk_lifecycle_status_conn(conn: &Connection, chunk_id: &str, status: &str) -> Result<()> {
let changed = conn.execute(
"UPDATE mem_tree_chunks SET lifecycle_status = ?1 WHERE id = ?2",
params![status, chunk_id],
)?;
if changed == 0 {
log::warn!(
"[memory::chunk_store] lifecycle update affected 0 rows chunk_id={} status={}",
@@ -1246,207 +1087,141 @@ pub fn delete_orphaned_source_tree(
"[memory::chunk_store] delete_orphaned_source_tree: source_id_hash={} has no source-scoped tree (gates_cleared={}); shared/collection trees left intact",
redact_value(source_id),
gate_ids.len(),
);
false
};
tx.commit()?;
Ok(cascaded)
})?;
if tree_cascaded {
remove_chunk_content_files(config, &content_paths);
}
Ok(tree_cascaded)
}
fn remove_chunk_content_files(config: &Config, content_paths: &[String]) {
use std::path::{Component, Path};
let root = config.memory_tree_content_root();
let canonical_root = match std::fs::canonicalize(&root) {
Ok(path) => path,
Err(error) => {
if error.kind() != std::io::ErrorKind::NotFound {
log::warn!(
"[memory_tree::store] failed to resolve content root {}: {error}",
root.display(),
);
}
return;
}
};
for rel in content_paths {
let rel_path = Path::new(rel);
let has_escape_component = rel_path.components().any(|component| {
matches!(
component,
Component::ParentDir | Component::RootDir | Component::Prefix(_)
)
});
if has_escape_component {
log::warn!(
"[memory_tree::store] refusing to remove chunk file with unsafe content_path path_hash={}",
redact::redact(rel),
);
continue;
}
let path = root.join(rel_path);
let resolved_path = match std::fs::canonicalize(&path) {
Ok(path) => path,
Err(error) => {
if error.kind() != std::io::ErrorKind::NotFound {
log::warn!(
"[memory_tree::store] failed to resolve chunk file path_hash={}: {error}",
redact::redact(rel),
);
}
continue;
}
};
if !resolved_path.starts_with(&canonical_root) {
log::warn!(
"[memory_tree::store] refusing to remove chunk file outside content root path_hash={}",
redact::redact(rel),
);
continue;
}
if let Err(error) = std::fs::remove_file(&path) {
if error.kind() != std::io::ErrorKind::NotFound {
log::warn!(
"[memory_tree::store] failed to remove chunk file path_hash={}: {error}",
redact::redact(rel),
);
}
}
}
}
-fn row_to_chunk(row: &rusqlite::Row<'_>) -> rusqlite::Result<Chunk> {
- let id: String = row.get(0)?;
- let source_kind_s: String = row.get(1)?;
- let source_id: String = row.get(2)?;
- let path_scope: Option<String> = row.get(3)?;
- let source_ref: Option<String> = row.get(4)?;
- let owner: String = row.get(5)?;
- let ts_ms: i64 = row.get(6)?;
- let trs_ms: i64 = row.get(7)?;
- let tre_ms: i64 = row.get(8)?;
- let tags_json: String = row.get(9)?;
- let content: String = row.get(10)?;
- let token_count: i64 = row.get(11)?;
- let seq: i64 = row.get(12)?;
- let created_ms: i64 = row.get(13)?;
-
- let source_kind = SourceKind::parse(&source_kind_s).map_err(|e| {
- rusqlite::Error::FromSqlConversionFailure(1, rusqlite::types::Type::Text, e.into())
- })?;
- let timestamp = ms_to_utc(ts_ms)?;
- let time_range = (ms_to_utc(trs_ms)?, ms_to_utc(tre_ms)?);
- let created_at = ms_to_utc(created_ms)?;
- let tags: Vec<String> = serde_json::from_str(&tags_json).map_err(|e| {
- rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Text, Box::new(e))
- })?;
-
- Ok(Chunk {
- id,
- content,
- metadata: Metadata {
- source_kind,
- source_id,
- owner,
- timestamp,
- time_range,
- tags,
- source_ref: source_ref.map(SourceRef::new),
- path_scope,
- },
- token_count: token_count.max(0) as u32,
- seq_in_source: seq.max(0) as u32,
- created_at,
- // partial_message is not stored in SQLite — it's a transient chunker
- // signal. Chunks read back from DB always get false (the column doesn't
- // exist; callers that need this flag hold the Chunk in memory).
- partial_message: false,
- })
-}
-
-fn ms_to_utc(ms: i64) -> rusqlite::Result<DateTime<Utc>> {
- Utc.timestamp_millis_opt(ms).single().ok_or_else(|| {
- rusqlite::Error::FromSqlConversionFailure(
- 0,
- rusqlite::types::Type::Integer,
- format!("invalid timestamp ms {ms}").into(),
- )
- })
-}
-
#[path = "connection.rs"]
mod connection;
pub(crate) use connection::recover_corrupt_db;
pub use connection::with_connection;
#[cfg(test)]
#[allow(unused_imports)]
pub(crate) use connection::{
clear_connection_cache, db_path_for, get_or_init_connection, invalidate_connection,
is_io_open_error, schema_apply_count_for_path_for_tests, CB_THRESHOLD,
};
#[cfg(test)]
pub(crate) use connection::{is_transient_cold_start, try_cleanup_stale_files};
#[path = "migrations.rs"]
mod migrations;
use migrations::{migrate_legacy_embeddings_to_sidecar, purge_global_topic_trees};
#[path = "raw_refs.rs"]
mod raw_refs;
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,
};
-fn normalized_limit(requested: Option<usize>) -> i64 {
- let clamped = requested
- .unwrap_or(DEFAULT_LIST_LIMIT)
- .clamp(1, MAX_LIST_LIMIT);
- i64::try_from(clamped).unwrap_or(MAX_LIST_LIMIT as i64)
-}
-
/// Idempotent `ALTER TABLE ADD COLUMN` — treats an existing column as success.
fn add_column_if_missing(conn: &Connection, table: &str, name: &str, sql_type: &str) -> Result<()> {
match conn.execute(
&format!("ALTER TABLE {table} ADD COLUMN {name} {sql_type}"),
[],
) {
Ok(_) => {
log::debug!(
"[memory::chunk_store] migration: added column {table}.{name} ({sql_type})"
);
Ok(())
}
Err(err) if err.to_string().contains("duplicate column name") => Ok(()),
Err(err) => Err(err).with_context(|| format!("Failed to add column {table}.{name}")),
}
}
#[path = "embeddings.rs"]
mod embeddings;
pub use embeddings::{
clear_chunk_reembed_skipped, clear_reembed_skipped_for_signature, 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,
};
#[cfg(test)]
pub(crate) use embeddings::{embedding_to_blob, REEMBED_SKIP_KEY_MAX_LEN};
pub(crate) use embeddings::{
has_uncovered_reembed_work, set_chunk_embedding_for_signature_tx, tree_active_signature,
validate_reembed_skip_key,
};
// ── Phase 2: embedding column accessors ─────────────────────────────────
#[cfg(test)]
#[path = "store_tests.rs"]
mod tests;