mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
444 lines
18 KiB
Rust
444 lines
18 KiB
Rust
//! `UnifiedMemory` constructor + schema bootstrap.
|
|
//!
|
|
//! Creates the workspace directories, opens the SQLite connection in WAL mode,
|
|
//! materialises every table the unified store owns (docs, kv, graph, vector
|
|
//! chunks, episodic FTS5, segments, events, profile), and runs idempotent
|
|
//! legacy-namespace migrations. Also exposes path / namespace helpers shared
|
|
//! by the rest of the unified module.
|
|
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::Context as _;
|
|
use parking_lot::Mutex;
|
|
use rusqlite::Connection;
|
|
|
|
use crate::openhuman::embeddings::EmbeddingProvider;
|
|
use crate::openhuman::memory_store::types::GLOBAL_NAMESPACE;
|
|
|
|
use super::UnifiedMemory;
|
|
|
|
impl UnifiedMemory {
|
|
/// Open (or create) the unified store rooted at `workspace_dir`.
|
|
///
|
|
/// Delegates to [`Self::new_with_memory_dir`] using the default
|
|
/// `"memory"` subdirectory name. Safe to call on every boot.
|
|
pub fn new(
|
|
workspace_dir: &Path,
|
|
embedder: Arc<dyn EmbeddingProvider>,
|
|
_open_timeout_secs: Option<u64>,
|
|
) -> anyhow::Result<Self> {
|
|
Self::new_with_memory_dir(workspace_dir, "memory", embedder, _open_timeout_secs)
|
|
}
|
|
|
|
/// Open (or create) a unified store using an explicit memory subdirectory
|
|
/// name under `workspace_dir`.
|
|
///
|
|
/// This enables multiple independent stores inside the same workspace (e.g.
|
|
/// per-personality databases) without path collisions. `memory_subdir` is
|
|
/// joined directly to `workspace_dir`, so `"memory-1"` yields
|
|
/// `workspace_dir/memory-1/memory.db`.
|
|
///
|
|
/// Creates the on-disk layout, runs all `CREATE TABLE` statements, and
|
|
/// applies idempotent legacy-namespace migrations. Safe to call on every
|
|
/// boot.
|
|
pub fn new_with_memory_dir(
|
|
workspace_dir: &Path,
|
|
memory_subdir: &str,
|
|
embedder: Arc<dyn EmbeddingProvider>,
|
|
_open_timeout_secs: Option<u64>,
|
|
) -> anyhow::Result<Self> {
|
|
use std::path::Component;
|
|
anyhow::ensure!(!memory_subdir.is_empty(), "memory_subdir must not be empty");
|
|
let subdir_path = Path::new(memory_subdir);
|
|
anyhow::ensure!(
|
|
subdir_path.components().count() == 1
|
|
&& subdir_path
|
|
.components()
|
|
.all(|c| matches!(c, Component::Normal(_))),
|
|
"memory_subdir must be a single relative path component without traversal"
|
|
);
|
|
let memory_dir = workspace_dir.join(subdir_path);
|
|
let namespaces_dir = memory_dir.join("namespaces");
|
|
let vectors_dir = memory_dir.join("vectors");
|
|
std::fs::create_dir_all(&namespaces_dir)?;
|
|
std::fs::create_dir_all(&vectors_dir)?;
|
|
|
|
let db_path = memory_dir.join("memory.db");
|
|
let conn = Connection::open(&db_path)?;
|
|
// Active storage layout for the core memory domain:
|
|
// - memory_docs: namespace-scoped source documents and markdown metadata.
|
|
// - vector_chunks: chunked document text plus optional local embedding bytes.
|
|
// - graph_namespace: namespace graph edges used for relation-first retrieval.
|
|
// - graph_global: cross-namespace graph edges used as fallback/shared memory.
|
|
// - kv_namespace: namespace-scoped durable preferences, decisions, and state.
|
|
// - kv_global: global durable key-value memories outside a namespace scope.
|
|
// Absorb concurrent write contention under cargo-llvm-cov and any other
|
|
// scenario where background workers hold the write lock while a second
|
|
// connection attempts a write. Without a timeout the driver returns
|
|
// SQLITE_BUSY immediately, which causes test flakes and runtime errors.
|
|
conn.busy_timeout(std::time::Duration::from_secs(15))
|
|
.context("configure unified memory busy_timeout")?;
|
|
|
|
conn.execute_batch(
|
|
"PRAGMA journal_mode = WAL;
|
|
PRAGMA synchronous = NORMAL;
|
|
|
|
CREATE TABLE IF NOT EXISTS memory_docs (
|
|
document_id TEXT PRIMARY KEY,
|
|
namespace TEXT NOT NULL,
|
|
key TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
source_type TEXT NOT NULL,
|
|
priority TEXT NOT NULL,
|
|
tags_json TEXT NOT NULL,
|
|
metadata_json TEXT NOT NULL,
|
|
category TEXT NOT NULL,
|
|
session_id TEXT,
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
markdown_rel_path TEXT NOT NULL,
|
|
taint TEXT NOT NULL DEFAULT 'internal',
|
|
UNIQUE(namespace, key)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_memory_docs_ns_updated ON memory_docs(namespace, updated_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS kv_global (
|
|
key TEXT PRIMARY KEY,
|
|
value_json TEXT NOT NULL,
|
|
updated_at REAL NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS kv_namespace (
|
|
namespace TEXT NOT NULL,
|
|
key TEXT NOT NULL,
|
|
value_json TEXT NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
PRIMARY KEY(namespace, key)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_kv_namespace_ns ON kv_namespace(namespace);
|
|
|
|
CREATE TABLE IF NOT EXISTS graph_global (
|
|
subject TEXT NOT NULL,
|
|
predicate TEXT NOT NULL,
|
|
object TEXT NOT NULL,
|
|
attrs_json TEXT NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
PRIMARY KEY(subject, predicate, object)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_graph_global_subject ON graph_global(subject, predicate);
|
|
|
|
CREATE TABLE IF NOT EXISTS graph_namespace (
|
|
namespace TEXT NOT NULL,
|
|
subject TEXT NOT NULL,
|
|
predicate TEXT NOT NULL,
|
|
object TEXT NOT NULL,
|
|
attrs_json TEXT NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
PRIMARY KEY(namespace, subject, predicate, object)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_graph_namespace_ns ON graph_namespace(namespace);
|
|
CREATE INDEX IF NOT EXISTS idx_graph_namespace_subject ON graph_namespace(namespace, subject, predicate);
|
|
|
|
CREATE TABLE IF NOT EXISTS vector_chunks (
|
|
namespace TEXT NOT NULL,
|
|
document_id TEXT NOT NULL,
|
|
chunk_id TEXT NOT NULL,
|
|
text TEXT NOT NULL,
|
|
embedding BLOB,
|
|
metadata_json TEXT NOT NULL,
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
model_signature TEXT,
|
|
dim INTEGER,
|
|
PRIMARY KEY(namespace, chunk_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_vector_chunks_ns_doc ON vector_chunks(namespace, document_id);",
|
|
)?;
|
|
|
|
// Tag vector_chunks with the embedding model that produced each vector
|
|
// on existing databases (idempotent). Fresh installs get these from the
|
|
// CREATE TABLE above; older DBs need the ALTERs so recall can exclude
|
|
// vectors generated by a different embedding model (cross-model cosine is
|
|
// garbage) and skip dimension mismatches instead of silently scoring 0.
|
|
for sql in [
|
|
"ALTER TABLE vector_chunks ADD COLUMN model_signature TEXT",
|
|
"ALTER TABLE vector_chunks ADD COLUMN dim INTEGER",
|
|
] {
|
|
match conn.execute(sql, []) {
|
|
Ok(_) => tracing::debug!("[vector_chunks:init] applied: {sql}"),
|
|
Err(e) => {
|
|
tracing::trace!("[vector_chunks:init] skipped (probably already exists): {e}")
|
|
}
|
|
}
|
|
}
|
|
|
|
// Backfill the `taint` column on existing `memory_docs` databases.
|
|
// Fresh installs get this via the CREATE TABLE above; older DBs need
|
|
// the ALTER so retrieval can carry the provenance signal up to the
|
|
// subconscious gate. Idempotent: a duplicate-column error on
|
|
// re-application is expected (logged at trace).
|
|
match conn.execute(
|
|
"ALTER TABLE memory_docs ADD COLUMN taint TEXT NOT NULL DEFAULT 'internal'",
|
|
[],
|
|
) {
|
|
Ok(_) => tracing::debug!("[memory_docs:init] applied taint column migration"),
|
|
Err(e) => {
|
|
tracing::trace!("[memory_docs:init] taint column already present: {e}")
|
|
}
|
|
}
|
|
|
|
// Create FTS5 episodic tables (episodic_log, episodic_fts, and their
|
|
// triggers) so the Archivist can call episodic_insert immediately after
|
|
// the store is initialised.
|
|
conn.execute_batch(super::fts5::EPISODIC_INIT_SQL)?;
|
|
|
|
// Conversation segmentation tables.
|
|
conn.execute_batch(super::segments::SEGMENTS_INIT_SQL)?;
|
|
|
|
// Backfill the (start_seq, end_seq) columns on existing databases
|
|
// — fresh installs get them from SEGMENTS_INIT_SQL above; older DBs
|
|
// need the ALTER TABLEs. Idempotent: a duplicate-column error is
|
|
// expected and logged at trace level.
|
|
for sql in super::segments::SEGMENTS_MIGRATIONS_SQL {
|
|
match conn.execute(sql, []) {
|
|
Ok(_) => tracing::debug!("[segments:init] applied: {sql}"),
|
|
Err(e) => {
|
|
tracing::trace!("[segments:init] skipped (probably already exists): {e}")
|
|
}
|
|
}
|
|
}
|
|
|
|
// Event extraction tables.
|
|
conn.execute_batch(super::events::EVENTS_INIT_SQL)?;
|
|
|
|
// Phase 3 (#566): add new columns to existing databases BEFORE running
|
|
// PROFILE_INIT_SQL. PROFILE_INIT_SQL includes indexes that reference
|
|
// state/stability/user_state — those CREATE INDEX statements fail on
|
|
// pre-Phase-3 databases unless the columns already exist.
|
|
// On fresh installs the table doesn't exist yet so ALTER TABLE fails
|
|
// silently here; PROFILE_INIT_SQL then creates it with all columns.
|
|
{
|
|
use super::profile::PHASE3_COLUMNS_SQL;
|
|
for sql in PHASE3_COLUMNS_SQL.iter() {
|
|
match conn.execute(sql, []) {
|
|
Ok(_) => tracing::debug!("[profile:init] applied: {sql}"),
|
|
Err(e) => {
|
|
tracing::trace!("[profile:init] skipped (probably already exists): {e}")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// User profile accumulation table (CREATE TABLE IF NOT EXISTS + all indexes).
|
|
// On existing databases the table creation is a no-op; the index creation
|
|
// succeeds because PHASE3_COLUMNS_SQL above has already added the columns.
|
|
conn.execute_batch(super::profile::PROFILE_INIT_SQL)?;
|
|
|
|
// Phase 3 indexes: idempotently restore performance indexes removed in #1616.
|
|
// New installs get these from PROFILE_INIT_SQL; existing DBs get them here,
|
|
// after PHASE3_COLUMNS_SQL has ensured the columns exist.
|
|
{
|
|
use super::profile::PHASE3_INDEXES_SQL;
|
|
for sql in PHASE3_INDEXES_SQL {
|
|
match conn.execute_batch(sql) {
|
|
Ok(_) => tracing::debug!("[profile:init] index applied: {sql}"),
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"[profile:init] index creation failed (non-fatal): {sql}: {e}"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Idempotent legacy-namespace migration.
|
|
//
|
|
// Older writes via MemoryStoreTool packed the intended namespace into
|
|
// the key as `"{namespace}/{actual_key}"` and stored the row under the
|
|
// GLOBAL_NAMESPACE. Split those rows now so the new trait surface can
|
|
// rely on the `namespace` column.
|
|
//
|
|
// The anti-join guard prevents duplicate-split collisions if a
|
|
// post-split row already exists (UNIQUE(namespace, key) would otherwise
|
|
// fail). Safe to run on every boot.
|
|
let migrated = conn.execute(
|
|
"UPDATE memory_docs
|
|
SET namespace = substr(key, 1, instr(key, '/') - 1),
|
|
key = substr(key, instr(key, '/') + 1)
|
|
WHERE namespace = ?1
|
|
AND instr(key, '/') > 0
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM memory_docs m2
|
|
WHERE m2.namespace = substr(memory_docs.key, 1, instr(memory_docs.key, '/') - 1)
|
|
AND m2.key = substr(memory_docs.key, instr(memory_docs.key, '/') + 1)
|
|
)",
|
|
rusqlite::params![GLOBAL_NAMESPACE],
|
|
)?;
|
|
if migrated > 0 {
|
|
log::info!(
|
|
"[memory] migrated {migrated} legacy `ns/key` rows out of the `{GLOBAL_NAMESPACE}` namespace"
|
|
);
|
|
}
|
|
|
|
// Companion migration: `vector_chunks` rows keyed by `document_id` still
|
|
// point at `GLOBAL_NAMESPACE` after the `memory_docs` split above, so
|
|
// namespace-scoped recall would miss them. Re-home each chunk to its
|
|
// document's new namespace. Idempotent: after both migrations run, no
|
|
// chunk under GLOBAL_NAMESPACE maps to a document in another namespace.
|
|
let chunks_migrated = conn.execute(
|
|
"UPDATE vector_chunks
|
|
SET namespace = (
|
|
SELECT namespace FROM memory_docs
|
|
WHERE memory_docs.document_id = vector_chunks.document_id
|
|
)
|
|
WHERE namespace = ?1
|
|
AND document_id IN (
|
|
SELECT document_id FROM memory_docs WHERE namespace != ?1
|
|
)",
|
|
rusqlite::params![GLOBAL_NAMESPACE],
|
|
)?;
|
|
if chunks_migrated > 0 {
|
|
log::info!(
|
|
"[memory] migrated {chunks_migrated} vector_chunks rows out of the `{GLOBAL_NAMESPACE}` namespace"
|
|
);
|
|
}
|
|
|
|
Ok(Self {
|
|
workspace_dir: workspace_dir.to_path_buf(),
|
|
memory_dir,
|
|
db_path,
|
|
vectors_dir,
|
|
conn: Arc::new(Mutex::new(conn)),
|
|
embedder,
|
|
})
|
|
}
|
|
|
|
/// Root workspace directory holding `memory/` and its subtrees.
|
|
pub fn workspace_dir(&self) -> &Path {
|
|
&self.workspace_dir
|
|
}
|
|
|
|
/// Filesystem path of the SQLite database file.
|
|
pub fn db_path(&self) -> &Path {
|
|
&self.db_path
|
|
}
|
|
|
|
/// Directory used for vector-related sidecar files.
|
|
pub fn vectors_dir(&self) -> &Path {
|
|
&self.vectors_dir
|
|
}
|
|
|
|
pub(crate) fn now_ts() -> f64 {
|
|
use std::time::{SystemTime, UNIX_EPOCH};
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.map(|d| d.as_secs_f64())
|
|
.unwrap_or(0.0)
|
|
}
|
|
|
|
pub(crate) fn sanitize_namespace(namespace: &str) -> String {
|
|
let trimmed = namespace.trim();
|
|
if trimmed.is_empty() {
|
|
return GLOBAL_NAMESPACE.to_string();
|
|
}
|
|
trimmed
|
|
.chars()
|
|
.map(|ch| {
|
|
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '/' {
|
|
ch
|
|
} else {
|
|
'_'
|
|
}
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Resolved memory subdirectory for this store instance (e.g.
|
|
/// `workspace_dir/memory` for the default store, or a custom subdir for
|
|
/// personality-specific stores).
|
|
pub fn memory_dir(&self) -> &Path {
|
|
&self.memory_dir
|
|
}
|
|
|
|
pub(crate) fn namespace_dir(&self, namespace: &str) -> PathBuf {
|
|
self.memory_dir
|
|
.join("namespaces")
|
|
.join(Self::sanitize_namespace(namespace))
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::openhuman::embeddings::NoopEmbedding;
|
|
use tempfile::TempDir;
|
|
|
|
#[test]
|
|
fn sanitize_namespace_defaults_and_scrubs() {
|
|
assert_eq!(UnifiedMemory::sanitize_namespace(""), GLOBAL_NAMESPACE);
|
|
assert_eq!(UnifiedMemory::sanitize_namespace(" "), GLOBAL_NAMESPACE);
|
|
assert_eq!(
|
|
UnifiedMemory::sanitize_namespace("team alpha/#1"),
|
|
"team_alpha/_1"
|
|
);
|
|
assert_eq!(UnifiedMemory::sanitize_namespace("a-b_c/ok"), "a-b_c/ok");
|
|
}
|
|
|
|
#[test]
|
|
fn namespace_dir_uses_sanitized_namespace() {
|
|
let tmp = TempDir::new().unwrap();
|
|
let memory = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap();
|
|
let dir = memory.namespace_dir("team alpha/#1");
|
|
assert_eq!(
|
|
dir,
|
|
tmp.path()
|
|
.join("memory")
|
|
.join("namespaces")
|
|
.join("team_alpha/_1")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn new_with_memory_dir_creates_separate_db() {
|
|
let tmp = TempDir::new().unwrap();
|
|
let mem1 = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap();
|
|
let mem2 = UnifiedMemory::new_with_memory_dir(
|
|
tmp.path(),
|
|
"memory-1",
|
|
Arc::new(NoopEmbedding),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
assert_ne!(mem1.db_path(), mem2.db_path());
|
|
assert!(
|
|
mem1.db_path().ends_with("memory/memory.db"),
|
|
"expected mem1 db under memory/memory.db, got {:?}",
|
|
mem1.db_path()
|
|
);
|
|
assert!(
|
|
mem2.db_path().ends_with("memory-1/memory.db"),
|
|
"expected mem2 db under memory-1/memory.db, got {:?}",
|
|
mem2.db_path()
|
|
);
|
|
assert!(mem1.db_path().exists(), "mem1 db file must exist on disk");
|
|
assert!(mem2.db_path().exists(), "mem2 db file must exist on disk");
|
|
}
|
|
|
|
#[test]
|
|
fn connection_has_busy_timeout_set() {
|
|
let tmp = TempDir::new().unwrap();
|
|
let memory = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap();
|
|
let conn = memory.conn.lock();
|
|
// SQLite reports busy_timeout as a PRAGMA; 0 means no timeout.
|
|
let timeout: i64 = conn
|
|
.query_row("PRAGMA busy_timeout", [], |row| row.get(0))
|
|
.unwrap();
|
|
assert!(
|
|
timeout > 0,
|
|
"busy_timeout must be non-zero to absorb write contention, got {timeout}"
|
|
);
|
|
}
|
|
}
|