use anyhow::Result;
use rusqlite::Connection;
use super::schema_drift::{SchemaObject, SCHEMA_INVARIANTS};
use super::{run_migrations, validate_schema_invariants};
const OLD_V10_SCHEMA: &str = "\
CREATE TABLE sdk_sessions (id INTEGER PRIMARY KEY, content_session_id TEXT UNIQUE NOT NULL, memory_session_id TEXT NOT NULL, project TEXT, user_prompt TEXT, started_at TEXT, started_at_epoch INTEGER, status TEXT DEFAULT 'active', prompt_counter INTEGER DEFAULT 1);
CREATE TABLE observations (id INTEGER PRIMARY KEY, memory_session_id TEXT NOT NULL, project TEXT, type TEXT NOT NULL, title TEXT, subtitle TEXT, narrative TEXT, facts TEXT, concepts TEXT, files_read TEXT, files_modified TEXT, prompt_number INTEGER, created_at TEXT, created_at_epoch INTEGER);
CREATE TABLE session_summaries (id INTEGER PRIMARY KEY, memory_session_id TEXT NOT NULL, project TEXT, request TEXT, completed TEXT, decisions TEXT, learned TEXT, next_steps TEXT, preferences TEXT, prompt_number INTEGER, created_at TEXT, created_at_epoch INTEGER);
CREATE TABLE pending_observations (id INTEGER PRIMARY KEY, session_id TEXT NOT NULL, project TEXT NOT NULL, tool_name TEXT NOT NULL, tool_input TEXT, tool_response TEXT, cwd TEXT, created_at_epoch INTEGER NOT NULL, lease_owner TEXT, lease_expires_epoch INTEGER);
CREATE TABLE memories (id INTEGER PRIMARY KEY, session_id TEXT, project TEXT NOT NULL, topic_key TEXT, title TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT NOT NULL, files TEXT, created_at_epoch INTEGER NOT NULL, updated_at_epoch INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'active');
CREATE TABLE events (id INTEGER PRIMARY KEY, session_id TEXT NOT NULL, project TEXT NOT NULL, event_type TEXT NOT NULL, summary TEXT NOT NULL, detail TEXT, files TEXT, exit_code INTEGER, created_at_epoch INTEGER NOT NULL);
CREATE TABLE summarize_cooldown (project TEXT PRIMARY KEY, last_summarize_epoch INTEGER NOT NULL, last_message_hash TEXT);
CREATE TABLE summarize_locks (project TEXT PRIMARY KEY, lock_epoch INTEGER NOT NULL);
CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5(title, content, content='memories', content_rowid='id', tokenize='trigram');
CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN INSERT INTO memories_fts(rowid, title, content) VALUES (new.id, new.title, new.content); END;
CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, title, content) VALUES ('delete', old.id, old.title, old.content); END;
CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, title, content) VALUES ('delete', old.id, old.title, old.content); INSERT INTO memories_fts(rowid, title, content) VALUES (new.id, new.title, new.content); END;";
const CONVERGENCE_TABLES: &[&str] = &[
"memories",
"observations",
"session_summaries",
"pending_observations",
"sdk_sessions",
"events",
"raw_messages",
"worker_heartbeats",
"hosts",
"workspaces",
"projects",
"sessions",
"event_blobs",
"captured_events",
"captured_event_commits",
"extraction_tasks",
"memory_candidates",
"rule_candidates",
"entities",
"memory_entities",
"memory_lessons",
"ai_usage_events",
"jobs",
"workstreams",
"workstream_sessions",
"workstream_aliases",
"workstream_alias_sources",
"git_commits",
"git_commit_files",
"git_commit_sessions",
"memory_facts",
"procedure_verifications",
"procedure_exports",
"context_injections",
"memory_state_keys",
"topic_segments",
"memory_operation_log",
"memory_edges",
"memory_claims",
"memory_candidate_noops",
"compressed_observation_sources",
"raw_ingest_failures",
"memory_embeddings",
"dream_cluster_decisions",
"dream_quarantine_artifacts",
"external_candidate_identities",
"external_candidate_recurrences",
"graph_candidates",
"graph_file_nodes",
"graph_edges",
"capture_drop_events",
"extraction_replay_ranges",
"context_injection_items",
"memory_citation_events",
"memory_usage_events",
"memory_poisoning_injection_drops",
"memory_lesson_feed_events",
"user_context_claims",
"user_context_summaries",
"user_context_candidates",
"memory_suppressions",
"memory_feedback",
"memory_preference_reinforcements",
"preference_rule_overrides",
"preference_rule_diagnostics",
"ingest_cursors",
"failure_lifecycle_daily",
];
#[test]
fn convergence_tables_cover_schema_invariant_tables() {
use std::collections::BTreeSet;
let convergence: BTreeSet<&str> = CONVERGENCE_TABLES.iter().copied().collect();
let missing: Vec<_> = SCHEMA_INVARIANTS
.iter()
.filter_map(|invariant| match invariant.object {
SchemaObject::Table(table) if !table.ends_with("_fts") => Some(table),
_ => None,
})
.filter(|table| !convergence.contains(table))
.collect();
assert!(
missing.is_empty(),
"schema invariant tables must be included in CONVERGENCE_TABLES: {missing:?}"
);
}
fn make_upgraded_v10_db() -> Result<Connection> {
let conn = Connection::open_in_memory()?;
conn.execute_batch("PRAGMA user_version = 10;")?;
conn.execute_batch(OLD_V10_SCHEMA)?;
run_migrations(&conn)?;
Ok(conn)
}
fn make_fresh_db() -> Result<Connection> {
let conn = Connection::open_in_memory()?;
conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
run_migrations(&conn)?;
Ok(conn)
}
fn normalize_sql_whitespace(sql: &str) -> String {
sql.to_ascii_lowercase()
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
}
fn normalize_identifier(token: &str) -> &str {
token.trim_matches(|c| matches!(c, '"' | '\'' | '`' | '[' | ']' | '('))
}
fn defines_memories_table(sql: &str) -> bool {
let normalized = normalize_sql_whitespace(sql);
let tokens: Vec<&str> = normalized.split_whitespace().collect();
tokens.iter().enumerate().any(|(index, token)| {
if *token != "create" || tokens.get(index + 1) != Some(&"table") {
return false;
}
let table_index = if tokens.get(index + 2..index + 5) == Some(&["if", "not", "exists"]) {
index + 5
} else {
index + 2
};
tokens
.get(table_index)
.is_some_and(|name| normalize_identifier(name) == "memories")
})
}
#[test]
fn memories_table_ddl_detection_collapses_whitespace() {
assert!(defines_memories_table(
"CREATE TABLE IF NOT EXISTS\n memories (id INTEGER PRIMARY KEY)"
));
}
#[test]
fn memories_table_ddl_detection_requires_exact_table_name() {
assert!(!defines_memories_table(
"CREATE TABLE IF NOT EXISTS memories_backup (id INTEGER PRIMARY KEY)"
));
assert!(!defines_memories_table(
"CREATE TABLE IF NOT EXISTS memories_fts (title, content)"
));
}
#[test]
fn migrations_define_memories_table_once() -> Result<()> {
let migrations_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("src")
.join("migrations");
let mut ddl_sources = Vec::new();
for entry in std::fs::read_dir(&migrations_dir)? {
let path = entry?.path();
if path.extension().and_then(|s| s.to_str()) != Some("sql") {
continue;
}
let file_name = path
.file_name()
.and_then(|s| s.to_str())
.unwrap_or_default()
.to_string();
assert!(
!file_name.starts_with("schema_"),
"secondary schema migration source must not exist: {file_name}"
);
let sql = std::fs::read_to_string(&path)?;
if defines_memories_table(&sql) {
ddl_sources.push(file_name);
}
}
ddl_sources.sort();
assert_eq!(
ddl_sources,
vec!["v001_baseline.sql".to_string()],
"exactly one canonical memories table DDL source is allowed"
);
Ok(())
}
fn table_columns(conn: &Connection, table: &str) -> Result<Vec<String>> {
let mut stmt = conn.prepare(&format!("PRAGMA table_info({})", table))?;
let mut cols: Vec<String> = stmt
.query_map([], |row| row.get::<_, String>(1))?
.filter_map(|r| r.ok())
.collect();
cols.sort();
Ok(cols)
}
fn table_indexes(conn: &Connection, table: &str) -> Result<Vec<String>> {
let mut stmt = conn.prepare(
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name=?1 \
AND name NOT LIKE 'sqlite_%' ORDER BY name",
)?;
let indexes = stmt
.query_map([table], |row| row.get::<_, String>(0))?
.filter_map(|r| r.ok())
.collect();
Ok(indexes)
}
fn table_triggers(conn: &Connection, table: &str) -> Result<Vec<String>> {
let mut stmt = conn.prepare(
"SELECT name FROM sqlite_master WHERE type='trigger' AND tbl_name=?1 ORDER BY name",
)?;
let triggers = stmt
.query_map([table], |row| row.get::<_, String>(0))?
.filter_map(|r| r.ok())
.collect();
Ok(triggers)
}
#[test]
fn columns_match_after_upgrade() -> Result<()> {
let fresh = make_fresh_db()?;
let upgraded = make_upgraded_v10_db()?;
let mut mismatches = Vec::new();
for table in CONVERGENCE_TABLES {
let fresh_cols = table_columns(&fresh, table)?;
let upgraded_cols = table_columns(&upgraded, table)?;
if fresh_cols.is_empty() && upgraded_cols.is_empty() {
continue;
}
let missing_in_upgraded: Vec<_> = fresh_cols
.iter()
.filter(|c| !upgraded_cols.contains(c))
.collect();
let extra_in_upgraded: Vec<_> = upgraded_cols
.iter()
.filter(|c| !fresh_cols.contains(c))
.collect();
if !missing_in_upgraded.is_empty() {
mismatches.push(format!(
"{}: columns in fresh but missing after upgrade: {:?}",
table, missing_in_upgraded
));
}
if !extra_in_upgraded.is_empty() {
mismatches.push(format!(
"{}: extra columns after upgrade not in fresh: {:?}",
table, extra_in_upgraded
));
}
}
assert!(
mismatches.is_empty(),
"schema divergence between fresh and upgraded DB:\n {}",
mismatches.join("\n ")
);
Ok(())
}
#[test]
fn indexes_match_after_upgrade() -> Result<()> {
let fresh = make_fresh_db()?;
let upgraded = make_upgraded_v10_db()?;
let mut mismatches = Vec::new();
for table in CONVERGENCE_TABLES {
let fresh_idx = table_indexes(&fresh, table)?;
let upgraded_idx = table_indexes(&upgraded, table)?;
let missing: Vec<_> = fresh_idx
.iter()
.filter(|i| !upgraded_idx.contains(i))
.collect();
if !missing.is_empty() {
mismatches.push(format!(
"{}: indexes in fresh but missing after upgrade: {:?}",
table, missing
));
}
}
assert!(
mismatches.is_empty(),
"index divergence between fresh and upgraded DB:\n {}",
mismatches.join("\n ")
);
Ok(())
}
#[test]
fn graph_edges_candidate_integrity_triggers_match_after_upgrade() -> Result<()> {
let fresh = make_fresh_db()?;
let upgraded = make_upgraded_v10_db()?;
for table in [
"graph_edges",
"memory_candidates",
"graph_candidates",
"memory_operation_log",
] {
let fresh_triggers = table_triggers(&fresh, table)?;
let upgraded_triggers = table_triggers(&upgraded, table)?;
assert_eq!(
fresh_triggers, upgraded_triggers,
"{table} triggers must converge between fresh and upgraded DBs"
);
}
for (table, trigger) in [
(
"graph_edges",
"graph_edges_validate_source_candidate_insert",
),
(
"graph_edges",
"graph_edges_validate_source_candidate_update",
),
("memory_candidates", "graph_edges_memory_candidates_delete"),
(
"memory_candidates",
"graph_edges_memory_candidates_update_id",
),
("graph_candidates", "graph_edges_graph_candidates_delete"),
("graph_candidates", "graph_edges_graph_candidates_update_id"),
(
"memory_operation_log",
"graph_edges_memory_operation_provenance_update",
),
] {
let triggers = table_triggers(&fresh, table)?;
assert!(
triggers.contains(&trigger.to_string()),
"missing graph edge candidate integrity trigger on {table}: {trigger}"
);
}
Ok(())
}
#[test]
fn real_queries_work_on_upgraded_db() -> Result<()> {
use crate::memory::{memory_current_filter_sql, MEMORY_COLS};
let conn = make_upgraded_v10_db()?;
let current_filter = memory_current_filter_sql("m.status", "m.expires_at_epoch", false);
let fts_query = format!(
"SELECT m.{} FROM memories m \
JOIN memories_fts ON memories_fts.rowid = m.id \
WHERE memories_fts MATCH 'test' \
AND {} \
ORDER BY bm25(memories_fts, 10.0, 1.0, 3.0) LIMIT 10",
MEMORY_COLS.replace(", ", ", m."),
current_filter
);
assert!(
conn.prepare(&fts_query).is_ok(),
"FTS search query must work on upgraded DB"
);
let current_filter = memory_current_filter_sql("m.status", "m.expires_at_epoch", false);
let like_query = format!(
"SELECT m.{} FROM memories m \
WHERE m.content LIKE '%test%' \
AND {} \
ORDER BY m.updated_at_epoch DESC LIMIT 10",
MEMORY_COLS.replace(", ", ", m."),
current_filter
);
assert!(
conn.prepare(&like_query).is_ok(),
"LIKE fallback query must work on upgraded DB"
);
let current_filter = memory_current_filter_sql("status", "expires_at_epoch", false);
let scope_query = format!(
"SELECT {} FROM memories \
WHERE (project = 'test' OR scope = 'global') AND {} \
ORDER BY updated_at_epoch DESC LIMIT 10",
MEMORY_COLS, current_filter
);
assert!(
conn.prepare(&scope_query).is_ok(),
"scope-aware read query must work on upgraded DB"
);
Ok(())
}
#[test]
fn user_version_tracks_max_applied_migration() -> Result<()> {
let expected = super::types::OLD_BASELINE_VERSION - 1 + super::latest_schema_version();
for conn in [make_fresh_db()?, make_upgraded_v10_db()?] {
let max_applied = super::state::applied_versions(&conn)?
.into_iter()
.max()
.expect("at least one migration applied");
let user_version: i64 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?;
assert_eq!(
max_applied,
super::latest_schema_version(),
"all migrations must be recorded in _schema_migrations"
);
assert_eq!(
user_version, expected,
"user_version must equal OLD_BASELINE_VERSION-1 + max applied version"
);
}
Ok(())
}
#[test]
fn legacy_upgrade_records_every_migration() -> Result<()> {
let conn = make_upgraded_v10_db()?;
let applied = super::state::applied_versions(&conn)?;
for migration in super::types::MIGRATIONS {
assert!(
applied.contains(&migration.version),
"migration v{} must be recorded after upgrade: {applied:?}",
migration.version
);
}
Ok(())
}
#[test]
fn open_db_enables_foreign_keys() -> Result<()> {
let _guard = crate::db::test_support::ScopedTestDataDir::new("fk-pragma");
let conn = crate::db::open_db()?;
let fk_on: i64 = conn.query_row("PRAGMA foreign_keys", [], |row| row.get(0))?;
assert_eq!(fk_on, 1, "open_db must enable PRAGMA foreign_keys");
Ok(())
}
fn assert_v076_shape_drift(conn: &Connection, expected: &str) -> Result<()> {
let drift = validate_schema_invariants(conn)?;
assert!(
drift.iter().any(|finding| {
finding.contains("v076_dream_poisoning_quarantine critical shape mismatch")
&& finding.contains(expected)
}),
"same-name wrong-shape object must fail loud for {expected}: {drift:?}"
);
Ok(())
}
#[test]
fn v076_same_name_wrong_shape_objects_fail_loud() -> Result<()> {
let wrong_index = Connection::open_in_memory()?;
run_migrations(&wrong_index)?;
wrong_index.execute_batch(
"DROP INDEX idx_external_candidate_identities_candidate;
CREATE INDEX idx_external_candidate_identities_candidate
ON external_candidate_identities(source_kind);",
)?;
assert_v076_shape_drift(
&wrong_index,
"index idx_external_candidate_identities_candidate",
)?;
let wrong_trigger = Connection::open_in_memory()?;
run_migrations(&wrong_trigger)?;
wrong_trigger.execute_batch(
"DROP TRIGGER external_candidate_identities_no_replace;
CREATE TRIGGER external_candidate_identities_no_replace
BEFORE INSERT ON external_candidate_identities BEGIN SELECT 1; END;
DROP TRIGGER dream_quarantine_artifacts_no_replace;
CREATE TRIGGER dream_quarantine_artifacts_no_replace
BEFORE INSERT ON dream_quarantine_artifacts BEGIN SELECT 1; END;
DROP TRIGGER dream_quarantine_artifacts_initial_counters;
CREATE TRIGGER dream_quarantine_artifacts_initial_counters
BEFORE INSERT ON dream_quarantine_artifacts BEGIN SELECT 1; END;
DROP TRIGGER dream_quarantine_artifacts_immutable_payload;
CREATE TRIGGER dream_quarantine_artifacts_immutable_payload
BEFORE UPDATE OF id ON dream_quarantine_artifacts BEGIN SELECT 1; END;
DROP TRIGGER external_candidate_identities_immutable_update;
CREATE TRIGGER external_candidate_identities_immutable_update
BEFORE UPDATE OF identity_sha256 ON external_candidate_identities
BEGIN SELECT 1; END;",
)?;
assert_v076_shape_drift(&wrong_trigger, "external_candidate_identities_no_replace")?;
assert_v076_shape_drift(&wrong_trigger, "dream_quarantine_artifacts_no_replace")?;
assert_v076_shape_drift(
&wrong_trigger,
"dream_quarantine_artifacts_initial_counters",
)?;
assert_v076_shape_drift(
&wrong_trigger,
"dream_quarantine_artifacts_immutable_payload",
)?;
assert_v076_shape_drift(
&wrong_trigger,
"external_candidate_identities_immutable_update",
)?;
let wrong_dream = Connection::open_in_memory()?;
run_migrations(&wrong_dream)?;
wrong_dream.execute_batch(
"DROP TABLE dream_quarantine_artifacts;
CREATE TABLE dream_quarantine_artifacts (
id INTEGER PRIMARY KEY,
version INTEGER NOT NULL DEFAULT 1,
project TEXT NOT NULL,
cluster_signature TEXT NOT NULL,
member_ids_json TEXT NOT NULL,
source_candidate_id INTEGER NOT NULL
REFERENCES memory_candidates(id) ON DELETE CASCADE,
decision_kind TEXT NOT NULL,
decision_ids_json TEXT NOT NULL,
decision_payload_sha256 TEXT NOT NULL,
intended_superseded_ids_json TEXT NOT NULL,
generated_topic_key TEXT,
generated_memory_type TEXT,
generated_title TEXT,
generated_content TEXT,
generated_field TEXT NOT NULL,
pattern_id TEXT NOT NULL,
pattern_version INTEGER NOT NULL,
source_operation TEXT NOT NULL,
source_trust_class TEXT NOT NULL,
occurrence_count INTEGER NOT NULL DEFAULT 1,
created_at_epoch INTEGER NOT NULL,
updated_at_epoch INTEGER NOT NULL,
UNIQUE(project, cluster_signature)
);",
)?;
assert_v076_shape_drift(
&wrong_dream,
"missing UNIQUE(project,cluster_signature,source_candidate_id)",
)?;
assert_v076_shape_drift(
&wrong_dream,
"source_candidate_id missing REFERENCES memory_candidates(id) ON DELETE RESTRICT",
)?;
assert_v076_shape_drift(&wrong_dream, "missing SQL contract `decision_kind in")?;
assert_v076_shape_drift(
&wrong_dream,
"missing SQL contract `json_valid(decision_ids_json)`",
)?;
assert_v076_shape_drift(
&wrong_dream,
"missing SQL contract `length(decision_payload_sha256) = 64`",
)?;
assert_v076_shape_drift(
&wrong_dream,
"missing SQL contract `generated_field in ( 'dream.topic_key'",
)?;
assert_v076_shape_drift(
&wrong_dream,
"missing SQL contract `length(trim(generated_topic_key)) > 0`",
)?;
assert_v076_shape_drift(
&wrong_dream,
"missing SQL contract `length(trim(generated_memory_type)) > 0`",
)?;
let wrong_identity = Connection::open_in_memory()?;
run_migrations(&wrong_identity)?;
wrong_identity.execute_batch(
"DROP TABLE external_candidate_recurrences;
DROP TABLE external_candidate_identities;
CREATE TABLE external_candidate_identities (
identity_sha256 TEXT PRIMARY KEY NOT NULL,
candidate_id INTEGER NOT NULL
REFERENCES memory_candidates(id) ON DELETE RESTRICT,
source_kind TEXT NOT NULL,
memory_type TEXT NOT NULL,
semantic_discriminator_sha256 TEXT,
source_project TEXT NOT NULL,
owner_scope TEXT NOT NULL,
owner_key TEXT NOT NULL,
target_project TEXT,
topic_key TEXT NOT NULL,
text_sha256 TEXT NOT NULL,
first_seen_epoch INTEGER NOT NULL,
last_seen_epoch INTEGER NOT NULL,
occurrence_count INTEGER NOT NULL
);",
)?;
assert_v076_shape_drift(
&wrong_identity,
"missing SQL contract `text_sha256 not glob '*[^0-9a-f]*'`",
)?;
let wrong_recurrence = Connection::open_in_memory()?;
run_migrations(&wrong_recurrence)?;
wrong_recurrence.execute_batch(
"DROP TABLE external_candidate_recurrences;
CREATE TABLE external_candidate_recurrences (
id INTEGER,
identity_sha256 TEXT NOT NULL
REFERENCES external_candidate_identities(identity_sha256)
ON DELETE RESTRICT,
canonical_candidate_id INTEGER NOT NULL
REFERENCES memory_candidates(id) ON DELETE RESTRICT,
candidate_id INTEGER NOT NULL
REFERENCES memory_candidates(id) ON DELETE RESTRICT,
recurrence_kind TEXT NOT NULL,
pattern_id TEXT,
pattern_version INTEGER,
occurred_at_epoch INTEGER NOT NULL
);",
)?;
assert_v076_shape_drift(
&wrong_recurrence,
"external_candidate_recurrences.id must be PRIMARY KEY",
)?;
Ok(())
}