use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use rusqlite::{Connection, OptionalExtension, TransactionBehavior};
mod artifacts;
mod fts;
use fts::create_fts_index;
pub use fts::sanitize_fts5_query;
mod legacy_import;
use legacy_import::{
import_legacy_json, migrate_pin_cognition_labels, migrate_pin_tenacity_split,
migrate_workspace_key,
};
mod liveness;
pub(crate) use liveness::pid_is_alive;
#[cfg(all(test, target_os = "linux"))]
use liveness::pid_start_unix_nanos;
use liveness::{current_host_boot, live_owner_row, system_liveness};
#[cfg(test)]
use liveness::{
open_process_failure_reports_live_or_unknown, pid_identity_matches,
wait_probe_reports_live_or_unknown,
};
pub use liveness::{ClaimOutcome, LivenessFn, StoredOwner};
mod prompts;
#[cfg(test)]
use prompts::{insert_prompt_receipt, MAX_PROMPT_LINEAGE_DEPTH};
use prompts::{
load_prompt_in_conversation_on_conn, prompt_chain_on_conn, validate_objective_root_on_conn,
};
mod schema;
#[cfg(test)]
use schema::wal_fallback_eligible;
use schema::{apply_journal_mode, create_schema, reconcile_schema, BUSY_TIMEOUT};
mod turn_chain;
use turn_chain::{
genesis_hash, insert_turn_row, last_turn, next_tick, turn_at_seq, TurnRow,
TURN_ENCODING_VERSION_CURRENT,
};
use crate::conversation::{
new_conversation_id, session_plan_dir, ConversationRecord, ConversationSummary,
ConversationTurn,
};
#[cfg(test)]
use crate::prompt::{PromptId, PromptOrigin, PromptReceipt};
const DB_FILE: &str = "conversations.db";
const NONCE_FILE: &str = "install-nonce";
const IDENTITY_PEM_FILE: &str = "identity.pem";
#[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,
host: String,
boot_id: String,
pid: i64,
liveness: LivenessFn,
}
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| migrate_roadmaps_pk(&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)?;
migrate_pin_cognition_labels(&conn)?;
migrate_pin_tenacity_split(&conn)?;
let (host, boot_id) = current_host_boot();
Ok(Self {
conn: Arc::new(Mutex::new(conn)),
workspace,
workspace_id,
writer_fingerprint,
max_per_workspace,
wal_fallback,
claim_clock: now_claim_nanos,
host,
boot_id,
pid: i64::from(std::process::id()),
liveness: system_liveness,
})
}
#[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 has_prompt_receipts: bool = tx.query_row(
"SELECT EXISTS(
SELECT 1 FROM prompt_receipts WHERE conversation_id = ?1
)",
[id],
|row| row.get(0),
)?;
if has_prompt_receipts {
anyhow::bail!(
"conversation `{id}` has immutable prompt receipts; refusing to \
recreate it implicitly (delete it explicitly before reusing the id)"
);
}
}
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()?;
}
if let Err(error) = self.prune_to_cap_excluding(id) {
tracing::warn!(
%error,
conversation_id = id,
"conversation created but retention pruning failed"
);
}
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],
sources: &[String],
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)?;
if !(sources.is_empty() || events.is_empty() && phantom_reaches.is_empty()) {
anyhow::bail!(
"refusing the append -- this turn claims derivation (non-empty \
sources) AND tool activity; a derived row is harness-minted and \
carries neither events nor phantom reaches, and a row with both \
could never be verified again"
);
}
let sources_json = canonical_sources_json(sources)?;
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let tick = next_tick(&tx, &self.writer_fingerprint)?;
let (recorded_tip, tip_writer): (String, String) = tx
.query_row(
"SELECT tip_hash, writer_fingerprint FROM conversations WHERE id = ?1",
[&id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?
.ok_or_else(|| anyhow::anyhow!("conversation `{id}` not found"))?;
if !recorded_tip.is_empty() {
let tip_writer_final = last_turn(&tx, &id, &tip_writer)?;
Self::check_tip_witness(&id, &recorded_tip, &tip_writer, tip_writer_final.as_ref())
.map_err(|e| {
anyhow::anyhow!(
"refusing the append -- the recorded chain tip could not be \
confirmed, so new work must not extend that tip: {e:#}"
)
})?;
if tip_writer != self.writer_fingerprint {
if let Some(final_row) = tip_writer_final.as_ref() {
let existing: Option<(String, i64)> = tx
.query_row(
"SELECT tip_hash, tip_seq FROM writer_tips
WHERE conversation_id = ?1 AND writer_fingerprint = ?2",
rusqlite::params![id, tip_writer],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
match existing {
None => {
tx.execute(
"INSERT INTO writer_tips
(conversation_id, writer_fingerprint, tip_hash, tip_seq)
VALUES (?1, ?2, ?3, ?4)",
rusqlite::params![id, tip_writer, recorded_tip, final_row.seq],
)?;
}
Some((existing_hash, existing_seq)) => {
let fetched;
let row_at: Option<&TurnRow> = if final_row.seq == existing_seq {
Some(final_row)
} else {
fetched = turn_at_seq(&tx, &id, &tip_writer, existing_seq)?;
fetched.as_ref()
};
Self::check_writer_tip_witness(
&id,
&tip_writer,
&existing_hash,
existing_seq,
final_row.seq,
row_at,
)
.map_err(|e| {
anyhow::anyhow!(
"refusing the append -- the outgoing writer's own \
recorded witness could not be confirmed, so the \
handoff must not overwrite it: {e:#}"
)
})?;
if existing_seq < final_row.seq {
tx.execute(
"UPDATE writer_tips SET tip_hash = ?3, tip_seq = ?4
WHERE conversation_id = ?1 AND writer_fingerprint = ?2",
rusqlite::params![id, tip_writer, recorded_tip, final_row.seq],
)?;
}
}
}
}
}
}
let own_last = last_turn(&tx, &id, &self.writer_fingerprint)?;
let own_witness: Option<(String, i64)> = tx
.query_row(
"SELECT tip_hash, tip_seq FROM writer_tips
WHERE conversation_id = ?1 AND writer_fingerprint = ?2",
rusqlite::params![id, self.writer_fingerprint],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((tip_hash, tip_seq)) = own_witness {
let final_seq = own_last.as_ref().map(|r| r.seq).unwrap_or(0);
let fetched;
let row_at_seq: Option<&TurnRow> =
if own_last.as_ref().is_some_and(|r| r.seq == tip_seq) {
own_last.as_ref()
} else {
fetched = turn_at_seq(&tx, &id, &self.writer_fingerprint, tip_seq)?;
fetched.as_ref()
};
Self::check_writer_tip_witness(
&id,
&self.writer_fingerprint,
&tip_hash,
tip_seq,
final_seq,
row_at_seq,
)
.map_err(|e| {
anyhow::anyhow!(
"refusing the append -- this writer's own recorded witness could \
not be confirmed, so new work must not chain past it: {e:#}"
)
})?;
}
let prev_hash = match own_last {
Some(ref 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,
phantom_reaches: phantom_reaches_json,
sources: sources_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)?;
let row_hash = row.content_hash()?;
tx.execute(
"INSERT INTO writer_tips (conversation_id, writer_fingerprint, tip_hash, tip_seq)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (conversation_id, writer_fingerprint)
DO UPDATE SET tip_hash = excluded.tip_hash, tip_seq = excluded.tip_seq",
rusqlite::params![id, self.writer_fingerprint, row_hash, tick],
)?;
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_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();
Self::load_record_on(&conn, &self.workspace_id, &id)
}
fn load_record_on(
conn: &Connection,
workspace_id: &str,
id: &str,
) -> anyhow::Result<ConversationRecord> {
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,
roadmap_id, node_id
FROM conversations
WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, 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(),
roadmap_id: row.get(9)?,
node_id: row.get(10)?,
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, sources
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)?,
row.get::<_, String>(6)?,
))
})?;
for turn in turns {
let (
user,
assistant,
events_json,
tokens_in,
tokens_out,
phantom_reaches_json,
sources_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"
)
})?;
let sources = parse_canonical_sources(&sources_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: turn sources column is not canonical \
provenance 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)?,
sources,
});
}
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, t.sources
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)?,
row.get::<_, String>(6)?,
))
},
)
.optional()?;
let Some((
user,
assistant,
events_json,
tokens_in,
tokens_out,
phantom_reaches_json,
sources_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"
)
})?;
let sources = parse_canonical_sources(&sources_json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: turn sources column is not canonical \
provenance 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)?,
sources,
}))
}
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 list_all(&self) -> anyhow::Result<Vec<(ConversationSummary, String)>> {
let conn = self.lock_conn();
let mut stmt = conn.prepare(
"SELECT c.id, c.title, c.persona, c.updated_at_claim, c.workspace_path,
(SELECT COUNT(*) FROM turns t WHERE t.conversation_id = c.id)
FROM conversations c
ORDER BY c.activity_tick DESC, c.id ASC",
)?;
let rows = stmt.query_map([], |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>(5)?.max(0) as usize,
},
row.get::<_, String>(4)?,
))
})?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
}
pub fn session_change_index(&self) -> anyhow::Result<Vec<(String, i64)>> {
let conn = self.lock_conn();
let mut stmt =
conn.prepare("SELECT c.id, c.activity_tick FROM conversations c ORDER BY c.id ASC")?;
let rows = stmt.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
})?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
}
pub fn inject_prompt(
&self,
conversation_id: &str,
body: &str,
idem_key: Option<&str>,
) -> anyhow::Result<InjectOutcome> {
validate_record_id(conversation_id)?;
let now = (self.claim_clock)();
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let owner: Option<String> = tx
.query_row(
"SELECT workspace_key FROM conversations WHERE id = ?1",
[conversation_id],
|row| row.get(0),
)
.optional()?;
match owner.as_deref() {
None => anyhow::bail!("cannot inject into unknown conversation `{conversation_id}`"),
Some(key) if key != self.workspace_id => {
anyhow::bail!("conversation `{conversation_id}` belongs to another workspace")
}
_ => {}
}
if let Some(key) = idem_key {
let seen = tx
.query_row(
"SELECT 1 FROM conversation_inbox
WHERE conversation_id = ?1 AND idem_key = ?2",
rusqlite::params![conversation_id, key],
|_| Ok(()),
)
.optional()?
.is_some();
if seen {
tx.commit()?;
return Ok(InjectOutcome::Duplicate);
}
}
let seq: i64 = tx.query_row(
"SELECT COALESCE(MAX(seq), 0) + 1 FROM conversation_inbox WHERE conversation_id = ?1",
[conversation_id],
|row| row.get(0),
)?;
tx.execute(
"INSERT INTO conversation_inbox
(id, conversation_id, workspace_key, seq, body, idem_key, delivered, injected_at_claim)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7)",
rusqlite::params![
new_conversation_id(),
conversation_id,
self.workspace_id,
seq,
body,
idem_key,
now,
],
)?;
tx.commit()?;
Ok(InjectOutcome::Enqueued)
}
pub fn take_injected_prompt(
&self,
conversation_id: &str,
) -> anyhow::Result<Option<InjectedPrompt>> {
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let row: Option<(String, String, i64)> = tx
.query_row(
"SELECT id, body, seq FROM conversation_inbox
WHERE conversation_id = ?1 AND workspace_key = ?2 AND delivered = 0
ORDER BY seq ASC LIMIT 1",
rusqlite::params![conversation_id, self.workspace_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.optional()?;
let Some((id, body, seq)) = row else {
tx.commit()?;
return Ok(None);
};
tx.execute(
"UPDATE conversation_inbox SET delivered = 1 WHERE id = ?1",
[&id],
)?;
tx.commit()?;
Ok(Some(InjectedPrompt { id, body, seq }))
}
pub fn link_inbox_delivery(&self, inbox_id: &str, receipt_id: &str) -> anyhow::Result<()> {
let conn = self.lock_conn();
conn.execute(
"UPDATE conversation_inbox SET delivered_receipt_id = ?2 WHERE id = ?1",
[inbox_id, receipt_id],
)?;
Ok(())
}
pub const PERMISSION_REQUEST_TTL_NANOS: i64 = 5 * 60 * 1_000_000_000;
pub(crate) const ENROLLMENT_REQUEST_TTL_NANOS: i64 = 5 * 60 * 1_000_000_000;
pub fn publish_enrollment_candidate(
&self,
conversation_id: &str,
candidate_json: &str,
) -> anyhow::Result<String> {
validate_record_id(conversation_id)?;
let now = (self.claim_clock)();
let request_id = new_conversation_id();
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let owner: Option<String> = tx
.query_row(
"SELECT workspace_key FROM conversations WHERE id = ?1",
[conversation_id],
|row| row.get(0),
)
.optional()?;
match owner.as_deref() {
None => anyhow::bail!(
"cannot stage an enrollment for unknown conversation `{conversation_id}`"
),
Some(key) if key != self.workspace_id => {
anyhow::bail!("conversation `{conversation_id}` belongs to another workspace")
}
_ => {}
}
tx.execute(
"INSERT INTO enrollment_requests
(request_id, conversation_id, workspace_key, candidate_json, resolved, created_tick)
VALUES (?1, ?2, ?3, ?4, 0, ?5)",
rusqlite::params![
request_id,
conversation_id,
self.workspace_id,
candidate_json,
now,
],
)?;
tx.commit()?;
Ok(request_id)
}
pub fn pending_enrollment_candidate(
&self,
conversation_id: &str,
) -> anyhow::Result<Option<PendingEnrollment>> {
let conn = self.lock_conn();
let cutoff = (self.claim_clock)().saturating_sub(Self::ENROLLMENT_REQUEST_TTL_NANOS);
conn.query_row(
"SELECT request_id, candidate_json FROM enrollment_requests
WHERE conversation_id = ?1 AND workspace_key = ?2 AND resolved = 0
AND created_tick > ?3
ORDER BY created_tick ASC LIMIT 1",
rusqlite::params![conversation_id, self.workspace_id, cutoff],
|row| {
Ok(PendingEnrollment {
request_id: row.get(0)?,
candidate_json: row.get(1)?,
})
},
)
.optional()
.map_err(Into::into)
}
pub fn take_enrollment_candidate(
&self,
conversation_id: &str,
request_id: &str,
) -> anyhow::Result<Option<String>> {
let cutoff = (self.claim_clock)().saturating_sub(Self::ENROLLMENT_REQUEST_TTL_NANOS);
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let candidate: Option<String> = tx
.query_row(
"SELECT candidate_json FROM enrollment_requests
WHERE request_id = ?1 AND conversation_id = ?2 AND workspace_key = ?3
AND resolved = 0 AND created_tick > ?4",
rusqlite::params![request_id, conversation_id, self.workspace_id, cutoff],
|row| row.get::<_, String>(0),
)
.optional()?;
if candidate.is_some() {
tx.execute(
"UPDATE enrollment_requests SET resolved = 1, resolved_by = 'terminal'
WHERE request_id = ?1",
[request_id],
)?;
}
tx.commit()?;
Ok(candidate)
}
pub fn decline_enrollment_candidate(
&self,
conversation_id: &str,
request_id: &str,
) -> anyhow::Result<bool> {
let conn = self.lock_conn();
let changed = conn.execute(
"UPDATE enrollment_requests SET resolved = 1, resolved_by = 'declined'
WHERE request_id = ?1 AND conversation_id = ?2 AND workspace_key = ?3
AND resolved = 0",
rusqlite::params![request_id, conversation_id, self.workspace_id],
)?;
Ok(changed == 1)
}
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 link_conversation_to_node(
&self,
id: &str,
roadmap_id: Option<&str>,
node_id: Option<&str>,
) -> anyhow::Result<()> {
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET roadmap_id = ?2, node_id = ?3
WHERE id = ?1 AND workspace_key = ?4",
rusqlite::params![id, roadmap_id, node_id, self.workspace_id],
)?;
Ok(())
}
pub fn create_roadmap(
&self,
id: &str,
title: &str,
tree: &crate::plan::Plan,
) -> anyhow::Result<()> {
validate_record_id(id)?;
let now = (self.claim_clock)();
let toml = tree.to_toml_string()?;
let conn = self.lock_conn();
conn.execute(
"INSERT OR REPLACE INTO roadmaps
(id, workspace_key, title, tree, schema_version, created_at_claim, updated_at_claim)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)",
rusqlite::params![
id,
self.workspace_id,
title.trim(),
toml,
ROADMAP_SCHEMA_VERSION,
now
],
)?;
Ok(())
}
pub fn load_roadmap(&self, id: &str) -> anyhow::Result<Option<Roadmap>> {
let conn = self.lock_conn();
let row = conn
.query_row(
"SELECT title, tree FROM roadmaps WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
)
.optional()?;
let Some((title, toml)) = row else {
return Ok(None);
};
let tree = crate::plan::Plan::from_toml_str(&toml).map_err(|e| {
anyhow::anyhow!("roadmap `{id}`: tree column is not valid plan TOML ({e})")
})?;
Ok(Some(Roadmap {
id: id.to_string(),
title,
tree,
}))
}
pub fn update_roadmap(&self, id: &str, tree: &crate::plan::Plan) -> anyhow::Result<()> {
let now = (self.claim_clock)();
let toml = tree.to_toml_string()?;
let conn = self.lock_conn();
conn.execute(
"UPDATE roadmaps SET tree = ?2, updated_at_claim = ?3
WHERE id = ?1 AND workspace_key = ?4",
rusqlite::params![id, toml, now, self.workspace_id],
)?;
Ok(())
}
pub fn list_roadmaps(&self) -> anyhow::Result<Vec<RoadmapSummary>> {
let conn = self.lock_conn();
let mut stmt = conn.prepare(
"SELECT id, title, tree FROM roadmaps
WHERE workspace_key = ?1
ORDER BY updated_at_claim DESC, id DESC",
)?;
let rows = stmt.query_map([&self.workspace_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})?;
let mut out = Vec::new();
for row in rows {
let (id, title, toml) = row?;
let node_count = crate::plan::Plan::from_toml_str(&toml)
.map(|p| p.subtasks.len())
.unwrap_or(0);
out.push(RoadmapSummary {
id,
title,
node_count,
});
}
Ok(out)
}
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 title(&self, id: &str) -> anyhow::Result<Option<String>> {
let conn = self.lock_conn();
let title = conn
.query_row(
"SELECT title FROM conversations WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
|row| row.get::<_, String>(0),
)
.optional()?;
Ok(title)
}
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 update_preference_pin(
&self,
id: &str,
pin: &crate::OperatorPreferencePin,
) -> anyhow::Result<()> {
let id = self.resolve_id(id)?;
let json = serde_json::to_string(pin)?;
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET preference_pin = ?2 WHERE id = ?1 AND workspace_key = ?3",
rusqlite::params![id, json, self.workspace_id],
)?;
Ok(())
}
#[doc(hidden)]
pub fn set_raw_preference_pin_for_test(&self, id: &str, raw: &str) -> anyhow::Result<()> {
let conn = self.lock_conn();
conn.execute(
"UPDATE conversations SET preference_pin = ?2 WHERE id = ?1",
rusqlite::params![id, raw],
)?;
Ok(())
}
pub fn preference_pin(&self, id: &str) -> anyhow::Result<Option<crate::OperatorPreferencePin>> {
let conn = self.lock_conn();
let json = conn
.query_row(
"SELECT preference_pin FROM conversations WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
|row| row.get::<_, String>(0),
)
.optional()?;
match json {
Some(json) => {
let pin = serde_json::from_str(&json).map_err(|e| {
anyhow::anyhow!(
"conversation `{id}`: preference_pin column is not a valid \
operator preference pin ({e}); refusing to load garbage"
)
})?;
Ok(Some(pin))
}
None => Ok(None),
}
}
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 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;
}
#[doc(hidden)]
pub fn set_liveness_for_test(&mut self, liveness: LivenessFn) {
self.liveness = liveness;
}
#[doc(hidden)]
pub fn set_owner_for_test(&mut self, host: &str, boot_id: &str, pid: i64) {
self.host = host.to_string();
self.boot_id = boot_id.to_string();
self.pid = pid;
}
fn prune_to_cap_excluding(&self, protected_id: &str) -> anyhow::Result<()> {
if self.max_per_workspace == 0 {
return Ok(());
}
let now = (self.claim_clock)();
let victims: Vec<String> = {
let conn = self.lock_conn();
let tx = rusqlite::Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)?;
let count: i64 = tx.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 candidates = {
let mut stmt = tx.prepare(
"SELECT c.id
FROM conversations c
WHERE c.workspace_key = ?1
AND c.id <> ?2
ORDER BY c.activity_tick ASC, c.id ASC",
)?;
let selected = stmt
.query_map(rusqlite::params![self.workspace_id, protected_id], |row| {
row.get::<_, String>(0)
})?
.collect::<Result<Vec<_>, _>>()?;
selected
};
let mut ids = Vec::with_capacity(excess as usize);
for id in candidates {
if ids.len() >= excess as usize {
break;
}
if let Some(owner) = live_owner_row(&tx, &id)? {
if (self.liveness)(&owner, now) {
continue;
}
tx.execute("DELETE FROM live_owners WHERE conversation_id = ?1", [&id])?;
}
ids.push(id);
}
for id in &ids {
tx.execute(
"DELETE FROM conversations WHERE id = ?1 AND workspace_key = ?2",
rusqlite::params![id, self.workspace_id],
)?;
}
tx.commit()?;
ids
};
for id in victims {
let plan_dir = self.workspace.join(session_plan_dir(&id));
let _ = std::fs::remove_dir_all(plan_dir);
}
Ok(())
}
pub(crate) fn workspace_fence(&self) -> &str {
&self.workspace_id
}
pub(crate) fn claim_tick(&self) -> i64 {
(self.claim_clock)()
}
pub(crate) 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,
}
fn is_content_id(s: &str) -> bool {
s.len() == 64
&& s.bytes()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}
fn parse_canonical_sources(stored: &str) -> anyhow::Result<Vec<String>> {
if stored == "[]" {
return Ok(Vec::new());
}
let ids: Vec<String> = serde_json::from_str(stored)
.map_err(|e| anyhow::anyhow!("not a JSON string array: {e}"))?;
for id in &ids {
if !is_content_id(id) {
anyhow::bail!("`{id}` is not a 64-lowercase-hex content id");
}
}
for pair in ids.windows(2) {
if pair[0] >= pair[1] {
anyhow::bail!("ids are not strictly sorted (duplicates included)");
}
}
let canonical = canonical_sources_json(&ids)?;
if canonical != stored {
anyhow::bail!("bytes differ from the canonical rendering");
}
Ok(ids)
}
fn canonical_sources_json(sources: &[String]) -> anyhow::Result<String> {
if sources.is_empty() {
return Ok("[]".to_string());
}
for id in sources {
if !is_content_id(id) {
anyhow::bail!(
"refusing the append — source reference `{id}` is not a content id \
(64 lowercase hex); a citation that cannot resolve must not be recorded"
);
}
}
let mut ids: Vec<&str> = sources.iter().map(String::as_str).collect();
ids.sort_unstable();
ids.dedup();
let mut out = String::with_capacity(2 + ids.len() * 67);
out.push('[');
for (i, id) in ids.iter().enumerate() {
if i > 0 {
out.push(',');
}
out.push('"');
out.push_str(id);
out.push('"');
}
out.push(']');
Ok(out)
}
fn migrate_roadmaps_pk(conn: &Connection) -> anyhow::Result<()> {
let sql: Option<String> = conn
.query_row(
"SELECT sql FROM sqlite_master WHERE type='table' AND name='roadmaps'",
[],
|r| r.get(0),
)
.optional()?;
let Some(sql) = sql else {
return Ok(());
};
let normalized = sql
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.to_ascii_lowercase();
if normalized.contains("primary key (id, workspace_key)") {
return Ok(()); }
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Immediate)?;
tx.execute_batch(
"CREATE TABLE roadmaps_v2 (
id TEXT NOT NULL,
workspace_key TEXT NOT NULL,
title TEXT NOT NULL DEFAULT '',
tree TEXT NOT NULL DEFAULT '',
schema_version INTEGER NOT NULL DEFAULT 1,
created_at_claim INTEGER NOT NULL DEFAULT 0,
updated_at_claim INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (id, workspace_key)
);
INSERT INTO roadmaps_v2
(id, workspace_key, title, tree, schema_version, created_at_claim, updated_at_claim)
SELECT id, workspace_key, title, tree, schema_version, created_at_claim, updated_at_claim
FROM roadmaps;
DROP TABLE roadmaps;
ALTER TABLE roadmaps_v2 RENAME TO roadmaps;
CREATE INDEX IF NOT EXISTS idx_roadmaps_ws ON roadmaps (workspace_key);",
)?;
tx.commit()?;
tracing::info!("migrated roadmaps to a composite (id, workspace_key) primary key (#1086)");
Ok(())
}
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)
}
const ROADMAP_SCHEMA_VERSION: i64 = 1;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Roadmap {
pub id: String,
pub title: String,
pub tree: crate::plan::Plan,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoadmapSummary {
pub id: String,
pub title: String,
pub node_count: usize,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InjectOutcome {
Enqueued,
Duplicate,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InjectedPrompt {
pub id: String,
pub body: String,
pub seq: i64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verdict {
AllowOnce,
AllowSession,
Deny,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingEnrollment {
pub request_id: String,
pub candidate_json: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AnswerOutcome {
Answered,
AlreadyResolved,
InvalidAction,
Unknown,
}
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)]
#[path = "store_tests/chain.rs"]
mod chain_tests;
#[cfg(test)]
#[path = "store_tests/claims.rs"]
mod claims_tests;
#[cfg(test)]
#[path = "store_tests/inbox.rs"]
mod inbox_tests;
#[cfg(test)]
#[path = "store_tests/liveness.rs"]
mod liveness_tests;
#[cfg(test)]
#[path = "store_tests/misc.rs"]
mod misc_tests;
#[cfg(test)]
#[path = "store_tests/prompts.rs"]
mod prompts_tests;
#[cfg(test)]
#[path = "store_tests/roadmaps.rs"]
mod roadmaps_tests;
#[cfg(test)]
#[path = "store_tests/schema.rs"]
mod schema_tests;