use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior};
use crate::conversation::{
new_conversation_id, session_plan_dir, ConversationRecord, ConversationSummary,
ConversationTurn,
};
const DB_FILE: &str = "conversations.db";
const LEGACY_JSON_DIR: &str = "conversations";
const LEGACY_BACKUP_DIR: &str = "conversations.imported";
const NONCE_FILE: &str = "install-nonce";
const IDENTITY_PEM_FILE: &str = "identity.pem";
const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
const TURN_ENCODING_V1_PREFIX: &[u8] = b"newt-turn:v1";
const TURN_ENCODING_VERSION_CURRENT: i64 = 1;
const GENESIS_PREFIX: &[u8] = b"newt-turn-chain-genesis:v1";
#[derive(Debug, Clone)]
pub struct ConversationStore {
conn: Arc<Mutex<Connection>>,
workspace: PathBuf,
workspace_id: String,
writer_fingerprint: String,
max_per_workspace: usize,
wal_fallback: Option<String>,
claim_clock: fn() -> i64,
}
impl ConversationStore {
pub fn new(
root: impl AsRef<Path>,
workspace: impl AsRef<Path>,
max_per_workspace: usize,
) -> anyhow::Result<Self> {
let root = root.as_ref().to_path_buf();
let workspace = std::fs::canonicalize(workspace.as_ref())?;
let workspace_id = crate::workspace_key::workspace_key_v2(&workspace)?;
std::fs::create_dir_all(&root)?;
let writer_fingerprint = resolve_writer_fingerprint(&root)?;
let conn = Connection::open(root.join(DB_FILE))?;
conn.busy_timeout(BUSY_TIMEOUT)?;
conn.pragma_update(None, "foreign_keys", "ON")?;
let wal_fallback = {
let mut attempt = 0u32;
loop {
match apply_journal_mode(&conn)
.and_then(|fb| create_schema(&conn).map(|()| fb))
.and_then(|fb| reconcile_schema(&conn).map(|()| fb))
.and_then(|fb| create_fts_index(&conn).map(|()| fb))
{
Ok(fb) => break fb,
Err(e)
if attempt < 20
&& e.to_string().to_ascii_lowercase().contains("locked") =>
{
attempt += 1;
std::thread::sleep(std::time::Duration::from_millis(
25 * u64::from(attempt.min(4)),
));
}
Err(e) => return Err(e),
}
}
};
import_legacy_json(&conn, &root, &writer_fingerprint)?;
migrate_workspace_key(&conn, &workspace, &workspace_id)?;
Ok(Self {
conn: Arc::new(Mutex::new(conn)),
workspace,
workspace_id,
writer_fingerprint,
max_per_workspace,
wal_fallback,
claim_clock: now_claim_nanos,
})
}
#[deprecated(
since = "0.6.8",
note = "v1 keying is path-fragile; use `newt_core::workspace_key_v2` \
(17.2). This stays only for the UUIDv5→v2 row migration and \
legacy-import dir names."
)]
pub fn workspace_id_for_path(path: impl AsRef<Path>) -> anyhow::Result<String> {
let canonical = std::fs::canonicalize(path.as_ref())?;
let normalized = canonical.to_string_lossy().replace('\\', "/");
Ok(uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, normalized.as_bytes()).to_string())
}
pub fn wal_fallback_notice(&self) -> Option<&str> {
self.wal_fallback.as_deref()
}
pub fn writer_fingerprint(&self) -> &str {
&self.writer_fingerprint
}
pub fn create(&self, title: &str, persona: Option<&str>) -> anyhow::Result<String> {
let id = new_conversation_id();
self.create_with_id(&id, title, persona)?;
Ok(id)
}
pub fn create_with_id(
&self,
id: &str,
title: &str,
persona: Option<&str>,
) -> anyhow::Result<()> {
validate_record_id(id)?;
let now = (self.claim_clock)();
{
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let foreign: Option<String> = tx
.query_row(
"SELECT workspace_key FROM conversations WHERE id = ?1",
rusqlite::params![id],
|row| row.get(0),
)
.optional()?;
if let Some(owner) = foreign {
if owner != self.workspace_id {
anyhow::bail!(
"conversation id `{id}` already exists in another workspace \
(key {owner}); refusing to overwrite across the workspace fence"
);
}
}
let tick = next_tick(&tx, &self.writer_fingerprint)?;
tx.execute(
"INSERT OR REPLACE INTO conversations
(id, title, workspace_path, workspace_key, persona, end_reason,
writer_fingerprint, activity_tick, tip_hash,
started_at_claim, updated_at_claim)
VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6, ?7, ?8, ?9, ?9)",
rusqlite::params![
id,
title.trim(),
self.workspace.to_string_lossy(),
self.workspace_id,
persona,
self.writer_fingerprint,
tick,
genesis_hash(id, &self.writer_fingerprint),
now,
],
)?;
tx.commit()?;
}
self.prune_to_cap()?;
Ok(())
}
pub fn exists(&self, id: &str) -> anyhow::Result<bool> {
let conn = self.lock_conn();
Ok(conn
.query_row(
"SELECT 1 FROM conversations WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
|_| Ok(()),
)
.optional()?
.is_some())
}
pub fn append_turn(&self, id: &str, user: &str, assistant: &str) -> anyhow::Result<()> {
self.append_turn_full(id, user, assistant, &[], &[], None, None)
}
#[allow(clippy::too_many_arguments)]
pub fn append_turn_full(
&self,
id: &str,
user: &str,
assistant: &str,
events: &[crate::ToolEvent],
phantom_reaches: &[crate::PhantomReach],
tokens_in: Option<u32>,
tokens_out: Option<u32>,
) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let now = (self.claim_clock)();
let events_json = serde_json::to_string(events)?;
let phantom_reaches_json = serde_json::to_string(phantom_reaches)?;
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let tick = next_tick(&tx, &self.writer_fingerprint)?;
let prev_hash = match last_turn(&tx, &id, &self.writer_fingerprint)? {
Some(prev) => prev.content_hash()?,
None => genesis_hash(&id, &self.writer_fingerprint),
};
let row = TurnRow {
conversation_id: id.clone(),
writer_fingerprint: self.writer_fingerprint.clone(),
seq: tick,
prev_hash,
user: user.to_string(),
assistant: assistant.to_string(),
events: events_json,
tokens_in: tokens_in.map(i64::from),
tokens_out: tokens_out.map(i64::from),
ts_claim: now,
encoding_version: TURN_ENCODING_VERSION_CURRENT,
};
insert_turn_row(&tx, &row, &phantom_reaches_json)?;
tx.execute(
"UPDATE conversations
SET writer_fingerprint = ?2, activity_tick = ?3, tip_hash = ?4,
updated_at_claim = ?5
WHERE id = ?1",
rusqlite::params![id, self.writer_fingerprint, tick, row.content_hash()?, now],
)?;
tx.commit()?;
Ok(())
}
pub fn load(&self, id: &str) -> anyhow::Result<ConversationRecord> {
let id = self.resolve_id(id)?;
let conn = self.lock_conn();
let (mut record, scratchpad_json, plan_json) = conn
.query_row(
"SELECT id, title, workspace_path, workspace_key, persona,
started_at_claim, updated_at_claim, scratchpad, plan
FROM conversations
WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
|row| {
Ok((
ConversationRecord {
id: row.get(0)?,
title: row.get(1)?,
workspace: row.get(2)?,
workspace_id: row.get(3)?,
persona: row.get(4)?,
turns: Vec::new(),
scratchpad: std::collections::BTreeMap::new(),
plan: crate::PlanSnapshot::default(),
created_at_unix_nanos: claim_to_u128(row.get(5)?),
updated_at_unix_nanos: claim_to_u128(row.get(6)?),
},
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
))
},
)
.optional()?
.ok_or_else(|| anyhow::anyhow!("conversation `{id}` not found"))?;
record.scratchpad = serde_json::from_str(&scratchpad_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: scratchpad column is not valid <state> JSON \
({e}); refusing to load garbage"
)
})?;
record.plan = serde_json::from_str(&plan_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: plan column is not valid <plan> snapshot JSON \
({e}); refusing to load garbage"
)
})?;
let mut stmt = conn.prepare(
"SELECT user, assistant, events, tokens_in, tokens_out, phantom_reaches FROM turns
WHERE conversation_id = ?1
ORDER BY seq ASC, writer_fingerprint ASC",
)?;
let turns = stmt.query_map([&id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<i64>>(3)?,
row.get::<_, Option<i64>>(4)?,
row.get::<_, String>(5)?,
))
})?;
for turn in turns {
let (user, assistant, events_json, tokens_in, tokens_out, phantom_reaches_json) = turn?;
let events: Vec<crate::ToolEvent> =
serde_json::from_str(&events_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: turn events column is not valid tool-event \
JSON ({e}); refusing to load garbage"
)
})?;
let phantom_reaches: Vec<crate::PhantomReach> =
serde_json::from_str(&phantom_reaches_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: turn phantom_reaches column is not valid \
phantom-reach JSON ({e}); refusing to load garbage"
)
})?;
record.turns.push(ConversationTurn {
user,
assistant,
events,
phantom_reaches,
tokens_in: tokens_from_sql(tokens_in)?,
tokens_out: tokens_from_sql(tokens_out)?,
});
}
Ok(record)
}
pub fn load_turn(&self, id: &str, seq: i64) -> anyhow::Result<Option<ConversationTurn>> {
let id = match self.resolve_id(id) {
Ok(id) => id,
Err(_) => return Ok(None),
};
let conn = self.lock_conn();
let row = conn
.query_row(
"SELECT t.user, t.assistant, t.events, t.tokens_in, t.tokens_out, t.phantom_reaches
FROM turns t
JOIN conversations c
ON c.id = t.conversation_id AND c.workspace_key = ?3
WHERE t.conversation_id = ?1 AND t.seq = ?2",
rusqlite::params![id, seq, self.workspace_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<i64>>(3)?,
row.get::<_, Option<i64>>(4)?,
row.get::<_, String>(5)?,
))
},
)
.optional()?;
let Some((user, assistant, events_json, tokens_in, tokens_out, phantom_reaches_json)) = row
else {
return Ok(None);
};
let events: Vec<crate::ToolEvent> = serde_json::from_str(&events_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: turn events column is not valid tool-event \
JSON ({e}); refusing to load garbage"
)
})?;
let phantom_reaches: Vec<crate::PhantomReach> = serde_json::from_str(&phantom_reaches_json)
.map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: turn phantom_reaches column is not valid \
phantom-reach JSON ({e}); refusing to load garbage"
)
})?;
Ok(Some(ConversationTurn {
user,
assistant,
events,
phantom_reaches,
tokens_in: tokens_from_sql(tokens_in)?,
tokens_out: tokens_from_sql(tokens_out)?,
}))
}
pub fn list(&self) -> anyhow::Result<Vec<ConversationSummary>> {
let conn = self.lock_conn();
let mut stmt = conn.prepare(
"SELECT c.id, c.title, c.persona, c.updated_at_claim,
(SELECT COUNT(*) FROM turns t WHERE t.conversation_id = c.id)
FROM conversations c
WHERE c.workspace_key = ?1
ORDER BY c.activity_tick ASC, c.id ASC",
)?;
let rows = stmt.query_map([&self.workspace_id], |row| {
Ok(ConversationSummary {
id: row.get(0)?,
title: row.get(1)?,
persona: row.get(2)?,
updated_at_unix_nanos: claim_to_u128(row.get(3)?),
turn_count: row.get::<_, i64>(4)?.max(0) as usize,
})
})?;
let mut summaries = Vec::new();
for row in rows {
summaries.push(row?);
}
Ok(summaries)
}
pub fn latest_open(&self) -> anyhow::Result<Option<ConversationSummary>> {
let conn = self.lock_conn();
conn.query_row(
"SELECT c.id, c.title, c.persona, c.updated_at_claim,
(SELECT COUNT(*) FROM turns t WHERE t.conversation_id = c.id)
FROM conversations c
WHERE c.workspace_key = ?1 AND c.end_reason IS NULL
ORDER BY c.activity_tick DESC, c.id DESC
LIMIT 1",
[&self.workspace_id],
|row| {
Ok(ConversationSummary {
id: row.get(0)?,
title: row.get(1)?,
persona: row.get(2)?,
updated_at_unix_nanos: claim_to_u128(row.get(3)?),
turn_count: row.get::<_, i64>(4)?.max(0) as usize,
})
},
)
.optional()
.map_err(Into::into)
}
pub fn end_conversation(&self, id: &str, reason: &str) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let now = (self.claim_clock)();
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET end_reason = ?2, updated_at_claim = ?3
WHERE id = ?1 AND workspace_key = ?4",
rusqlite::params![id, reason.trim(), now, self.workspace_id],
)?;
Ok(())
}
pub fn rename(&self, id: &str, title: &str) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let now = (self.claim_clock)();
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET title = ?2, updated_at_claim = ?3 WHERE id = ?1",
rusqlite::params![id, title.trim(), now],
)?;
Ok(())
}
pub fn update_scratchpad(
&self,
id: &str,
scratchpad: &std::collections::BTreeMap<String, String>,
) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let json = serde_json::to_string(scratchpad)?;
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET scratchpad = ?2 WHERE id = ?1 AND workspace_key = ?3",
rusqlite::params![id, json, self.workspace_id],
)?;
Ok(())
}
pub fn update_plan_snapshot(&self, id: &str, plan: &crate::PlanSnapshot) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let json = serde_json::to_string(plan)?;
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET plan = ?2 WHERE id = ?1 AND workspace_key = ?3",
rusqlite::params![id, json, self.workspace_id],
)?;
Ok(())
}
pub fn delete(&self, id: &str) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
{
let conn = self.lock_conn();
conn.execute(
"DELETE FROM conversations WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
)?;
}
let plan_dir = self.workspace.join(session_plan_dir(&id));
let _ = std::fs::remove_dir_all(plan_dir);
Ok(())
}
pub fn resolve_id(&self, id_or_prefix: &str) -> anyhow::Result<String> {
validate_record_id(id_or_prefix)?;
let conn = self.lock_conn();
let exact = conn
.query_row(
"SELECT id FROM conversations WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id_or_prefix, self.workspace_id],
|row| row.get::<_, String>(0),
)
.optional()?;
if let Some(id) = exact {
return Ok(id);
}
let mut stmt = conn.prepare(
"SELECT id FROM conversations
WHERE workspace_key = ?1 AND substr(id, 1, length(?2)) = ?2
ORDER BY id ASC",
)?;
let matches = stmt
.query_map(rusqlite::params![self.workspace_id, id_or_prefix], |row| {
row.get::<_, String>(0)
})?
.collect::<Result<Vec<_>, _>>()?;
match matches.as_slice() {
[id] => Ok(id.clone()),
[] => anyhow::bail!("conversation `{id_or_prefix}` not found"),
many => anyhow::bail!(
"ambiguous conversation id prefix `{}`; matches: {}",
id_or_prefix,
many.join(", ")
),
}
}
pub fn verify_chain(&self, id: &str) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let conn = self.lock_conn();
let (tip, tip_writer): (String, String) = conn.query_row(
"SELECT tip_hash, writer_fingerprint FROM conversations WHERE id = ?1",
[&id],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
let mut stmt = conn.prepare(
"SELECT conversation_id, writer_fingerprint, seq, prev_hash, user, assistant,
events, tokens_in, tokens_out, ts_claim, encoding_version
FROM turns
WHERE conversation_id = ?1
ORDER BY writer_fingerprint ASC, seq ASC",
)?;
let rows = stmt
.query_map([&id], turn_row_from_sql)?
.collect::<Result<Vec<_>, _>>()?;
let mut prev: Option<&TurnRow> = None;
for row in &rows {
let same_writer = prev.is_some_and(|p| p.writer_fingerprint == row.writer_fingerprint);
if same_writer {
let p = prev.expect("same_writer implies prev");
if row.seq <= p.seq {
anyhow::bail!(
"chain violation in `{id}`: seq {} not strictly after {}",
row.seq,
p.seq
);
}
if row.prev_hash != p.content_hash()? {
anyhow::bail!(
"chain violation in `{id}`: turn seq {} does not link to seq {} \
(row tampered or out of order)",
row.seq,
p.seq
);
}
} else {
let genesis = genesis_hash(&id, &row.writer_fingerprint);
if row.prev_hash != genesis {
anyhow::bail!(
"chain violation in `{id}`: first turn of writer {} (seq {}) does \
not link to the genesis hash",
row.writer_fingerprint,
row.seq
);
}
}
prev = Some(row);
}
let expected_tip = match rows.iter().rfind(|r| r.writer_fingerprint == tip_writer) {
Some(row) => row.content_hash()?,
None => genesis_hash(&id, &tip_writer),
};
if tip != expected_tip {
anyhow::bail!("chain violation in `{id}`: stored tip_hash does not match the chain");
}
Ok(())
}
pub fn search(&self, query: &str, limit: usize) -> anyhow::Result<Vec<SearchHit>> {
let fts_query = sanitize_fts5_query(query)?;
let limit = i64::try_from(limit).unwrap_or(i64::MAX);
let conn = self.lock_conn();
let mut stmt = conn.prepare(
"SELECT t.conversation_id, c.title, t.seq,
snippet(turns_fts, -1, '>>>', '<<<', '…', 21),
bm25(turns_fts)
FROM turns_fts
JOIN turns t ON t.rowid = turns_fts.rowid
JOIN conversations c
ON c.id = t.conversation_id AND c.workspace_key = ?2
WHERE turns_fts MATCH ?1
ORDER BY bm25(turns_fts) ASC, t.conversation_id ASC, t.seq ASC
LIMIT ?3",
)?;
let rows = stmt.query_map(
rusqlite::params![fts_query, self.workspace_id, limit],
|row| {
Ok(SearchHit {
conversation_id: row.get(0)?,
title: row.get(1)?,
seq: row.get(2)?,
snippet: row.get(3)?,
rank: row.get(4)?,
})
},
)?;
let mut hits = Vec::new();
for row in rows {
hits.push(row?);
}
Ok(hits)
}
#[doc(hidden)]
pub fn set_claim_clock_for_test(&mut self, clock: fn() -> i64) {
self.claim_clock = clock;
}
fn prune_to_cap(&self) -> anyhow::Result<()> {
if self.max_per_workspace == 0 {
return Ok(());
}
let victims: Vec<String> = {
let conn = self.lock_conn();
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM conversations WHERE workspace_key = ?1",
[&self.workspace_id],
|row| row.get(0),
)?;
let excess = count - self.max_per_workspace as i64;
if excess <= 0 {
return Ok(());
}
let mut stmt = conn.prepare(
"SELECT id FROM conversations
WHERE workspace_key = ?1
ORDER BY activity_tick ASC, id ASC
LIMIT ?2",
)?;
let ids = stmt
.query_map(rusqlite::params![self.workspace_id, excess], |row| {
row.get::<_, String>(0)
})?
.collect::<Result<Vec<_>, _>>()?;
ids
};
for id in victims {
self.delete(&id)?;
}
Ok(())
}
fn lock_conn(&self) -> std::sync::MutexGuard<'_, Connection> {
self.conn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct SearchHit {
pub conversation_id: String,
pub title: String,
pub seq: i64,
pub snippet: String,
pub rank: f64,
}
#[derive(Debug)]
struct TurnRow {
conversation_id: String,
writer_fingerprint: String,
seq: i64,
prev_hash: String,
user: String,
assistant: String,
events: String,
tokens_in: Option<i64>,
tokens_out: Option<i64>,
ts_claim: i64,
encoding_version: i64,
}
impl TurnRow {
fn content_hash(&self) -> anyhow::Result<String> {
match self.encoding_version {
1 => Ok(blake3::hash(&self.canonical_encoding_v1())
.to_hex()
.to_string()),
other => anyhow::bail!(
"turn (conversation `{}`, writer {}, seq {}) carries encoding_version {other}, \
which this newt does not understand (known: 1) — upgrade newt to verify \
or extend this chain",
self.conversation_id,
self.writer_fingerprint,
self.seq
),
}
}
fn canonical_encoding_v1(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(
64 + self.conversation_id.len()
+ self.writer_fingerprint.len()
+ self.prev_hash.len()
+ self.user.len()
+ self.assistant.len()
+ self.events.len(),
);
out.extend_from_slice(TURN_ENCODING_V1_PREFIX);
for field in [
self.conversation_id.as_bytes(),
self.writer_fingerprint.as_bytes(),
self.prev_hash.as_bytes(),
self.user.as_bytes(),
self.assistant.as_bytes(),
self.events.as_bytes(),
] {
out.extend_from_slice(&(field.len() as u64).to_le_bytes());
out.extend_from_slice(field);
}
out.extend_from_slice(&self.seq.to_le_bytes());
for opt in [self.tokens_in, self.tokens_out] {
match opt {
Some(v) => {
out.push(1);
out.extend_from_slice(&v.to_le_bytes());
}
None => out.push(0),
}
}
out.extend_from_slice(&self.ts_claim.to_le_bytes());
out
}
}
fn turn_row_from_sql(row: &rusqlite::Row<'_>) -> rusqlite::Result<TurnRow> {
Ok(TurnRow {
conversation_id: row.get(0)?,
writer_fingerprint: row.get(1)?,
seq: row.get(2)?,
prev_hash: row.get(3)?,
user: row.get(4)?,
assistant: row.get(5)?,
events: row.get(6)?,
tokens_in: row.get(7)?,
tokens_out: row.get(8)?,
ts_claim: row.get(9)?,
encoding_version: row.get(10)?,
})
}
fn insert_turn_row(
conn: &Connection,
row: &TurnRow,
phantom_reaches_json: &str,
) -> anyhow::Result<()> {
conn.execute(
"INSERT INTO turns
(conversation_id, writer_fingerprint, seq, prev_hash, user, assistant,
events, tokens_in, tokens_out, ts_claim, encoding_version, phantom_reaches)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
rusqlite::params![
row.conversation_id,
row.writer_fingerprint,
row.seq,
row.prev_hash,
row.user,
row.assistant,
row.events,
row.tokens_in,
row.tokens_out,
row.ts_claim,
row.encoding_version,
phantom_reaches_json,
],
)?;
Ok(())
}
fn genesis_hash(conversation_id: &str, writer_fingerprint: &str) -> String {
let mut hasher = blake3::Hasher::new();
hasher.update(GENESIS_PREFIX);
for field in [conversation_id.as_bytes(), writer_fingerprint.as_bytes()] {
hasher.update(&(field.len() as u64).to_le_bytes());
hasher.update(field);
}
hasher.finalize().to_hex().to_string()
}
fn last_turn(
conn: &Connection,
conversation_id: &str,
writer_fingerprint: &str,
) -> anyhow::Result<Option<TurnRow>> {
Ok(conn
.query_row(
"SELECT conversation_id, writer_fingerprint, seq, prev_hash, user, assistant,
events, tokens_in, tokens_out, ts_claim, encoding_version
FROM turns
WHERE conversation_id = ?1 AND writer_fingerprint = ?2
ORDER BY seq DESC
LIMIT 1",
rusqlite::params![conversation_id, writer_fingerprint],
turn_row_from_sql,
)
.optional()?)
}
fn next_tick(conn: &Connection, writer_fingerprint: &str) -> anyhow::Result<i64> {
let bumped = conn.execute(
"UPDATE writer_clock SET last_tick = last_tick + 1 WHERE writer_fingerprint = ?1",
[writer_fingerprint],
)?;
if bumped == 0 {
conn.execute(
"INSERT OR IGNORE INTO writer_clock (writer_fingerprint, last_tick)
SELECT ?1, COALESCE(MAX(t), 0) FROM (
SELECT MAX(seq) AS t FROM turns
UNION ALL
SELECT MAX(activity_tick) AS t FROM conversations
UNION ALL
-- Other writers' issued ticks: keeps the seed at the true
-- issued-max even when their rows were pruned (review
-- finding N6 on #261 — Lamport receive rule over all
-- observable evidence, not just surviving rows).
SELECT MAX(last_tick) AS t FROM writer_clock
)",
[writer_fingerprint],
)?;
conn.execute(
"UPDATE writer_clock SET last_tick = last_tick + 1 WHERE writer_fingerprint = ?1",
[writer_fingerprint],
)?;
}
Ok(conn.query_row(
"SELECT last_tick FROM writer_clock WHERE writer_fingerprint = ?1",
[writer_fingerprint],
|row| row.get(0),
)?)
}
fn apply_journal_mode(conn: &Connection) -> anyhow::Result<Option<String>> {
let wal: Result<String, rusqlite::Error> =
conn.pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get(0));
match wal {
Ok(mode) if mode.eq_ignore_ascii_case("wal") => {
conn.pragma_update(None, "synchronous", "NORMAL")?;
Ok(None)
}
Ok(mode) => {
tracing::warn!(%mode, "journal_mode=WAL did not take; keeping synchronous=FULL");
Ok(Some(format!("journal_mode pragma returned `{mode}`")))
}
Err(e) if wal_fallback_eligible(&e.to_string()) => {
let captured = e.to_string();
conn.pragma_update(None, "journal_mode", "DELETE")?;
tracing::warn!(
error = %captured,
"SQLite refused WAL (network filesystem?); conversations.db is running \
on the slower journal_mode=DELETE fallback"
);
Ok(Some(captured))
}
Err(e) => Err(e.into()),
}
}
fn wal_fallback_eligible(error_text: &str) -> bool {
let lower = error_text.to_lowercase();
lower.contains("locking protocol") || lower.contains("disk i/o error")
}
fn create_schema(conn: &Connection) -> anyhow::Result<()> {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS conversations (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
workspace_path TEXT NOT NULL, -- display only
workspace_key TEXT NOT NULL, -- scoping key: workspace_key_v2 (17.2 — blake3 remote+branch, path fallback)
persona TEXT,
end_reason TEXT, -- set by 17.7
writer_fingerprint TEXT NOT NULL, -- §6 ordering key, half 1
activity_tick INTEGER NOT NULL, -- §6 ordering key, half 2 (per-writer Lamport tick)
tip_hash TEXT NOT NULL, -- §6 chain tip (BLAKE3)
started_at_claim INTEGER NOT NULL, -- DISPLAY ONLY (wall-clock claim, unix nanos)
updated_at_claim INTEGER NOT NULL, -- DISPLAY ONLY
scratchpad TEXT NOT NULL DEFAULT '{}', -- JSON scratchpad <state> snapshot (#713); working memory, NOT hashed (§6 chain unchanged)
plan TEXT NOT NULL DEFAULT '{}' -- JSON plan-ledger snapshot (#715); working memory, NOT hashed (§6 chain unchanged)
);
CREATE TABLE IF NOT EXISTS turns (
conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
writer_fingerprint TEXT NOT NULL, -- §6: whose clock ticked
seq INTEGER NOT NULL, -- §6: strictly monotonic per writer — THE ordering key
prev_hash TEXT NOT NULL, -- §6: BLAKE3 of prior turn's canonical encoding
user TEXT NOT NULL,
assistant TEXT NOT NULL,
events TEXT NOT NULL DEFAULT '[]', -- JSON tool events; filled by 17.6
tokens_in INTEGER, -- filled by 17.6, consumed by 18.x
tokens_out INTEGER,
ts_claim INTEGER NOT NULL, -- DISPLAY ONLY (wall-clock claim, unix nanos)
encoding_version INTEGER NOT NULL DEFAULT 1, -- canonical-encoding dispatch (N1 on #261)
phantom_reaches TEXT NOT NULL DEFAULT '[]', -- JSON phantom-reach telemetry (#717); NOT hashed (§6 chain unchanged)
PRIMARY KEY (conversation_id, writer_fingerprint, seq)
);
-- The per-writer Lamport clock (§6 'each agent is its own clock').
CREATE TABLE IF NOT EXISTS writer_clock (
writer_fingerprint TEXT PRIMARY KEY,
last_tick INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_conversations_ws_tick
ON conversations (workspace_key, activity_tick);",
)?;
Ok(())
}
const EXPECTED_COLUMNS: &[(&str, &[(&str, &str)])] = &[
(
"conversations",
&[
("id", "TEXT"),
("title", "TEXT NOT NULL DEFAULT ''"),
("workspace_path", "TEXT NOT NULL DEFAULT ''"),
("workspace_key", "TEXT NOT NULL DEFAULT ''"),
("persona", "TEXT"),
("end_reason", "TEXT"),
("writer_fingerprint", "TEXT NOT NULL DEFAULT ''"),
("activity_tick", "INTEGER NOT NULL DEFAULT 0"),
("tip_hash", "TEXT NOT NULL DEFAULT ''"),
("started_at_claim", "INTEGER NOT NULL DEFAULT 0"),
("updated_at_claim", "INTEGER NOT NULL DEFAULT 0"),
("scratchpad", "TEXT NOT NULL DEFAULT '{}'"),
("plan", "TEXT NOT NULL DEFAULT '{}'"),
],
),
(
"turns",
&[
("conversation_id", "TEXT"),
("writer_fingerprint", "TEXT NOT NULL DEFAULT ''"),
("seq", "INTEGER NOT NULL DEFAULT 0"),
("prev_hash", "TEXT NOT NULL DEFAULT ''"),
("user", "TEXT NOT NULL DEFAULT ''"),
("assistant", "TEXT NOT NULL DEFAULT ''"),
("events", "TEXT NOT NULL DEFAULT '[]'"),
("tokens_in", "INTEGER"),
("tokens_out", "INTEGER"),
("ts_claim", "INTEGER NOT NULL DEFAULT 0"),
("encoding_version", "INTEGER NOT NULL DEFAULT 1"),
("phantom_reaches", "TEXT NOT NULL DEFAULT '[]'"),
],
),
(
"writer_clock",
&[
("writer_fingerprint", "TEXT"),
("last_tick", "INTEGER NOT NULL DEFAULT 0"),
],
),
];
fn reconcile_schema(conn: &Connection) -> anyhow::Result<()> {
for (table, expected) in EXPECTED_COLUMNS {
let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
let present: Vec<String> = stmt
.query_map([], |row| row.get::<_, String>(1))?
.collect::<Result<Vec<_>, _>>()?;
for (name, decl) in *expected {
if !present.iter().any(|c| c == name) {
conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {name} {decl}"))?;
tracing::info!(
table = *table,
column = *name,
"schema migration: added missing column"
);
}
}
}
Ok(())
}
fn events_extract_sql(source: &str, key: &str) -> String {
format!(
"CASE WHEN json_valid({source}) THEN \
(SELECT coalesce(group_concat(json_extract(value, '$.{key}'), ' '), '') \
FROM json_each({source})) \
ELSE '' END"
)
}
fn create_fts_index(conn: &Connection) -> anyhow::Result<()> {
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Immediate)?;
let have_index = tx
.query_row(
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'turns_fts'",
[],
|_| Ok(()),
)
.optional()?
.is_some();
let view_tools = events_extract_sql("events", "tool");
let view_digests = events_extract_sql("events", "args_digest");
let new_tools = events_extract_sql("new.events", "tool");
let new_digests = events_extract_sql("new.events", "args_digest");
let old_tools = events_extract_sql("old.events", "tool");
let old_digests = events_extract_sql("old.events", "args_digest");
tx.execute_batch(&format!(
"CREATE VIEW IF NOT EXISTS turns_fts_content AS
SELECT rowid,
user,
assistant,
{view_tools} AS tool_names,
{view_digests} AS tool_args_digest
FROM turns;
CREATE VIRTUAL TABLE IF NOT EXISTS turns_fts USING fts5(
user, assistant, tool_names, tool_args_digest,
content='turns_fts_content',
content_rowid='rowid',
tokenize='unicode61'
);
CREATE TRIGGER IF NOT EXISTS turns_fts_after_insert
AFTER INSERT ON turns BEGIN
INSERT INTO turns_fts(rowid, user, assistant, tool_names, tool_args_digest)
VALUES (new.rowid, new.user, new.assistant, {new_tools}, {new_digests});
END;
-- Fires per cascaded row on conversation delete. The 'delete'
-- command must receive the values that were indexed at insert
-- time — guaranteed by the append-only invariant on turns.
CREATE TRIGGER IF NOT EXISTS turns_fts_after_delete
AFTER DELETE ON turns BEGIN
INSERT INTO turns_fts(turns_fts, rowid, user, assistant, tool_names, tool_args_digest)
VALUES ('delete', old.rowid, old.user, old.assistant, {old_tools}, {old_digests});
END;"
))?;
if !have_index {
tx.execute(
&format!(
"INSERT INTO turns_fts(rowid, user, assistant, tool_names, tool_args_digest)
SELECT rowid, user, assistant, {view_tools}, {view_digests} FROM turns"
),
[],
)?;
tracing::info!("created the FTS5 recall index and backfilled existing turns (17.3)");
}
tx.commit()?;
Ok(())
}
enum QueryPart {
Term(String),
Op(&'static str),
}
pub fn sanitize_fts5_query(raw: &str) -> anyhow::Result<String> {
let mut parts: Vec<QueryPart> = Vec::new();
let mut rest = raw;
loop {
let Some(open) = rest.find('"') else {
push_plain_tokens(rest, &mut parts);
break;
};
push_plain_tokens(&rest[..open], &mut parts);
let after_open = &rest[open + 1..];
match after_open.find('"') {
Some(close) => {
let phrase = after_open[..close].trim();
if phrase.chars().any(char::is_alphanumeric) {
parts.push(QueryPart::Term(format!("\"{phrase}\"")));
}
rest = &after_open[close + 1..];
}
None => {
push_plain_tokens(after_open, &mut parts);
break;
}
}
}
let mut out: Vec<String> = Vec::new();
let mut pending_op: Option<&'static str> = None;
for part in parts {
match part {
QueryPart::Term(term) => {
if let Some(op) = pending_op.take() {
out.push(op.to_string());
}
out.push(term);
}
QueryPart::Op(op) => {
if !out.is_empty() && pending_op.is_none() {
pending_op = Some(op);
}
}
}
}
if out.is_empty() {
anyhow::bail!("search query reduced to nothing after FTS5 sanitizing: {raw:?}");
}
Ok(out.join(" "))
}
fn push_plain_tokens(text: &str, parts: &mut Vec<QueryPart>) {
for token in text.split_whitespace() {
match token {
"AND" => parts.push(QueryPart::Op("AND")),
"OR" => parts.push(QueryPart::Op("OR")),
"NOT" => parts.push(QueryPart::Op("NOT")),
"NEAR" => parts.push(QueryPart::Term("\"NEAR\"".to_string())),
_ => {
if let Some(term) = sanitize_bare_token(token) {
parts.push(QueryPart::Term(term));
}
}
}
}
}
fn sanitize_bare_token(token: &str) -> Option<String> {
let stripped: String = token
.chars()
.filter(|c| !matches!(c, '(' | ')' | '*' | '^' | '"'))
.collect();
if !stripped.chars().any(char::is_alphanumeric) {
return None;
}
let is_bareword = stripped
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || !c.is_ascii());
Some(if is_bareword {
stripped
} else {
format!("\"{stripped}\"")
})
}
fn import_legacy_json(
conn: &Connection,
root: &Path,
writer_fingerprint: &str,
) -> anyhow::Result<()> {
let legacy_root = root.join(LEGACY_JSON_DIR);
if !legacy_root.is_dir() {
return Ok(());
}
let mut records = collect_legacy_records(&legacy_root)?;
records.sort_by(|a, b| {
a.updated_at_unix_nanos
.cmp(&b.updated_at_unix_nanos)
.then_with(|| a.created_at_unix_nanos.cmp(&b.created_at_unix_nanos))
.then_with(|| a.id.cmp(&b.id))
});
let mut imported = 0usize;
for record in &records {
if import_one_record(conn, record, writer_fingerprint)? {
imported += 1;
}
}
let backup = retire_legacy_dir(root, &legacy_root)?;
tracing::info!(
imported,
found = records.len(),
backup = %backup.display(),
"one-time import of legacy JSON conversations complete; \
the original tree is kept as a backup"
);
Ok(())
}
fn migrate_workspace_key(conn: &Connection, workspace: &Path, v2_key: &str) -> anyhow::Result<()> {
#[allow(deprecated)]
let old_key = ConversationStore::workspace_id_for_path(workspace)?;
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Immediate)?;
let migrated = tx.execute(
"UPDATE conversations SET workspace_key = ?2 WHERE workspace_key = ?1",
rusqlite::params![old_key, v2_key],
)?;
tx.commit()?;
if migrated > 0 {
tracing::info!(
migrated,
workspace = %workspace.display(),
"re-keyed conversations from the retired UUIDv5 workspace key to v2"
);
}
Ok(())
}
fn collect_legacy_records(legacy_root: &Path) -> anyhow::Result<Vec<ConversationRecord>> {
let mut records = Vec::new();
for ws_entry in std::fs::read_dir(legacy_root)? {
let ws_dir = ws_entry?.path();
if !ws_dir.is_dir() {
continue;
}
let dir_key = ws_dir
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
for entry in std::fs::read_dir(&ws_dir)? {
let path = entry?.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let parsed = std::fs::read_to_string(&path)
.map_err(anyhow::Error::from)
.and_then(|text| Ok(serde_json::from_str::<ConversationRecord>(&text)?));
let record = match parsed {
Ok(record) => record,
Err(e) => {
tracing::warn!(
path = %path.display(),
error = %e,
"skipping unreadable legacy conversation record \
(the file is kept in the import backup)"
);
continue;
}
};
if record.workspace_id != dir_key {
tracing::warn!(
path = %path.display(),
body_workspace = %record.workspace_id,
dir_workspace = %dir_key,
"skipping legacy record whose workspace id does not match its dir"
);
continue;
}
if let Err(e) = validate_record_id(&record.id) {
tracing::warn!(path = %path.display(), error = %e, "skipping legacy record");
continue;
}
records.push(record);
}
}
Ok(records)
}
fn import_one_record(
conn: &Connection,
record: &ConversationRecord,
writer_fingerprint: &str,
) -> anyhow::Result<bool> {
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Immediate)?;
let already: Option<i64> = tx
.query_row(
"SELECT 1 FROM conversations WHERE id = ?1",
[&record.id],
|row| row.get(0),
)
.optional()?;
if already.is_some() {
tracing::debug!(id = %record.id, "legacy conversation already in the db; skipping");
return Ok(false);
}
let started_claim = clamp_claim(record.created_at_unix_nanos);
let updated_claim = clamp_claim(record.updated_at_unix_nanos);
let create_tick = next_tick(&tx, writer_fingerprint)?;
tx.execute(
"INSERT INTO conversations
(id, title, workspace_path, workspace_key, persona, end_reason,
writer_fingerprint, activity_tick, tip_hash,
started_at_claim, updated_at_claim)
VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6, ?7, ?8, ?9, ?10)",
rusqlite::params![
record.id,
record.title,
record.workspace,
record.workspace_id,
record.persona,
writer_fingerprint,
create_tick,
genesis_hash(&record.id, writer_fingerprint),
started_claim,
updated_claim,
],
)?;
let mut prev_hash = genesis_hash(&record.id, writer_fingerprint);
let mut last_tick = create_tick;
for turn in &record.turns {
let seq = next_tick(&tx, writer_fingerprint)?;
let row = TurnRow {
conversation_id: record.id.clone(),
writer_fingerprint: writer_fingerprint.to_string(),
seq,
prev_hash,
user: turn.user.clone(),
assistant: turn.assistant.clone(),
events: "[]".to_string(),
tokens_in: None,
tokens_out: None,
ts_claim: updated_claim,
encoding_version: TURN_ENCODING_VERSION_CURRENT,
};
insert_turn_row(&tx, &row, "[]")?;
prev_hash = row.content_hash()?;
last_tick = seq;
}
if !record.turns.is_empty() {
tx.execute(
"UPDATE conversations SET activity_tick = ?2, tip_hash = ?3 WHERE id = ?1",
rusqlite::params![record.id, last_tick, prev_hash],
)?;
}
tx.commit()?;
Ok(true)
}
fn retire_legacy_dir(root: &Path, legacy_root: &Path) -> anyhow::Result<PathBuf> {
for attempt in 0u32..100 {
let candidate = if attempt == 0 {
root.join(LEGACY_BACKUP_DIR)
} else {
root.join(format!("{LEGACY_BACKUP_DIR}.{attempt}"))
};
if candidate.exists() {
continue;
}
return match std::fs::rename(legacy_root, &candidate) {
Ok(()) => Ok(candidate),
Err(_) if !legacy_root.exists() => Ok(candidate),
Err(e) => Err(anyhow::Error::from(e).context(format!(
"imported legacy conversations but could not move {} aside to {}",
legacy_root.display(),
candidate.display()
))),
};
}
anyhow::bail!(
"no free backup name for {} (conversations.imported* all taken)",
legacy_root.display()
)
}
fn clamp_claim(nanos: u128) -> i64 {
i64::try_from(nanos).unwrap_or(i64::MAX)
}
fn resolve_writer_fingerprint(root: &Path) -> anyhow::Result<String> {
let pem = root.join(IDENTITY_PEM_FILE);
if pem.is_file() {
match agent_mesh_protocol::UserKey::load(&pem) {
Ok(user) => return Ok(user.fingerprint().hex()),
Err(e) => {
tracing::warn!(
path = %pem.display(),
error = %e,
"identity.pem exists but did not parse; \
falling back to the per-install nonce writer fingerprint"
);
}
}
}
load_or_create_writer_fingerprint(root)
}
fn load_or_create_writer_fingerprint(root: &Path) -> anyhow::Result<String> {
let path = root.join(NONCE_FILE);
let nonce = match std::fs::read_to_string(&path) {
Ok(text) if !text.trim().is_empty() => text.trim().to_string(),
_ => {
let minted = uuid::Uuid::new_v4().to_string();
let tmp = root.join(format!(
"{NONCE_FILE}.{}.{:?}.tmp",
std::process::id(),
std::thread::current().id()
));
std::fs::write(&tmp, &minted)?;
let publish = std::fs::hard_link(&tmp, &path);
let _ = std::fs::remove_file(&tmp);
match publish {
Ok(()) => minted,
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
let adopted = std::fs::read_to_string(&path)?.trim().to_string();
if adopted.is_empty() {
anyhow::bail!(
"install nonce at {} exists but is empty — remove it and retry",
path.display()
);
}
adopted
}
Err(e) => return Err(e.into()),
}
}
};
Ok(blake3::hash(nonce.as_bytes()).to_hex().to_string())
}
fn now_claim_nanos() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| i64::try_from(d.as_nanos()).unwrap_or(i64::MAX))
.unwrap_or(0)
}
fn claim_to_u128(claim: i64) -> u128 {
claim.max(0) as u128
}
fn tokens_from_sql(value: Option<i64>) -> anyhow::Result<Option<u32>> {
value
.map(|v| {
u32::try_from(v)
.map_err(|_| anyhow::anyhow!("token count {v} out of range (tampered row?)"))
})
.transpose()
}
fn validate_record_id(id: &str) -> anyhow::Result<()> {
if id.is_empty() || !id.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') {
anyhow::bail!("invalid conversation id `{id}`");
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn wal_fallback_classifier_matches_known_nfs_failures() {
assert!(wal_fallback_eligible("locking protocol"));
assert!(wal_fallback_eligible("disk I/O error"));
assert!(wal_fallback_eligible(
"sqlite failure: `Error code 15: Locking Protocol`"
));
assert!(!wal_fallback_eligible("no such table: turns"));
assert!(!wal_fallback_eligible("database is locked"));
assert!(!wal_fallback_eligible(""));
}
#[test]
fn canonical_encoding_is_unambiguous_across_field_boundaries() {
let base = TurnRow {
conversation_id: "c".into(),
writer_fingerprint: "w".into(),
seq: 1,
prev_hash: "p".into(),
user: "ab".into(),
assistant: "c".into(),
events: "[]".into(),
tokens_in: None,
tokens_out: None,
ts_claim: 7,
encoding_version: 1,
};
let shifted = TurnRow {
user: "a".into(),
assistant: "bc".into(),
..clone_row(&base)
};
assert_ne!(
base.canonical_encoding_v1(),
shifted.canonical_encoding_v1(),
"length prefixes must prevent (ab,c) == (a,bc)"
);
let skewed = TurnRow {
ts_claim: 8,
..clone_row(&base)
};
assert_ne!(base.content_hash().unwrap(), skewed.content_hash().unwrap());
let tokens = TurnRow {
tokens_in: Some(5),
..clone_row(&base)
};
assert_ne!(base.content_hash().unwrap(), tokens.content_hash().unwrap());
}
#[test]
fn content_hash_rejects_unknown_encoding_versions() {
let v1 = TurnRow {
conversation_id: "c".into(),
writer_fingerprint: "w".into(),
seq: 1,
prev_hash: "p".into(),
user: "u".into(),
assistant: "a".into(),
events: "[]".into(),
tokens_in: None,
tokens_out: None,
ts_claim: 7,
encoding_version: 1,
};
v1.content_hash().expect("v1 must hash");
let future = TurnRow {
encoding_version: 2,
..clone_row(&v1)
};
let err = future.content_hash().unwrap_err().to_string();
assert!(
err.contains("encoding_version 2") && err.contains("known: 1"),
"unknown version must error clearly: {err}"
);
}
#[test]
fn clamp_claim_saturates_oversized_legacy_nanos() {
assert_eq!(clamp_claim(0), 0);
assert_eq!(clamp_claim(42), 42);
assert_eq!(clamp_claim(u128::MAX), i64::MAX);
}
#[test]
fn genesis_hash_is_deterministic_and_writer_scoped() {
assert_eq!(genesis_hash("conv", "w1"), genesis_hash("conv", "w1"));
assert_ne!(genesis_hash("conv", "w1"), genesis_hash("conv", "w2"));
assert_ne!(genesis_hash("conv", "w1"), genesis_hash("other", "w1"));
assert_ne!(genesis_hash("ab", "c"), genesis_hash("a", "bc"));
}
#[test]
fn writer_fingerprint_is_stable_per_install_and_distinct_across_installs() {
let root_a = tempfile::tempdir().unwrap();
let root_b = tempfile::tempdir().unwrap();
let first = load_or_create_writer_fingerprint(root_a.path()).unwrap();
let again = load_or_create_writer_fingerprint(root_a.path()).unwrap();
let other = load_or_create_writer_fingerprint(root_b.path()).unwrap();
assert_eq!(first, again, "fingerprint must be stable per install");
assert_ne!(first, other, "two installs must not share a fingerprint");
assert_eq!(first.len(), 64, "blake3 hex");
}
#[test]
fn wal_mode_pairs_with_synchronous_normal_on_the_stores_connection() {
let root = tempfile::tempdir().unwrap();
let workspace = tempfile::tempdir().unwrap();
let store = ConversationStore::new(root.path(), workspace.path(), 100).unwrap();
let conn = store.lock_conn();
let sync_level: i64 = conn
.query_row("PRAGMA synchronous", [], |row| row.get(0))
.unwrap();
assert_eq!(sync_level, 1, "WAL must run at synchronous=NORMAL (1)");
}
#[test]
fn claim_clock_saturates_instead_of_wrapping() {
let now = now_claim_nanos();
assert!(now > 0);
assert_eq!(claim_to_u128(-5), 0);
assert_eq!(claim_to_u128(42), 42);
}
#[test]
fn load_turn_reads_one_turn_by_seq_and_misses_are_none() {
let root = tempfile::tempdir().unwrap();
let workspace = tempfile::tempdir().unwrap();
let store = ConversationStore::new(root.path(), workspace.path(), 100).unwrap();
let conv = store.create("t", None).unwrap();
store
.append_turn(&conv, "the question", "the answer")
.unwrap();
let hits = store.search("question", 5).unwrap();
assert_eq!(hits.len(), 1);
let seq = hits[0].seq;
let turn = store.load_turn(&conv, seq).unwrap().expect("turn exists");
assert_eq!(turn.user, "the question");
assert_eq!(turn.assistant, "the answer");
assert!(store.load_turn(&conv, seq + 9_999).unwrap().is_none());
assert!(store.load_turn("no-such-conv", seq).unwrap().is_none());
}
#[test]
fn end_conversation_hides_row_from_latest_open_but_not_from_list() {
let root = tempfile::tempdir().unwrap();
let workspace = tempfile::tempdir().unwrap();
let store = ConversationStore::new(root.path(), workspace.path(), 100).unwrap();
let c1 = store.create("first", None).unwrap();
store.append_turn(&c1, "q1", "a1").unwrap();
let c2 = store.create("second", None).unwrap();
store.append_turn(&c2, "q2", "a2").unwrap();
assert_eq!(store.latest_open().unwrap().unwrap().id, c2);
store.end_conversation(&c2, "wq").unwrap();
assert_eq!(
store.latest_open().unwrap().unwrap().id,
c1,
"an ended conversation is skipped on auto-resume"
);
assert_eq!(store.list().unwrap().len(), 2);
assert!(
store
.search("q2", 5)
.unwrap()
.iter()
.any(|h| h.conversation_id == c2),
"ended conversation stays in the FTS index for /recall"
);
store.end_conversation(&c1, "end").unwrap();
assert!(store.latest_open().unwrap().is_none());
assert_eq!(store.list().unwrap().len(), 2, "still listed after ending");
}
#[test]
fn end_conversation_does_not_tick_activity_and_is_idempotent() {
let root = tempfile::tempdir().unwrap();
let workspace = tempfile::tempdir().unwrap();
let store = ConversationStore::new(root.path(), workspace.path(), 100).unwrap();
let older = store.create("older", None).unwrap();
store.append_turn(&older, "q", "a").unwrap();
let newer = store.create("newer", None).unwrap();
store.append_turn(&newer, "q", "a").unwrap();
let tick_of = |id: &str| -> i64 {
let conn = store.lock_conn();
conn.query_row(
"SELECT activity_tick FROM conversations WHERE id = ?1",
[id],
|row| row.get(0),
)
.unwrap()
};
let before = tick_of(&older);
store.end_conversation(&older, "new").unwrap();
assert_eq!(tick_of(&older), before, "ending must not bump the tick");
store.end_conversation(&older, "new").unwrap();
assert_eq!(store.latest_open().unwrap().unwrap().id, newer);
}
fn s(raw: &str) -> String {
sanitize_fts5_query(raw).unwrap()
}
#[test]
fn sanitizer_auto_quotes_dotted_hyphenated_and_path_tokens() {
assert_eq!(s("chat-send"), "\"chat-send\"");
assert_eq!(s("P2.2"), "\"P2.2\"");
assert_eq!(s("my-app.config.ts"), "\"my-app.config.ts\"");
assert_eq!(s("src/store.rs"), "\"src/store.rs\"");
assert_eq!(s("tcp:p4d.p4d-ascii:1666"), "\"tcp:p4d.p4d-ascii:1666\"");
assert_eq!(s("issue #246"), "issue \"#246\"");
assert_eq!(s("hello world"), "hello world");
assert_eq!(s("writer_clock"), "writer_clock");
assert_eq!(s("schlüssel wörter"), "schlüssel wörter");
}
#[test]
fn sanitizer_preserves_balanced_phrases_and_drops_dangling_quotes() {
assert_eq!(s("\"exact phrase\" extra"), "\"exact phrase\" extra");
assert_eq!(s("say \"hello world\" now"), "say \"hello world\" now");
assert_eq!(s("foo \"bar"), "foo bar");
assert_eq!(s("\"unclosed"), "unclosed");
assert_eq!(s("\"a b\" \"c"), "\"a b\" c");
assert_eq!(s("\"AND OR\""), "\"AND OR\"");
assert_eq!(s("\"P2.2 chat-send\""), "\"P2.2 chat-send\"");
let err = sanitize_fts5_query("\"\"").unwrap_err().to_string();
assert!(err.contains("reduced to nothing"), "{err}");
let err = sanitize_fts5_query("\"--\"").unwrap_err().to_string();
assert!(err.contains("reduced to nothing"), "{err}");
}
#[test]
fn sanitizer_trims_dangling_operators() {
assert_eq!(s("foo AND"), "foo");
assert_eq!(s("OR foo"), "foo");
assert_eq!(s("NOT foo"), "foo");
assert_eq!(s("foo AND AND bar"), "foo AND bar");
assert_eq!(s("foo AND OR bar"), "foo AND bar");
assert_eq!(s("AND foo OR"), "foo");
assert_eq!(s("foo OR bar"), "foo OR bar");
assert_eq!(s("foo NOT bar"), "foo NOT bar");
assert_eq!(s("a OR b OR c"), "a OR b OR c");
assert_eq!(s("foo and bar"), "foo and bar");
let err = sanitize_fts5_query("AND").unwrap_err().to_string();
assert!(err.contains("reduced to nothing"), "{err}");
assert_eq!(s("NEAR"), "\"NEAR\"");
assert_eq!(s("near"), "near");
}
#[test]
fn sanitizer_strips_metacharacter_injection() {
assert_eq!(s("(foo OR bar) AND baz"), "foo OR bar AND baz");
assert_eq!(s("foo* ^bar"), "foo bar");
assert_eq!(s("col*umn"), "column");
for q in ["\"", "*", "^", "( )", "*^()"] {
let err = sanitize_fts5_query(q).unwrap_err().to_string();
assert!(err.contains("reduced to nothing"), "{q:?}: {err}");
}
assert_eq!(s("fo\"o bar"), "fo o bar");
assert_eq!(s("?? foo !!"), "foo");
assert_eq!(s("foo \u{a0} "), "foo"); }
#[test]
fn sanitizer_handles_mixed_phrases_terms_and_operators() {
assert_eq!(
s("\"tuning writeback\" OR coverage-floor"),
"\"tuning writeback\" OR \"coverage-floor\""
);
assert_eq!(
s("error \"chain violation\" NOT P2.2"),
"error \"chain violation\" NOT \"P2.2\""
);
assert_eq!(s("AND \"lead phrase\" tail"), "\"lead phrase\" tail");
}
#[test]
fn sanitizer_errors_on_empty_and_whitespace_queries() {
for q in ["", " ", "\t\n"] {
let err = sanitize_fts5_query(q).unwrap_err().to_string();
assert!(err.contains("reduced to nothing"), "{q:?}: {err}");
}
}
#[test]
fn events_extract_sql_guards_and_targets_the_seam_keys() {
let sql = events_extract_sql("new.events", "tool");
assert!(sql.contains("json_valid(new.events)"));
assert!(sql.contains("json_each(new.events)"));
assert!(sql.contains("'$.tool'"));
assert!(sql.contains("ELSE '' END"));
}
fn clone_row(row: &TurnRow) -> TurnRow {
TurnRow {
conversation_id: row.conversation_id.clone(),
writer_fingerprint: row.writer_fingerprint.clone(),
seq: row.seq,
prev_hash: row.prev_hash.clone(),
user: row.user.clone(),
assistant: row.assistant.clone(),
events: row.events.clone(),
tokens_in: row.tokens_in,
tokens_out: row.tokens_out,
ts_claim: row.ts_claim,
encoding_version: row.encoding_version,
}
}
}