use std::collections::BTreeMap;
use bamboo_domain::{ConversationSummary, Message};
use rusqlite::types::Value;
use tempfile::TempDir;
use super::*;
fn fixture(id: &str) -> Session {
let mut session = Session::new(id, "model");
session.title = "search beacon".to_string();
for index in 0..3 {
let mut message = Message::user(format!("quartz message {index}"));
message.id = format!("m{index}");
session.messages.push(message);
}
session
}
fn rows(conn: &Connection, sql: &str) -> Vec<Vec<Value>> {
let mut statement = conn.prepare(sql).unwrap();
let columns = statement.column_count();
statement
.query_map([], |row| (0..columns).map(|index| row.get(index)).collect())
.unwrap()
.collect::<Result<_, _>>()
.unwrap()
}
fn search_snapshot(path: &Path, query: &str) -> serde_json::Value {
let snapshot = serde_json::to_value(search_db(path, query, 10).unwrap()).unwrap();
for result in snapshot.as_array().unwrap() {
let rank = result.as_object().unwrap().get("rank").unwrap();
assert!(rank.as_f64().unwrap().is_finite());
}
snapshot
}
fn identity_rows(conn: &Connection) -> BTreeMap<(String, String), i64> {
conn.prepare("SELECT session_id, message_id, search_rowid FROM session_messages_search")
.unwrap()
.query_map([], |row| Ok(((row.get(0)?, row.get(1)?), row.get(2)?)))
.unwrap()
.collect::<Result<_, _>>()
.unwrap()
}
fn legacy_db(path: &Path) -> Connection {
let conn = Connection::open(path).unwrap();
conn.execute_batch(
"CREATE TABLE session_search_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO session_search_meta VALUES ('schema_version', '3');
CREATE TABLE sessions_search (
session_id TEXT PRIMARY KEY, title TEXT NOT NULL, kind TEXT NOT NULL,
root_session_id TEXT NOT NULL, parent_session_id TEXT, pinned INTEGER NOT NULL,
updated_at TEXT NOT NULL, summary TEXT);
CREATE TABLE session_messages_search (
session_id TEXT NOT NULL, message_id TEXT NOT NULL, message_index INTEGER NOT NULL,
role TEXT NOT NULL, content TEXT NOT NULL, compressed INTEGER NOT NULL,
created_at TEXT NOT NULL, PRIMARY KEY(session_id, message_id));
CREATE INDEX idx_session_messages_search_session_id
ON session_messages_search(session_id, message_index);
CREATE VIRTUAL TABLE sessions_search_fts USING fts5(session_id UNINDEXED, title, summary);
CREATE VIRTUAL TABLE session_messages_search_fts USING fts5(
session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED, role UNINDEXED, content);
CREATE TABLE user_audit (kind TEXT NOT NULL);
CREATE TABLE unrelated (value TEXT NOT NULL);
INSERT INTO unrelated VALUES ('preserve this table');",
).unwrap();
let now = Utc::now().to_rfc3339();
conn.execute(
"INSERT INTO sessions_search VALUES
('legacy', 'legacy beacon', 'child', 'root', 'parent', 1, ?1, 'archived summary')",
[&now],
)
.unwrap();
conn.execute(
"INSERT INTO session_messages_search (rowid, session_id, message_id,
message_index, role, content, compressed, created_at)
VALUES (7, 'legacy', 'a', 0, 'user', 'quartz legacy message', 1, ?1),
(29, 'legacy', 'b', 1, 'assistant', 'unrelated answer', 0, ?1)",
[&now],
)
.unwrap();
conn.execute_batch("INSERT INTO sessions_search_fts (rowid, session_id, title, summary)
VALUES (900, 'legacy', 'legacy beacon', 'archived summary');
INSERT INTO session_messages_search_fts (rowid, session_id, message_id, message_index, role, content)
VALUES (901, 'legacy', 'a', 0, 'user', 'quartz legacy message'),
(905, 'legacy', 'b', 1, 'assistant', 'unrelated answer');
CREATE INDEX user_title_index ON sessions_search(title COLLATE NOCASE);
CREATE UNIQUE INDEX user_message_expression ON session_messages_search(length(content), message_id);
CREATE TRIGGER user_session_update AFTER UPDATE ON sessions_search BEGIN
INSERT INTO user_audit VALUES ('session'); END;
CREATE TRIGGER user_message_update AFTER UPDATE ON session_messages_search BEGIN
INSERT INTO user_audit SELECT title FROM sessions_search WHERE session_id=NEW.session_id; END;")
.unwrap();
conn
}
fn v4_db(path: &Path) -> Connection {
let conn = Connection::open(path).unwrap();
conn.execute_batch(
"CREATE TABLE session_search_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO session_search_meta VALUES ('schema_version', '4');
CREATE TABLE sessions_search (
search_rowid INTEGER PRIMARY KEY, session_id TEXT NOT NULL UNIQUE,
title TEXT NOT NULL, kind TEXT NOT NULL, root_session_id TEXT NOT NULL,
parent_session_id TEXT, pinned INTEGER NOT NULL, updated_at TEXT NOT NULL,
summary TEXT);
CREATE TABLE session_messages_search (
search_rowid INTEGER PRIMARY KEY, session_id TEXT NOT NULL,
message_id TEXT NOT NULL, message_index INTEGER NOT NULL, role TEXT NOT NULL,
content TEXT NOT NULL, compressed INTEGER NOT NULL, created_at TEXT NOT NULL,
UNIQUE(session_id, message_id));
CREATE INDEX idx_session_messages_search_session_id
ON session_messages_search(session_id, message_index);
CREATE VIRTUAL TABLE sessions_search_fts USING fts5(
session_id UNINDEXED, title, summary);
CREATE VIRTUAL TABLE session_messages_search_fts USING fts5(
session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content);
CREATE INDEX user_v4_title_index ON sessions_search(title COLLATE NOCASE);",
)
.unwrap();
let now = Utc::now().to_rfc3339();
conn.execute(
"INSERT INTO sessions_search
(search_rowid, session_id, title, kind, root_session_id, parent_session_id,
pinned, updated_at, summary)
VALUES (17, 'v4', 'v4 history', 'root', 'v4', NULL, 0, ?1, NULL)",
[&now],
)
.unwrap();
conn.execute(
"INSERT INTO session_messages_search
(search_rowid, session_id, message_id, message_index, role, content, compressed,
created_at)
VALUES (41, 'v4', 'v4-message', 0, 'user', '我们要压缩上下文 release notes', 1, ?1)",
[&now],
)
.unwrap();
conn.execute_batch(
"INSERT INTO sessions_search_fts (rowid, session_id, title, summary)
VALUES (17, 'v4', 'v4 history', '');
INSERT INTO session_messages_search_fts
(rowid, session_id, message_id, message_index, role, content)
VALUES (41, 'v4', 'v4-message', 0, 'user', '我们要压缩上下文');",
)
.unwrap();
conn
}
fn assert_identity_alignment(conn: &Connection) {
for (normal, fts) in [
("sessions_search", "sessions_search_fts"),
("session_messages_search", "session_messages_search_fts"),
(
"session_messages_search",
"session_messages_current_search_fts",
),
] {
let mut columns = "session_id".to_string();
if normal == "session_messages_search" {
columns.push_str(", message_id");
}
assert_eq!(
rows(
conn,
&format!("SELECT search_rowid, {columns} FROM {normal} ORDER BY search_rowid")
),
rows(
conn,
&format!("SELECT rowid, {columns} FROM {fts} ORDER BY rowid")
)
);
}
}
#[test]
fn independent_v3_migration_preserves_search_cache_and_user_objects() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
let conn = legacy_db(&path);
let object_sql = rows(
&conn,
"SELECT name, sql FROM sqlite_schema
WHERE name LIKE 'user_%' OR name='idx_session_messages_search_session_id' ORDER BY name",
);
let before_search = search_snapshot(&path, "quartz");
let before_session = search_snapshot(&path, "beacon");
let before_cache =
serde_json::to_value(read_compressed_cache_db(&path, "legacy", 0, 20, 100).unwrap())
.unwrap();
drop(conn);
for _ in 0..3 {
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
assert_eq!(
conn.query_row(
"SELECT value FROM session_search_meta WHERE key='schema_version'",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"5"
);
assert_identity_alignment(&conn);
assert_eq!(
identity_rows(&conn).values().copied().collect::<Vec<_>>(),
vec![7, 29]
);
assert_eq!(rows(&conn, "SELECT name, sql FROM sqlite_schema
WHERE name LIKE 'user_%' OR name='idx_session_messages_search_session_id' ORDER BY name"), object_sql);
assert_eq!(
rows(&conn, "SELECT * FROM unrelated"),
vec![vec![Value::Text("preserve this table".into())]]
);
assert_eq!(search_snapshot(&path, "quartz"), before_search);
assert_eq!(search_snapshot(&path, "beacon"), before_session);
assert_eq!(
serde_json::to_value(read_compressed_cache_db(&path, "legacy", 0, 20, 100).unwrap())
.unwrap(),
before_cache
);
}
let conn = open_db(&path).unwrap();
conn.execute(
"UPDATE session_messages_search SET compressed=0 WHERE message_id='a'",
[],
)
.unwrap();
conn.execute("UPDATE sessions_search SET pinned=0", [])
.unwrap();
assert_eq!(
rows(&conn, "SELECT * FROM user_audit ORDER BY rowid"),
vec![
vec![Value::Text("legacy beacon".into())],
vec![Value::Text("session".into())]
]
);
}
#[test]
fn independent_v4_migration_rebuilds_search_projections_and_is_idempotent() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
let conn = v4_db(&path);
let ordinary_before = rows(
&conn,
"SELECT search_rowid, session_id, message_id, message_index, role, content,
compressed, created_at
FROM session_messages_search",
);
drop(conn);
for _ in 0..3 {
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
assert_eq!(
conn.query_row(
"SELECT value FROM session_search_meta WHERE key='schema_version'",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"5"
);
assert_eq!(
rows(
&conn,
"SELECT search_rowid, session_id, message_id, message_index, role, content,
compressed, created_at
FROM session_messages_search"
),
ordinary_before
);
assert_eq!(
conn.query_row(
"SELECT source_revision FROM sessions_search WHERE session_id='v4'",
[],
|row| row.get::<_, Option<String>>(0)
)
.unwrap(),
None
);
assert!(conn
.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='user_v4_title_index')",
[],
|row| row.get::<_, bool>(0)
)
.unwrap());
assert_identity_alignment(&conn);
drop(conn);
let page = search_session_messages_db(&path, "v4", "压缩", usize::MAX, &[], 10).unwrap();
assert_eq!(page.matches.len(), 1);
assert_eq!(page.matches[0].message_id, "v4-message");
assert_eq!(page.matches[0].match_source, "fts_cjk_bigram");
assert!(!page.used_literal_fallback);
let page = search_session_messages_db(&path, "v4", "lease", usize::MAX, &[], 10).unwrap();
assert_eq!(page.matches.len(), 1);
assert_eq!(page.matches[0].message_id, "v4-message");
assert_eq!(page.matches[0].match_source, "fts_literal_trigram");
assert!(!page.used_literal_fallback);
}
}
#[test]
fn stale_legacy_version_on_complete_v5_index_preserves_search_data() {
for stale_version in ["3", "4"] {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
drop(legacy_db(&path));
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
let tables = [
"sessions_search",
"session_messages_search",
"sessions_search_fts",
"session_messages_search_fts",
"session_messages_current_search_fts",
];
let before = tables
.iter()
.map(|table| {
rows(
&conn,
&format!("SELECT rowid, * FROM {table} ORDER BY rowid"),
)
})
.collect::<Vec<_>>();
let schema_before = rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name");
let search_before = search_snapshot(&path, "quartz");
conn.execute(
"UPDATE session_search_meta SET value=?1 WHERE key='schema_version'",
[stale_version],
)
.unwrap();
drop(conn);
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
assert_eq!(
conn.query_row(
"SELECT value FROM session_search_meta WHERE key='schema_version'",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"5"
);
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
schema_before
);
for (table, expected) in tables.into_iter().zip(before) {
assert_eq!(
rows(
&conn,
&format!("SELECT rowid, * FROM {table} ORDER BY rowid")
),
expected,
"{table} changed while repairing stale version {stale_version}"
);
}
assert_identity_alignment(&conn);
assert_eq!(search_snapshot(&path, "quartz"), search_before);
}
}
#[test]
fn stale_legacy_version_does_not_complete_partial_v5_index() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
drop(legacy_db(&path));
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
conn.execute_batch(
"UPDATE session_search_meta SET value='3' WHERE key='schema_version';
DROP TABLE session_messages_current_search_fts;",
)
.unwrap();
let schema_before = rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name");
let messages_before = rows(
&conn,
"SELECT rowid, * FROM session_messages_search ORDER BY rowid",
);
drop(conn);
assert!(init_db(&path)
.unwrap_err()
.to_string()
.contains("unsupported search FTS shape: session_messages_current_search_fts"));
let conn = open_db(&path).unwrap();
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
schema_before
);
assert_eq!(
rows(
&conn,
"SELECT rowid, * FROM session_messages_search ORDER BY rowid"
),
messages_before
);
assert_eq!(
conn.query_row(
"SELECT value FROM session_search_meta WHERE key='schema_version'",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"3"
);
}
#[test]
#[ignore = "manual synthetic v4-to-v5 migration evidence for #1156"]
fn benchmark_v4_to_v5_search_projection_rebuild() {
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
fn live_sqlite_bytes(path: &Path) -> u64 {
[
path.to_path_buf(),
PathBuf::from(format!("{}-wal", path.display())),
PathBuf::from(format!("{}-shm", path.display())),
]
.into_iter()
.filter_map(|file| std::fs::metadata(file).ok().map(|metadata| metadata.len()))
.sum()
}
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
let mut conn = v4_db(&path);
let now = Utc::now().to_rfc3339();
let transaction = conn.transaction().unwrap();
{
let mut ordinary = transaction
.prepare(
"INSERT INTO session_messages_search
(search_rowid, session_id, message_id, message_index, role, content,
compressed, created_at)
VALUES (?1, 'v4', ?2, ?3, 'user', ?4, 0, ?5)",
)
.unwrap();
let mut fts = transaction
.prepare(
"INSERT INTO session_messages_search_fts
(rowid, session_id, message_id, message_index, role, content)
VALUES (?1, 'v4', ?2, ?3, 'user', ?4)",
)
.unwrap();
for index in 1..=5_000i64 {
let rowid = 100 + index;
let message_id = format!("migration-{index}");
let content = if index % 97 == 0 {
format!("synthetic migration item {index}: 我们要压缩上下文")
} else {
format!("synthetic migration item {index}: stable payload")
};
ordinary
.execute(params![rowid, message_id, index, content, now])
.unwrap();
fts.execute(params![rowid, message_id, index, content])
.unwrap();
}
}
transaction.commit().unwrap();
let before_bytes = live_sqlite_bytes(&path);
drop(conn);
let migration_done = Arc::new(AtomicBool::new(false));
let migration_done_worker = migration_done.clone();
let migration_path = path.clone();
let started = Instant::now();
let migration = std::thread::spawn(move || {
let result = init_db(&migration_path);
migration_done_worker.store(true, Ordering::Release);
result
});
let mut sampled_peak_bytes = before_bytes;
while !migration_done.load(Ordering::Acquire) {
sampled_peak_bytes = sampled_peak_bytes.max(live_sqlite_bytes(&path));
std::thread::sleep(Duration::from_micros(100));
}
migration.join().unwrap().unwrap();
let rebuild_us = started.elapsed().as_micros();
let live_bytes_after_commit = live_sqlite_bytes(&path);
sampled_peak_bytes = sampled_peak_bytes.max(live_bytes_after_commit);
let main_db_bytes = std::fs::metadata(&path).unwrap().len();
let page = search_session_messages_db(&path, "v4", "压缩", usize::MAX, &[], 20).unwrap();
assert_eq!(page.matches.len(), 20);
assert!(page
.matches
.iter()
.all(|hit| hit.match_source == "fts_cjk_bigram"));
println!(
"BENCH migration messages=5001 rebuild_us={rebuild_us} \
before_live_bytes={before_bytes} after_live_bytes={live_bytes_after_commit} \
main_db_bytes={main_db_bytes} sampled_peak_bytes={sampled_peak_bytes} \
peak_sampling_interval_us=100"
);
}
#[test]
fn v4_migration_failure_rolls_back_projection_revision_column_and_version() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
let conn = v4_db(&path);
conn.execute_batch(
"CREATE TRIGGER reject_v5_version BEFORE UPDATE ON session_search_meta BEGIN
SELECT RAISE(ABORT, 'injected v5 publication failure'); END;",
)
.unwrap();
let schema_before = rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name");
let ordinary_before = rows(&conn, "SELECT rowid, * FROM session_messages_search");
let fts_before = rows(&conn, "SELECT rowid, * FROM session_messages_search_fts");
assert!(init_db(&path)
.unwrap_err()
.to_string()
.contains("injected v5 publication failure"));
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
schema_before
);
assert_eq!(
rows(&conn, "SELECT rowid, * FROM session_messages_search"),
ordinary_before
);
assert_eq!(
rows(&conn, "SELECT rowid, * FROM session_messages_search_fts"),
fts_before
);
assert_eq!(
conn.query_row(
"SELECT value FROM session_search_meta WHERE key='schema_version'",
[],
|row| row.get::<_, String>(0)
)
.unwrap(),
"4"
);
conn.execute_batch("DROP TRIGGER reject_v5_version")
.unwrap();
init_db(&path).unwrap();
assert_identity_alignment(&conn);
}
#[test]
fn migration_failure_rolls_back_tables_fts_objects_and_version() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
let conn = legacy_db(&path);
conn.execute_batch(
"CREATE TRIGGER reject_version BEFORE UPDATE ON session_search_meta BEGIN
SELECT RAISE(ABORT, 'injected publication failure'); END;",
)
.unwrap();
let schema_before = rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name");
let fts_before = rows(
&conn,
"SELECT rowid, * FROM session_messages_search_fts ORDER BY rowid",
);
let normal_before = rows(
&conn,
"SELECT rowid, * FROM session_messages_search ORDER BY rowid",
);
assert!(init_db(&path)
.unwrap_err()
.to_string()
.contains("injected publication failure"));
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
schema_before
);
assert_eq!(
rows(
&conn,
"SELECT rowid, * FROM session_messages_search_fts ORDER BY rowid"
),
fts_before
);
assert_eq!(
rows(
&conn,
"SELECT rowid, * FROM session_messages_search ORDER BY rowid"
),
normal_before
);
assert!(!search_db(&path, "quartz", 10).unwrap().is_empty());
conn.execute_batch("DROP TRIGGER reject_version").unwrap();
init_db(&path).unwrap();
assert_identity_alignment(&conn);
}
#[test]
fn unsupported_version_and_shape_fail_without_republishing_schema() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
let conn = legacy_db(&path);
conn.execute("UPDATE session_search_meta SET value='99'", [])
.unwrap();
let before = rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name");
assert!(init_db(&path).unwrap_err().to_string().contains("99"));
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
before
);
assert_eq!(
rows(&conn, "SELECT value FROM session_search_meta"),
vec![vec![Value::Text("99".into())]]
);
conn.execute("UPDATE session_search_meta SET value='4'", [])
.unwrap();
assert!(init_db(&path).unwrap_err().to_string().contains("shape"));
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
before
);
}
fn assert_shape_rejected_without_mutation(path: &Path, conn: &Connection, reason: &str) {
let snapshot = || {
[
"SELECT name, sql FROM sqlite_schema ORDER BY name",
"SELECT * FROM session_search_meta ORDER BY key",
"SELECT rowid, * FROM sessions_search ORDER BY rowid",
"SELECT rowid, * FROM session_messages_search ORDER BY rowid",
"SELECT rowid, * FROM sessions_search_fts ORDER BY rowid",
"SELECT rowid, * FROM session_messages_search_fts ORDER BY rowid",
"SELECT rowid, * FROM session_messages_current_search_fts ORDER BY rowid",
]
.map(|sql| rows(conn, sql))
};
let before = snapshot();
let error = init_db(path).unwrap_err().to_string();
assert!(error.contains(reason), "{error}");
assert_eq!(
snapshot(),
before,
"rejection must preserve schema, version and payloads"
);
}
#[test]
fn v5_rejects_changed_fts_indexing_tokenizer_and_content_mode() {
for definition in [
"session_id, title, summary",
"session_id UNINDEXED, title, summary, tokenize='porter'",
"session_id UNINDEXED, title, summary, content=''",
] {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
upsert_session_db(&path, &fixture("fts-shape"), None).unwrap();
let conn = open_db(&path).unwrap();
conn.execute_batch(&format!("DROP TABLE sessions_search_fts;
CREATE VIRTUAL TABLE sessions_search_fts USING fts5({definition});
INSERT INTO sessions_search_fts(rowid, session_id, title, summary)
SELECT search_rowid, session_id, title, COALESCE(summary, '') FROM sessions_search;"))
.unwrap();
assert_shape_rejected_without_mutation(&path, &conn, "unsupported search FTS shape");
}
}
#[test]
fn v5_rejects_missing_unindexed_or_retokenized_message_projections() {
for definition in [
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content",
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content, content_cjk_bigrams",
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content, content_cjk_bigrams UNINDEXED, content_literal_trigrams",
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content, content_cjk_bigrams, content_literal_trigrams UNINDEXED",
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content, content_cjk_bigrams, content_literal_trigrams,
tokenize='porter'",
] {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
upsert_session_db(&path, &fixture("message-fts-shape"), None).unwrap();
let conn = open_db(&path).unwrap();
conn.execute_batch(&format!(
"DROP TABLE session_messages_current_search_fts;
CREATE VIRTUAL TABLE session_messages_current_search_fts USING fts5({definition});"
))
.unwrap();
assert_shape_rejected_without_mutation(&path, &conn, "unsupported search FTS shape");
}
}
#[test]
fn v5_rejects_changed_or_missing_canonical_message_fts() {
for definition in [
Some(
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content UNINDEXED",
),
Some(
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content, content_cjk_bigrams",
),
Some(
"session_id UNINDEXED, message_id UNINDEXED, message_index UNINDEXED,
role UNINDEXED, content, tokenize='porter'",
),
None,
] {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
upsert_session_db(&path, &fixture("canonical-message-fts-shape"), None).unwrap();
let conn = open_db(&path).unwrap();
conn.execute_batch("DROP TABLE session_messages_search_fts;")
.unwrap();
if let Some(definition) = definition {
conn.execute_batch(&format!(
"CREATE VIRTUAL TABLE session_messages_search_fts USING fts5({definition});"
))
.unwrap();
}
let before = rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name");
let error = init_db(&path).unwrap_err().to_string();
assert!(error.contains("unsupported search FTS shape"), "{error}");
assert_eq!(
rows(&conn, "SELECT name, sql FROM sqlite_schema ORDER BY name"),
before,
"shape rejection must not recreate or mutate the canonical FTS"
);
}
}
#[test]
fn v5_rejects_non_alias_primary_key_and_missing_or_invalid_identity_uniqueness() {
for (table, from, to, extra_sql, reason) in [
(
"sessions_search",
"INTEGER PRIMARY KEY",
"INTEGER PRIMARY KEY DESC",
"",
"must alias rowid",
),
(
"sessions_search",
"TEXT NOT NULL UNIQUE",
"TEXT NOT NULL",
"",
"missing search identity constraint",
),
(
"sessions_search",
"TEXT NOT NULL UNIQUE",
"TEXT NOT NULL",
"CREATE UNIQUE INDEX partial_identity ON sessions_search(session_id) WHERE pinned=1;",
"missing search identity constraint",
),
(
"session_messages_search",
"UNIQUE(session_id, message_id)",
"UNIQUE(message_id)",
"",
"missing search identity constraint",
),
] {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
upsert_session_db(&path, &fixture("identity-shape"), None).unwrap();
let conn = open_db(&path).unwrap();
let original: String = conn
.query_row(
"SELECT sql FROM sqlite_schema WHERE name=?1",
[table],
|row| row.get(0),
)
.unwrap();
let changed = original.replace(from, to);
assert_ne!(
original, changed,
"fixture must alter the actual v5 definition"
);
conn.execute_batch(&format!(
"CREATE TEMP TABLE saved_rows AS SELECT * FROM {table};
DROP TABLE {table}; {changed}; INSERT INTO {table} SELECT * FROM saved_rows;
DROP TABLE saved_rows; {extra_sql}"
))
.unwrap();
assert_shape_rejected_without_mutation(&path, &conn, reason);
}
}
fn install_audit(conn: &Connection) {
conn.execute_batch("CREATE TABLE write_audit (table_name TEXT, operation TEXT)")
.unwrap();
for table in ["sessions_search", "session_messages_search"] {
for operation in ["INSERT", "UPDATE", "DELETE"] {
conn.execute_batch(&format!(
"CREATE TRIGGER audit_{table}_{operation} AFTER {operation} ON {table}
BEGIN INSERT INTO write_audit VALUES ('{table}', '{operation}'); END;"
))
.unwrap();
}
}
}
fn apply_checked(path: &Path, session: &Session, expected: delta::Changes) {
let conn = open_db(path).unwrap();
conn.execute("DELETE FROM write_audit", []).unwrap();
let before_ids = identity_rows(&conn);
assert_eq!(upsert_session_db(path, session, None).unwrap(), expected);
for (identity, rowid) in identity_rows(&conn) {
if let Some(previous) = before_ids.get(&identity) {
assert_eq!(*previous, rowid);
}
}
for (table, count) in [
("sessions_search", expected.sessions),
("session_messages_search", expected.messages),
] {
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM write_audit WHERE table_name=?1",
[table],
|row| row.get::<_, i64>(0)
)
.unwrap(),
count as i64
);
}
assert_identity_alignment(&conn);
}
#[test]
fn differential_writes_cover_identical_metadata_append_edit_order_role_and_delete() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
install_audit(&conn);
let mut session = fixture("delta");
apply_checked(
&path,
&session,
delta::Changes {
sessions: 1,
messages: 3,
session_fts: 1,
message_fts: 3,
},
);
apply_checked(&path, &session, delta::Changes::default());
session.title = "updated beacon".into();
apply_checked(
&path,
&session,
delta::Changes {
sessions: 1,
session_fts: 1,
..Default::default()
},
);
session.conversation_summary = Some(ConversationSummary::new("summary recollection", 1, 2));
apply_checked(
&path,
&session,
delta::Changes {
sessions: 1,
session_fts: 1,
..Default::default()
},
);
session.kind = SessionKind::Child;
session.root_session_id = "root".into();
session.parent_session_id = Some("parent".into());
session.pinned = true;
session.updated_at += Duration::seconds(1);
apply_checked(
&path,
&session,
delta::Changes {
sessions: 1,
..Default::default()
},
);
let metadata = search_db(&path, "beacon", 10).unwrap();
assert!(metadata[0].pinned);
assert_eq!(metadata[0].parent_session_id.as_deref(), Some("parent"));
session.messages[0].compressed = true;
apply_checked(
&path,
&session,
delta::Changes {
messages: 1,
..Default::default()
},
);
session.messages[0].created_at += Duration::seconds(1);
apply_checked(
&path,
&session,
delta::Changes {
messages: 1,
..Default::default()
},
);
let cache = read_compressed_cache_db(&path, &session.id, 0, 10, 100).unwrap();
assert_eq!(cache.messages[0].created_at, session.messages[0].created_at);
let mut appended = Message::assistant("appended zircon", None);
appended.id = "m3".into();
session.messages.push(appended);
apply_checked(
&path,
&session,
delta::Changes {
messages: 1,
message_fts: 1,
..Default::default()
},
);
session.messages[1].content = "amended sapphire".into();
apply_checked(
&path,
&session,
delta::Changes {
messages: 1,
message_fts: 1,
..Default::default()
},
);
session.messages[2].role = Role::Tool;
apply_checked(
&path,
&session,
delta::Changes {
messages: 1,
message_fts: 1,
..Default::default()
},
);
session.messages.swap(0, 2);
apply_checked(
&path,
&session,
delta::Changes {
messages: 2,
message_fts: 2,
..Default::default()
},
);
let hits = search_db(&path, "quartz", 10).unwrap();
assert!(hits
.iter()
.any(|hit| hit.message_id.as_deref() == Some("m2")
&& hit.message_index == Some(0)
&& hit.role.as_deref() == Some("tool")));
session.messages.remove(1);
apply_checked(
&path,
&session,
delta::Changes {
messages: 3,
message_fts: 3,
..Default::default()
},
);
assert!(search_db(&path, "sapphire", 10).unwrap().is_empty());
session.messages.clear();
apply_checked(
&path,
&session,
delta::Changes {
messages: 3,
message_fts: 3,
..Default::default()
},
);
assert!(search_db(&path, "quartz", 10).unwrap().is_empty());
apply_checked(&path, &session, delta::Changes::default());
}
#[test]
fn unchanged_projection_repairs_missing_stale_and_null_fts_payloads() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
let session = fixture("repair");
upsert_session_db(&path, &session, None).unwrap();
let conn = open_db(&path).unwrap();
install_audit(&conn);
conn.execute_batch(
"UPDATE sessions_search_fts SET title='corrupt', summary=NULL;
DELETE FROM session_messages_current_search_fts WHERE rowid=1;
UPDATE session_messages_current_search_fts
SET content='stale payload', content_cjk_bigrams='错误 投影',
content_literal_trigrams='stale projection', role=NULL, message_index=99
WHERE rowid=2;
UPDATE session_messages_search_fts SET content='stale canonical payload'
WHERE rowid=3;",
)
.unwrap();
apply_checked(
&path,
&session,
delta::Changes {
session_fts: 1,
message_fts: 3,
..Default::default()
},
);
assert_eq!(search_db(&path, "quartz", 10).unwrap().len(), 3);
assert!(search_db(&path, "corrupt", 10).unwrap().is_empty());
assert!(search_db(&path, "stale", 10).unwrap().is_empty());
conn.execute_batch(
"DELETE FROM sessions_search_fts;
DELETE FROM session_messages_search_fts;
DELETE FROM session_messages_current_search_fts;",
)
.unwrap();
apply_checked(
&path,
&session,
delta::Changes {
session_fts: 1,
message_fts: 3,
..Default::default()
},
);
apply_checked(&path, &session, delta::Changes::default());
}
#[test]
fn duplicate_ids_and_fts_payload_failure_leave_entire_snapshot_unchanged() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
let session = fixture("rollback");
upsert_session_db(&path, &session, None).unwrap();
let conn = open_db(&path).unwrap();
let normal = rows(
&conn,
"SELECT * FROM session_messages_search ORDER BY search_rowid",
);
let titles = rows(&conn, "SELECT * FROM sessions_search ORDER BY search_rowid");
let mut duplicate = session.clone();
duplicate.title = "must rollback".into();
duplicate.messages.push(duplicate.messages[0].clone());
assert!(upsert_session_db(&path, &duplicate, None).is_err());
assert_eq!(
rows(
&conn,
"SELECT * FROM session_messages_search ORDER BY search_rowid"
),
normal
);
assert_eq!(
rows(&conn, "SELECT * FROM sessions_search ORDER BY search_rowid"),
titles
);
conn.execute_batch(
"ALTER TABLE session_messages_search_fts RENAME TO saved_canonical_fts;
CREATE TABLE session_messages_search_fts (session_id, message_id, message_index, role,
content CHECK(content NOT LIKE '%rejectpayload%'));
INSERT INTO session_messages_search_fts
(rowid, session_id, message_id, message_index, role, content)
SELECT rowid, session_id, message_id, message_index, role, content
FROM saved_canonical_fts;",
)
.unwrap();
let canonical_fts_before = rows(
&conn,
"SELECT rowid, * FROM session_messages_search_fts ORDER BY rowid",
);
let current_fts_before = rows(
&conn,
"SELECT rowid, * FROM session_messages_current_search_fts ORDER BY rowid",
);
let mut changed = session.clone();
changed.title = "must rollback".into();
changed.messages[0].content = "rejectpayload".into();
assert!(upsert_session_db(&path, &changed, None).is_err());
assert_eq!(
rows(
&conn,
"SELECT * FROM session_messages_search ORDER BY search_rowid"
),
normal
);
assert_eq!(
rows(&conn, "SELECT * FROM sessions_search ORDER BY search_rowid"),
titles
);
assert_eq!(
rows(
&conn,
"SELECT rowid, * FROM session_messages_search_fts ORDER BY rowid"
),
canonical_fts_before
);
assert_eq!(
rows(
&conn,
"SELECT rowid, * FROM session_messages_current_search_fts ORDER BY rowid"
),
current_fts_before
);
assert!(search_db(&path, "beacon", 1).unwrap()[0].session_title == session.title);
conn.execute_batch(
"DROP TABLE session_messages_search_fts;
ALTER TABLE saved_canonical_fts RENAME TO session_messages_search_fts;",
)
.unwrap();
assert_eq!(search_db(&path, "quartz", 10).unwrap().len(), 3);
}
#[test]
fn explicit_ids_survive_vacuum_holes_restart_and_later_delta_writes() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
for id in ["hole", "kept", "unrelated"] {
upsert_session_db(&path, &fixture(id), None).unwrap();
}
delete_session_db(&path, "hole", None).unwrap();
let conn = open_db(&path).unwrap();
let before = identity_rows(&conn);
let session_ids = rows(
&conn,
"SELECT search_rowid, session_id FROM sessions_search ORDER BY search_rowid",
);
conn.execute_batch("VACUUM").unwrap();
drop(conn);
init_db(&path).unwrap();
let conn = open_db(&path).unwrap();
assert_eq!(identity_rows(&conn), before);
assert_eq!(
rows(
&conn,
"SELECT search_rowid, session_id FROM sessions_search ORDER BY search_rowid"
),
session_ids
);
assert_identity_alignment(&conn);
let unrelated = rows(
&conn,
"SELECT * FROM session_messages_search WHERE session_id='unrelated' ORDER BY search_rowid",
);
let mut updated = fixture("kept");
updated.messages[0].content = "edited postvacuum".into();
updated.messages.remove(1);
let mut new = Message::user("added postvacuum");
new.id = "new".into();
updated.messages.push(new);
upsert_session_db(&path, &updated, None).unwrap();
for (identity, rowid) in identity_rows(&conn) {
if let Some(previous) = before.get(&identity) {
assert_eq!(*previous, rowid);
}
}
assert_identity_alignment(&conn);
assert_eq!(rows(&conn, "SELECT * FROM session_messages_search WHERE session_id='unrelated' ORDER BY search_rowid"), unrelated);
assert_eq!(search_db(&path, "postvacuum", 10).unwrap().len(), 2);
}
#[test]
fn deletion_and_pruning_use_keyed_fts_access_among_unrelated_sessions() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("search.db");
init_db(&path).unwrap();
for index in 0..128 {
upsert_session_db(&path, &fixture(&format!("s{index}")), None).unwrap();
}
let conn = open_db(&path).unwrap();
for sql in [
delta::DELETE_SESSION_FTS,
delta::DELETE_MESSAGE_FTS,
delta::DELETE_CURRENT_MESSAGE_FTS,
] {
let plan = conn
.prepare(&format!("EXPLAIN QUERY PLAN {sql}"))
.unwrap()
.query_map(["s0"], |row| row.get::<_, String>(3))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
println!("Production keyed FTS deletion plan: {plan:?}");
assert!(
plan.iter()
.any(|detail| detail.contains("VIRTUAL TABLE INDEX") && detail.contains('=')),
"{plan:?}"
);
assert!(
plan.iter().any(|detail| detail.contains("SEARCH")
&& detail.contains("INDEX")
&& detail.contains("session_id=?")),
"{plan:?}"
);
}
let unrelated = rows(&conn, "SELECT * FROM session_messages_search WHERE session_id NOT IN ('s0', 's1') ORDER BY search_rowid");
delete_session_db(&path, "s0", None).unwrap();
conn.execute(
"UPDATE sessions_search SET updated_at=?1 WHERE session_id='s1'",
[(Utc::now() - Duration::days(11)).to_rfc3339()],
)
.unwrap();
assert_eq!(prune_stale_sessions_db(&path).unwrap(), 1);
assert_identity_alignment(&conn);
assert_eq!(
rows(
&conn,
"SELECT * FROM session_messages_search ORDER BY search_rowid"
),
unrelated
);
assert_eq!(
conn.query_row("SELECT COUNT(*) FROM sessions_search", [], |row| row
.get::<_, i64>(0))
.unwrap(),
126
);
}