use crate::{
ClaimId, DailyReviewId, EvidenceId, EvidenceRelation, MemoryRef, PersonId, RetrievalItem,
RetrievalPack, SourceId, SourceKind, TenantId, Timestamp,
};
use rusqlite::{Connection, OptionalExtension, Transaction, params};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::{
collections::{HashMap, HashSet},
path::Path,
};
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
enum RetrievalTarget {
Source(SourceId),
Evidence(EvidenceId),
Claim(ClaimId),
}
pub trait Embedder {
type Error: std::error::Error + Send + Sync + 'static;
fn embed(&self, input: &str) -> std::result::Result<Embedding, Self::Error>;
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Embedding {
pub vector: Vec<f32>,
pub model: String,
pub version: String,
pub input_hash: String,
pub normalization: VectorNormalization,
pub distance: VectorDistance,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum VectorNormalization {
None,
L2,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum VectorDistance {
Cosine,
Dot,
Euclidean,
}
#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "id", rename_all = "snake_case")]
pub enum EmbeddingTarget {
Source(SourceId),
Evidence(EvidenceId),
Claim(ClaimId),
}
#[derive(Debug, Deserialize)]
pub struct EmbeddingInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub target: EmbeddingTarget,
pub embedding: Embedding,
}
#[derive(Debug, Serialize)]
pub struct StoredEmbedding {
pub target: EmbeddingTarget,
pub dimension: usize,
pub target_revision: i64,
pub input_hash: String,
pub created_at: Timestamp,
}
#[derive(Debug, Deserialize)]
pub struct ProjectionAuditInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub model: String,
pub version: String,
#[serde(default = "default_limit")]
pub limit: u32,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ProjectionState {
Missing,
Stale,
}
#[derive(Debug, Serialize)]
pub struct ProjectionInput {
pub target: EmbeddingTarget,
pub text: String,
pub target_revision: i64,
pub input_hash: String,
}
#[derive(Debug, Serialize)]
pub struct ProjectionIssue {
pub state: ProjectionState,
pub input: ProjectionInput,
pub stored_target_revision: Option<i64>,
pub stored_input_hash: Option<String>,
pub stored_created_at: Option<Timestamp>,
}
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error(transparent)]
Sql(#[from] rusqlite::Error),
#[error(transparent)]
Json(#[from] serde_json::Error),
#[error("{0}")]
Invalid(String),
#[error("record not found")]
NotFound,
}
pub type Result<T> = std::result::Result<T, Error>;
#[derive(Debug, Deserialize)]
pub struct RememberInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
#[serde(default)]
pub ingestion_key: Option<String>,
pub kind: SourceKind,
pub text: String,
pub captured_at: Timestamp,
pub claim: Option<ClaimInput>,
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct TranscriptLocator {
pub device_id: String,
pub provider: String,
pub stream_id: String,
pub segment_id: String,
pub start_ms: u64,
pub end_ms: u64,
}
#[derive(Debug, Deserialize)]
pub struct EvidenceLocatorInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub evidence_id: EvidenceId,
}
#[derive(Debug, Deserialize)]
pub struct RememberRequest {
#[serde(flatten)]
pub memory: RememberInput,
#[serde(default)]
pub locator: Option<TranscriptLocator>,
}
#[derive(Debug, Deserialize)]
pub struct ClaimInput {
pub subject: String,
pub predicate: String,
pub value: String,
pub valid_from: Timestamp,
}
#[derive(Debug, Serialize)]
pub struct Remembered {
pub source_id: SourceId,
pub evidence_id: EvidenceId,
pub claim_id: Option<ClaimId>,
}
#[derive(Debug, Deserialize)]
pub struct SearchInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub query: String,
#[serde(default = "default_limit")]
pub limit: u32,
#[serde(default)]
pub query_embedding: Option<DenseQuery>,
}
#[derive(Debug, Deserialize)]
pub struct GetInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub target: EmbeddingTarget,
}
#[derive(Debug, Deserialize)]
pub struct DenseQuery {
pub vector: Vec<f32>,
pub model: String,
pub version: String,
}
#[derive(Debug, Deserialize)]
pub struct CorrectInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub claim_id: ClaimId,
pub text: String,
pub value: String,
pub occurred_at: Timestamp,
}
#[derive(Debug, Serialize)]
pub struct Corrected {
pub source_id: SourceId,
pub evidence_id: EvidenceId,
pub claim_id: ClaimId,
pub superseded_claim_id: ClaimId,
}
#[derive(Debug, Deserialize)]
pub struct DeleteInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub source_id: SourceId,
pub deleted_at: Timestamp,
}
#[derive(Debug, Serialize)]
pub struct Deleted {
pub source_id: SourceId,
pub evidence_count: u64,
pub claim_count: u64,
}
#[derive(Debug, Deserialize)]
pub struct ReviewInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
pub day: String,
pub summary: String,
pub evidence_ids: Vec<EvidenceId>,
pub recorded_at: Timestamp,
}
#[derive(Debug, Deserialize)]
pub struct ReviewsInput {
pub tenant_id: TenantId,
pub person_id: PersonId,
#[serde(default = "default_limit")]
pub limit: u32,
}
#[derive(Debug, Serialize)]
pub struct StoredReview {
pub id: DailyReviewId,
}
#[derive(Debug, Serialize)]
pub struct ReviewRecord {
pub id: DailyReviewId,
pub day: String,
pub summary: String,
pub evidence_ids: Vec<EvidenceId>,
pub recorded_at: Timestamp,
}
pub struct MemoryDb {
connection: Connection,
}
impl MemoryDb {
pub fn open(path: impl AsRef<Path>) -> Result<Self> {
let connection = Connection::open(path)?;
let mut database = Self { connection };
database.migrate()?;
Ok(database)
}
pub fn remember(&mut self, input: RememberInput) -> Result<Remembered> {
self.remember_with_locator(input, None)
}
pub fn remember_with_locator(
&mut self,
input: RememberInput,
locator: Option<TranscriptLocator>,
) -> Result<Remembered> {
require_scope(&input.tenant_id, &input.person_id)?;
require_text("text", &input.text)?;
if let Some(locator) = &locator {
validate_transcript_locator(locator)?;
}
if let Some(key) = &input.ingestion_key {
require_text("ingestion_key", key)?;
}
let transaction = self.connection.transaction()?;
let source_id = SourceId(new_id(&transaction)?);
let evidence_id = EvidenceId(new_id(&transaction)?);
let kind = serde_json::to_string(&input.kind)?;
let inserted = transaction.execute(
"INSERT OR IGNORE INTO sources(id, tenant_id, person_id, ingestion_key, revision, kind, content, captured_at, recorded_at) VALUES(?1, ?2, ?3, ?4, 1, ?5, ?6, ?7, ?7)",
params![source_id.0, input.tenant_id.0, input.person_id.0, input.ingestion_key, kind, input.text, input.captured_at],
)?;
if inserted == 0 {
let replay = transaction.query_row(
"SELECT s.id, e.id, c.id, s.kind, s.content, s.captured_at, s.deleted_at, e.deleted_at, c.subject, c.predicate, c.value, c.valid_from FROM sources s JOIN evidence e ON e.source_id = s.id AND e.tenant_id = s.tenant_id AND e.person_id = s.person_id LEFT JOIN claim_evidence ce ON ce.evidence_id = e.id AND ce.tenant_id = e.tenant_id AND ce.person_id = e.person_id LEFT JOIN claims c ON c.id = ce.claim_id AND c.tenant_id = ce.tenant_id AND c.person_id = ce.person_id WHERE s.tenant_id = ?1 AND s.person_id = ?2 AND s.ingestion_key = ?3 ORDER BY e.id, c.id LIMIT 1",
params![input.tenant_id.0, input.person_id.0, input.ingestion_key],
|row| {
Ok((
Remembered {
source_id: SourceId(row.get(0)?),
evidence_id: EvidenceId(row.get(1)?),
claim_id: row.get::<_, Option<String>>(2)?.map(ClaimId),
},
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, i64>(5)?,
row.get::<_, Option<i64>>(6)?,
row.get::<_, Option<i64>>(7)?,
row.get::<_, Option<String>>(8)?,
row.get::<_, Option<String>>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, Option<i64>>(11)?,
))
},
).optional()?.ok_or(Error::NotFound)?;
if replay.4.is_some() || replay.5.is_some() {
return Err(Error::NotFound);
}
let stored_locator = transaction
.query_row(
"SELECT device_id, provider, stream_id, segment_id, start_ms, end_ms FROM evidence_locators WHERE tenant_id = ?1 AND person_id = ?2 AND evidence_id = ?3",
params![input.tenant_id.0, input.person_id.0, replay.0.evidence_id.0],
|row| {
Ok(TranscriptLocator {
device_id: row.get(0)?,
provider: row.get(1)?,
stream_id: row.get(2)?,
segment_id: row.get(3)?,
start_ms: row.get(4)?,
end_ms: row.get(5)?,
})
},
)
.optional()?;
let stored_claim = match (&replay.6, &replay.7, &replay.8, replay.9) {
(Some(subject), Some(predicate), Some(value), Some(valid_from)) => {
Some((subject, predicate, value, valid_from))
}
(None, None, None, None) => None,
_ => return Err(Error::NotFound),
};
let input_claim = input.claim.as_ref().map(|claim| {
(
&claim.subject,
&claim.predicate,
&claim.value,
claim.valid_from,
)
});
if replay.1 != kind
|| replay.2 != input.text
|| replay.3 != input.captured_at
|| stored_claim != input_claim
|| stored_locator.as_ref() != locator.as_ref()
{
return Err(Error::Invalid(
"ingestion_key conflicts with different memory payload".to_owned(),
));
}
transaction.commit()?;
return Ok(replay.0);
}
transaction.execute(
"INSERT INTO source_fts(source_id, tenant_id, person_id, content) VALUES(?1, ?2, ?3, ?4)",
params![source_id.0, input.tenant_id.0, input.person_id.0, input.text],
)?;
transaction.execute(
"INSERT INTO evidence(id, tenant_id, person_id, source_id, source_revision, quote, recorded_at) VALUES(?1, ?2, ?3, ?4, 1, ?5, ?6)",
params![evidence_id.0, input.tenant_id.0, input.person_id.0, source_id.0, input.text, input.captured_at],
)?;
if let Some(locator) = locator {
transaction.execute(
"INSERT INTO evidence_locators(tenant_id, person_id, evidence_id, device_id, provider, stream_id, segment_id, start_ms, end_ms) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![input.tenant_id.0, input.person_id.0, evidence_id.0, locator.device_id, locator.provider, locator.stream_id, locator.segment_id, locator.start_ms, locator.end_ms],
)?;
}
let claim_id = input
.claim
.map(|claim| {
insert_claim(
&transaction,
&input.tenant_id,
&input.person_id,
&evidence_id,
claim,
input.captured_at,
)
})
.transpose()?;
transaction.commit()?;
Ok(Remembered {
source_id,
evidence_id,
claim_id,
})
}
pub fn evidence_locator(
&self,
input: EvidenceLocatorInput,
) -> Result<Option<TranscriptLocator>> {
require_scope(&input.tenant_id, &input.person_id)?;
let locator = self
.connection
.query_row(
"SELECT l.device_id, l.provider, l.stream_id, l.segment_id, l.start_ms, l.end_ms FROM evidence_locators l JOIN evidence e ON e.id = l.evidence_id AND e.tenant_id = l.tenant_id AND e.person_id = l.person_id WHERE l.tenant_id = ?1 AND l.person_id = ?2 AND l.evidence_id = ?3 AND e.deleted_at IS NULL",
params![input.tenant_id.0, input.person_id.0, input.evidence_id.0],
|row| {
Ok(TranscriptLocator {
device_id: row.get(0)?,
provider: row.get(1)?,
stream_id: row.get(2)?,
segment_id: row.get(3)?,
start_ms: row.get(4)?,
end_ms: row.get(5)?,
})
},
)
.optional()?;
Ok(locator)
}
pub fn search(&self, input: SearchInput) -> Result<RetrievalPack> {
require_scope(&input.tenant_id, &input.person_id)?;
require_text("query", &input.query)?;
let limit = bounded_limit(input.limit);
let candidate_limit = limit * 4;
let query = format!("\"{}\"", input.query.replace('"', "\"\""));
let mut statement = self.connection.prepare(
"SELECT s.id, c.id
FROM source_fts
JOIN sources s ON s.id = source_fts.source_id AND s.tenant_id = source_fts.tenant_id AND s.person_id = source_fts.person_id
JOIN evidence e ON e.source_id = s.id AND e.tenant_id = s.tenant_id AND e.person_id = s.person_id AND e.deleted_at IS NULL
LEFT JOIN claim_evidence ce ON ce.evidence_id = e.id AND ce.tenant_id = e.tenant_id AND ce.person_id = e.person_id
LEFT JOIN claims c ON c.id = ce.claim_id AND c.tenant_id = ce.tenant_id AND c.person_id = ce.person_id AND c.status = 'accepted'
WHERE source_fts MATCH ?1 AND source_fts.tenant_id = ?2 AND source_fts.person_id = ?3 AND s.deleted_at IS NULL
AND (c.id IS NOT NULL OR NOT EXISTS (
SELECT 1 FROM evidence live_e
JOIN claim_evidence live_ce ON live_ce.evidence_id = live_e.id AND live_ce.tenant_id = live_e.tenant_id AND live_ce.person_id = live_e.person_id
JOIN claims live_c ON live_c.id = live_ce.claim_id AND live_c.tenant_id = live_ce.tenant_id AND live_c.person_id = live_ce.person_id
WHERE live_e.source_id = s.id AND live_e.tenant_id = s.tenant_id AND live_e.person_id = s.person_id AND live_e.deleted_at IS NULL AND live_c.status = 'accepted'
))
ORDER BY bm25(source_fts), s.id, c.id LIMIT ?4",
)?;
let rows = statement.query_map(
params![query, input.tenant_id.0, input.person_id.0, candidate_limit],
|row| {
let source_id = row.get::<_, String>(0)?;
Ok(match row.get::<_, Option<String>>(1)? {
Some(claim_id) => RetrievalTarget::Claim(ClaimId(claim_id)),
None => RetrievalTarget::Source(SourceId(source_id)),
})
},
)?;
let lexical = rows.collect::<std::result::Result<Vec<_>, _>>()?;
let dense = input
.query_embedding
.as_ref()
.map(|query| self.dense_claims(&input.tenant_id, &input.person_id, query))
.transpose()?
.unwrap_or_default();
let ranked = reciprocal_rank_fusion(&lexical, &dense, limit as usize);
let mut items = Vec::with_capacity(ranked.len());
for (target, relevance_basis_points) in ranked {
items.push(self.retrieval_item(
&input.tenant_id,
&input.person_id,
target,
relevance_basis_points,
)?);
}
let gaps = if items.is_empty() {
vec!["no cited memory matched".to_owned()]
} else {
Vec::new()
};
Ok(RetrievalPack {
query: input.query,
items,
gaps,
})
}
pub fn get(&self, input: GetInput) -> Result<RetrievalItem> {
require_scope(&input.tenant_id, &input.person_id)?;
let target = match input.target {
EmbeddingTarget::Source(id) => RetrievalTarget::Source(id),
EmbeddingTarget::Evidence(id) => RetrievalTarget::Evidence(id),
EmbeddingTarget::Claim(id) => RetrievalTarget::Claim(id),
};
self.retrieval_item(&input.tenant_id, &input.person_id, target, 10_000)
}
fn dense_claims(
&self,
tenant_id: &TenantId,
person_id: &PersonId,
query: &DenseQuery,
) -> Result<Vec<RetrievalTarget>> {
validate_dense_query(query)?;
let mut statement = self.connection.prepare(
"SELECT embeddings.target_kind, embeddings.target_id, embeddings.dimension, embeddings.normalization, embeddings.distance, embeddings.vector, embeddings.target_revision, embeddings.input_hash
FROM embeddings
WHERE embeddings.tenant_id = ?1 AND embeddings.person_id = ?2 AND embeddings.model = ?3 AND embeddings.version = ?4",
)?;
let rows = statement.query_map(
params![tenant_id.0, person_id.0, query.model, query.version],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, usize>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, i64>(6)?,
row.get::<_, String>(7)?,
))
},
)?;
let mut scores = HashMap::<RetrievalTarget, f32>::new();
let mut lane = None;
for row in rows {
let (
target_kind,
target_id,
dimension,
normalization,
distance,
vector,
target_revision,
input_hash,
) = row?;
let target = embedding_target(&target_kind, &target_id)?;
let current = match self.projection_input(tenant_id, person_id, target) {
Ok(current) => current,
Err(Error::NotFound) => continue,
Err(error) => return Err(error),
};
if current.target_revision != target_revision || current.input_hash != input_hash {
continue;
}
if dimension != query.vector.len() {
return Err(Error::Invalid(format!(
"query embedding dimension {} does not match stored dimension {dimension}",
query.vector.len()
)));
}
let normalization: VectorNormalization = serde_json::from_str(&normalization)?;
let distance: VectorDistance = serde_json::from_str(&distance)?;
let configuration = (normalization, distance, dimension);
if lane.as_ref().is_some_and(|lane| lane != &configuration) {
return Err(Error::Invalid(
"stored embeddings mix incompatible vector configurations".to_owned(),
));
}
lane = Some(configuration.clone());
let vector: Vec<f32> = serde_json::from_str(&vector)?;
if vector.len() != dimension || vector.iter().any(|value| !value.is_finite()) {
return Err(Error::Invalid("stored embedding is invalid".to_owned()));
}
let score = vector_score(&query.vector, &vector, &configuration.1)?;
for target in self.retrieval_targets_for_embedding(
tenant_id,
person_id,
&target_kind,
&target_id,
)? {
scores
.entry(target)
.and_modify(|existing| *existing = existing.max(score))
.or_insert(score);
}
}
let mut ranked = scores.into_iter().collect::<Vec<_>>();
ranked.sort_by(|left, right| {
right
.1
.total_cmp(&left.1)
.then_with(|| left.0.cmp(&right.0))
});
Ok(ranked.into_iter().map(|(target, _)| target).collect())
}
fn retrieval_targets_for_embedding(
&self,
tenant_id: &TenantId,
person_id: &PersonId,
target_kind: &str,
target_id: &str,
) -> Result<Vec<RetrievalTarget>> {
let sql = match target_kind {
"claim" => {
"SELECT id FROM claims WHERE id = ?1 AND tenant_id = ?2 AND person_id = ?3 AND status = 'accepted'"
}
"evidence" => {
"SELECT c.id FROM evidence e JOIN sources s ON s.id = e.source_id AND s.tenant_id = e.tenant_id AND s.person_id = e.person_id LEFT JOIN claim_evidence ce ON ce.evidence_id = e.id AND ce.tenant_id = e.tenant_id AND ce.person_id = e.person_id LEFT JOIN claims c ON c.id = ce.claim_id AND c.tenant_id = ce.tenant_id AND c.person_id = ce.person_id AND c.status = 'accepted' WHERE e.id = ?1 AND e.tenant_id = ?2 AND e.person_id = ?3 AND e.deleted_at IS NULL AND s.deleted_at IS NULL ORDER BY c.id"
}
"source" => {
"SELECT DISTINCT c.id FROM sources s JOIN evidence e ON e.source_id = s.id AND e.tenant_id = s.tenant_id AND e.person_id = s.person_id LEFT JOIN claim_evidence ce ON ce.evidence_id = e.id AND ce.tenant_id = e.tenant_id AND ce.person_id = e.person_id LEFT JOIN claims c ON c.id = ce.claim_id AND c.tenant_id = ce.tenant_id AND c.person_id = ce.person_id AND c.status = 'accepted' WHERE s.id = ?1 AND s.tenant_id = ?2 AND s.person_id = ?3 AND s.deleted_at IS NULL AND e.deleted_at IS NULL ORDER BY c.id"
}
_ => {
return Err(Error::Invalid(
"stored embedding target is invalid".to_owned(),
));
}
};
let mut statement = self.connection.prepare(sql)?;
let rows = statement.query_map(params![target_id, tenant_id.0, person_id.0], |row| {
row.get::<_, Option<String>>(0)
})?;
let rows = rows.collect::<std::result::Result<Vec<_>, _>>()?;
if rows.is_empty() {
return Ok(Vec::new());
}
let claims = rows
.into_iter()
.flatten()
.map(|id| RetrievalTarget::Claim(ClaimId(id)))
.collect::<Vec<_>>();
if !claims.is_empty() {
return Ok(claims);
}
Ok(match target_kind {
"source" => vec![RetrievalTarget::Source(SourceId(target_id.to_owned()))],
"evidence" => vec![RetrievalTarget::Evidence(EvidenceId(target_id.to_owned()))],
"claim" => Vec::new(),
_ => unreachable!(),
})
}
fn retrieval_item(
&self,
tenant_id: &TenantId,
person_id: &PersonId,
target: RetrievalTarget,
relevance_basis_points: u16,
) -> Result<RetrievalItem> {
let (sql, id) = match &target {
RetrievalTarget::Claim(id) => (
"SELECT c.subject || ' ' || c.predicate || ' ' || c.value, ce.evidence_id
FROM claims c
JOIN claim_evidence ce ON ce.claim_id = c.id AND ce.tenant_id = c.tenant_id AND ce.person_id = c.person_id
JOIN evidence e ON e.id = ce.evidence_id AND e.tenant_id = ce.tenant_id AND e.person_id = ce.person_id
JOIN sources s ON s.id = e.source_id AND s.tenant_id = e.tenant_id AND s.person_id = e.person_id
WHERE c.id = ?1 AND c.tenant_id = ?2 AND c.person_id = ?3 AND c.status = 'accepted' AND e.deleted_at IS NULL AND s.deleted_at IS NULL
ORDER BY ce.evidence_id LIMIT 1",
&id.0,
),
RetrievalTarget::Source(id) => (
"SELECT s.content, e.id FROM sources s JOIN evidence e ON e.source_id = s.id AND e.tenant_id = s.tenant_id AND e.person_id = s.person_id WHERE s.id = ?1 AND s.tenant_id = ?2 AND s.person_id = ?3 AND s.deleted_at IS NULL AND e.deleted_at IS NULL ORDER BY e.id LIMIT 1",
&id.0,
),
RetrievalTarget::Evidence(id) => (
"SELECT e.quote, e.id FROM evidence e JOIN sources s ON s.id = e.source_id AND s.tenant_id = e.tenant_id AND s.person_id = e.person_id WHERE e.id = ?1 AND e.tenant_id = ?2 AND e.person_id = ?3 AND e.deleted_at IS NULL AND s.deleted_at IS NULL",
&id.0,
),
};
let memory = match &target {
RetrievalTarget::Claim(id) => MemoryRef::Claim(id.clone()),
RetrievalTarget::Source(id) => MemoryRef::Source(id.clone()),
RetrievalTarget::Evidence(id) => MemoryRef::Evidence(id.clone()),
};
self.connection
.query_row(sql, params![id, tenant_id.0, person_id.0], |row| {
Ok(RetrievalItem {
memory,
excerpt: row.get(0)?,
relevance_basis_points,
evidence_ids: vec![EvidenceId(row.get(1)?)],
})
})
.optional()?
.ok_or(Error::NotFound)
}
pub fn correct(&mut self, input: CorrectInput) -> Result<Corrected> {
require_scope(&input.tenant_id, &input.person_id)?;
require_text("correction text", &input.text)?;
require_text("value", &input.value)?;
let transaction = self.connection.transaction()?;
let old = transaction
.query_row(
"SELECT subject, predicate, valid_from, recorded_from FROM claims WHERE id = ?1 AND tenant_id = ?2 AND person_id = ?3 AND status = 'accepted'",
params![input.claim_id.0, input.tenant_id.0, input.person_id.0],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, i64>(2)?, row.get::<_, i64>(3)?)),
)
.optional()?
.ok_or(Error::NotFound)?;
if input.occurred_at <= old.2 || input.occurred_at <= old.3 {
return Err(Error::Invalid(
"occurred_at must be after the original claim became valid and was recorded"
.to_owned(),
));
}
let source_id = SourceId(new_id(&transaction)?);
let evidence_id = EvidenceId(new_id(&transaction)?);
transaction.execute(
"INSERT INTO sources(id, tenant_id, person_id, revision, kind, content, captured_at, recorded_at) VALUES(?1, ?2, ?3, 1, '\"user_correction\"', ?4, ?5, ?5)",
params![source_id.0, input.tenant_id.0, input.person_id.0, input.text, input.occurred_at],
)?;
transaction.execute(
"INSERT INTO source_fts(source_id, tenant_id, person_id, content) VALUES(?1, ?2, ?3, ?4)",
params![source_id.0, input.tenant_id.0, input.person_id.0, input.text],
)?;
transaction.execute(
"INSERT INTO evidence(id, tenant_id, person_id, source_id, source_revision, quote, recorded_at) VALUES(?1, ?2, ?3, ?4, 1, ?5, ?6)",
params![evidence_id.0, input.tenant_id.0, input.person_id.0, source_id.0, input.text, input.occurred_at],
)?;
transaction.execute(
"UPDATE claims SET status = 'superseded', valid_until = ?1, recorded_until = ?1 WHERE id = ?2 AND tenant_id = ?3 AND person_id = ?4",
params![input.occurred_at, input.claim_id.0, input.tenant_id.0, input.person_id.0],
)?;
let claim_id = insert_claim(
&transaction,
&input.tenant_id,
&input.person_id,
&evidence_id,
ClaimInput {
subject: old.0,
predicate: old.1,
value: input.value,
valid_from: input.occurred_at.max(old.2),
},
input.occurred_at,
)?;
transaction.commit()?;
Ok(Corrected {
source_id,
evidence_id,
claim_id,
superseded_claim_id: input.claim_id,
})
}
pub fn delete_source(&mut self, input: DeleteInput) -> Result<Deleted> {
require_scope(&input.tenant_id, &input.person_id)?;
let transaction = self.connection.transaction()?;
let recorded_at = transaction
.query_row(
"SELECT recorded_at FROM sources WHERE id = ?1 AND tenant_id = ?2 AND person_id = ?3 AND deleted_at IS NULL",
params![input.source_id.0, input.tenant_id.0, input.person_id.0],
|row| row.get::<_, i64>(0),
)
.optional()?
.ok_or(Error::NotFound)?;
if input.deleted_at < recorded_at {
return Err(Error::Invalid(
"deleted_at cannot predate source recording".to_owned(),
));
}
let changed = transaction.execute(
"UPDATE sources SET deleted_at = ?1, revision = revision + 1 WHERE id = ?2 AND tenant_id = ?3 AND person_id = ?4 AND deleted_at IS NULL",
params![input.deleted_at, input.source_id.0, input.tenant_id.0, input.person_id.0],
)?;
if changed == 0 {
return Err(Error::NotFound);
}
transaction.execute(
"DELETE FROM source_fts WHERE source_id = ?1 AND tenant_id = ?2 AND person_id = ?3",
params![input.source_id.0, input.tenant_id.0, input.person_id.0],
)?;
let evidence_count = transaction.execute(
"UPDATE evidence SET deleted_at = ?1 WHERE source_id = ?2 AND tenant_id = ?3 AND person_id = ?4 AND deleted_at IS NULL",
params![input.deleted_at, input.source_id.0, input.tenant_id.0, input.person_id.0],
)? as u64;
let claim_count = transaction.execute(
"UPDATE claims SET status = 'rejected', recorded_until = ?1
WHERE tenant_id = ?2 AND person_id = ?3 AND status = 'accepted'
AND id IN (SELECT ce.claim_id FROM claim_evidence ce JOIN evidence e ON e.id = ce.evidence_id AND e.tenant_id = ce.tenant_id AND e.person_id = ce.person_id WHERE e.source_id = ?4)
AND NOT EXISTS (SELECT 1 FROM claim_evidence live_ce JOIN evidence live_e ON live_e.id = live_ce.evidence_id AND live_e.tenant_id = live_ce.tenant_id AND live_e.person_id = live_ce.person_id WHERE live_ce.claim_id = claims.id AND live_e.deleted_at IS NULL)",
params![input.deleted_at, input.tenant_id.0, input.person_id.0, input.source_id.0],
)? as u64;
transaction.execute(
"DELETE FROM embeddings WHERE tenant_id = ?1 AND person_id = ?2 AND ((target_kind = 'source' AND target_id = ?3) OR (target_kind = 'evidence' AND target_id IN (SELECT id FROM evidence WHERE source_id = ?3 AND tenant_id = ?1 AND person_id = ?2)) OR (target_kind = 'claim' AND target_id IN (SELECT id FROM claims WHERE tenant_id = ?1 AND person_id = ?2 AND status = 'rejected')))",
params![input.tenant_id.0, input.person_id.0, input.source_id.0],
)?;
transaction.execute(
"DELETE FROM daily_reviews WHERE tenant_id = ?1 AND person_id = ?2 AND EXISTS (SELECT 1 FROM json_each(evidence_ids) citation JOIN evidence e ON e.id = citation.value WHERE e.source_id = ?3 AND e.tenant_id = ?1 AND e.person_id = ?2)",
params![input.tenant_id.0, input.person_id.0, input.source_id.0],
)?;
transaction.commit()?;
Ok(Deleted {
source_id: input.source_id,
evidence_count,
claim_count,
})
}
pub fn store_review(&mut self, input: ReviewInput) -> Result<StoredReview> {
require_scope(&input.tenant_id, &input.person_id)?;
require_text("day", &input.day)?;
require_text("summary", &input.summary)?;
if input.evidence_ids.is_empty() {
return Err(Error::Invalid("review needs evidence_ids".to_owned()));
}
let transaction = self.connection.transaction()?;
for evidence_id in &input.evidence_ids {
let found: bool = transaction.query_row(
"SELECT EXISTS(SELECT 1 FROM evidence WHERE id = ?1 AND tenant_id = ?2 AND person_id = ?3 AND deleted_at IS NULL)",
params![evidence_id.0, input.tenant_id.0, input.person_id.0],
|row| row.get(0),
)?;
if !found {
return Err(Error::Invalid(format!(
"evidence {} is unavailable",
evidence_id.0
)));
}
}
let id = DailyReviewId(new_id(&transaction)?);
transaction.execute(
"INSERT INTO daily_reviews(id, tenant_id, person_id, day, summary, evidence_ids, recorded_at) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![id.0, input.tenant_id.0, input.person_id.0, input.day, input.summary, serde_json::to_string(&input.evidence_ids)?, input.recorded_at],
)?;
transaction.commit()?;
Ok(StoredReview { id })
}
pub fn reviews(&self, input: ReviewsInput) -> Result<Vec<ReviewRecord>> {
require_scope(&input.tenant_id, &input.person_id)?;
let mut statement = self.connection.prepare(
"SELECT id, day, summary, evidence_ids, recorded_at FROM daily_reviews WHERE tenant_id = ?1 AND person_id = ?2 ORDER BY day DESC, recorded_at DESC LIMIT ?3",
)?;
let rows = statement.query_map(
params![
input.tenant_id.0,
input.person_id.0,
bounded_limit(input.limit)
],
|row| {
let json: String = row.get(3)?;
let evidence_ids = serde_json::from_str(&json).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Text,
Box::new(error),
)
})?;
Ok(ReviewRecord {
id: DailyReviewId(row.get(0)?),
day: row.get(1)?,
summary: row.get(2)?,
evidence_ids,
recorded_at: row.get(4)?,
})
},
)?;
Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
}
pub fn rebuild_embedding<E: Embedder>(
&mut self,
tenant_id: TenantId,
person_id: PersonId,
target: EmbeddingTarget,
embedder: &E,
) -> Result<StoredEmbedding> {
require_scope(&tenant_id, &person_id)?;
let projection = self.projection_input(&tenant_id, &person_id, target.clone())?;
let embedding = embedder
.embed(&projection.text)
.map_err(|error| Error::Invalid(format!("embedder failed: {error}")))?;
self.upsert_embedding(EmbeddingInput {
tenant_id,
person_id,
target,
embedding,
})
}
pub fn upsert_embedding(&mut self, input: EmbeddingInput) -> Result<StoredEmbedding> {
require_scope(&input.tenant_id, &input.person_id)?;
validate_embedding(&input.embedding)?;
let projection =
self.projection_input(&input.tenant_id, &input.person_id, input.target.clone())?;
if input.embedding.input_hash != projection.input_hash {
return Err(Error::Invalid(format!(
"embedding input_hash does not match current target input; expected {}",
projection.input_hash
)));
}
let (target_kind, target_id) = embedding_target_parts(&input.target);
let dimension = input.embedding.vector.len();
let created_at = self
.connection
.query_row("SELECT unixepoch()", [], |row| row.get(0))?;
self.connection.execute(
"INSERT INTO embeddings(tenant_id, person_id, target_kind, target_id, model, version, dimension, input_hash, target_revision, created_at, normalization, distance, vector)
VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
ON CONFLICT(tenant_id, person_id, target_kind, target_id, model, version) DO UPDATE SET dimension = excluded.dimension, input_hash = excluded.input_hash, target_revision = excluded.target_revision, created_at = excluded.created_at, normalization = excluded.normalization, distance = excluded.distance, vector = excluded.vector",
params![input.tenant_id.0, input.person_id.0, target_kind, target_id, input.embedding.model, input.embedding.version, dimension, projection.input_hash, projection.target_revision, created_at, serde_json::to_string(&input.embedding.normalization)?, serde_json::to_string(&input.embedding.distance)?, serde_json::to_string(&input.embedding.vector)?],
)?;
Ok(StoredEmbedding {
target: input.target,
dimension,
target_revision: projection.target_revision,
input_hash: projection.input_hash,
created_at,
})
}
pub fn projection_input(
&self,
tenant_id: &TenantId,
person_id: &PersonId,
target: EmbeddingTarget,
) -> Result<ProjectionInput> {
require_scope(tenant_id, person_id)?;
let (table, id, expression, revision, live) = match &target {
EmbeddingTarget::Source(id) => (
"sources",
&id.0,
"content",
"revision",
"deleted_at IS NULL",
),
EmbeddingTarget::Evidence(id) => (
"evidence",
&id.0,
"quote",
"source_revision",
"deleted_at IS NULL",
),
EmbeddingTarget::Claim(id) => (
"claims",
&id.0,
"subject || ' ' || predicate || ' ' || value",
"recorded_from",
"status = 'accepted'",
),
};
let (text, target_revision) = self
.connection
.query_row(
&format!("SELECT {expression}, {revision} FROM {table} WHERE id = ?1 AND tenant_id = ?2 AND person_id = ?3 AND {live}"),
params![id, tenant_id.0, person_id.0],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)),
)
.optional()?
.ok_or(Error::NotFound)?;
Ok(ProjectionInput {
target,
input_hash: input_hash(&text),
text,
target_revision,
})
}
pub fn projection_issues(&self, input: ProjectionAuditInput) -> Result<Vec<ProjectionIssue>> {
require_scope(&input.tenant_id, &input.person_id)?;
require_text("embedding model", &input.model)?;
require_text("embedding version", &input.version)?;
let mut statement = self.connection.prepare(
"SELECT target_kind, target_id FROM (
SELECT 'source' AS target_kind, id AS target_id FROM sources WHERE tenant_id = ?1 AND person_id = ?2 AND deleted_at IS NULL
UNION ALL SELECT 'evidence', e.id FROM evidence e JOIN sources s ON s.id = e.source_id AND s.tenant_id = e.tenant_id AND s.person_id = e.person_id WHERE e.tenant_id = ?1 AND e.person_id = ?2 AND e.deleted_at IS NULL AND s.deleted_at IS NULL
UNION ALL SELECT 'claim', c.id FROM claims c WHERE c.tenant_id = ?1 AND c.person_id = ?2 AND c.status = 'accepted'
) ORDER BY target_kind, target_id",
)?;
let targets = statement
.query_map(params![input.tenant_id.0, input.person_id.0], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?
.collect::<std::result::Result<Vec<_>, _>>()?;
let mut issues = Vec::new();
let limit = bounded_limit(input.limit) as usize;
for (kind, id) in targets {
let projection = self.projection_input(
&input.tenant_id,
&input.person_id,
embedding_target(&kind, &id)?,
)?;
let stored = self
.connection
.query_row(
"SELECT target_revision, input_hash, created_at FROM embeddings WHERE tenant_id = ?1 AND person_id = ?2 AND target_kind = ?3 AND target_id = ?4 AND model = ?5 AND version = ?6",
params![input.tenant_id.0, input.person_id.0, kind, embedding_target_parts(&projection.target).1, input.model, input.version],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, i64>(2)?)),
)
.optional()?;
let state = match &stored {
None => ProjectionState::Missing,
Some((revision, hash, _))
if *revision != projection.target_revision
|| hash != &projection.input_hash =>
{
ProjectionState::Stale
}
Some(_) => continue,
};
issues.push(ProjectionIssue {
state,
input: projection,
stored_target_revision: stored.as_ref().map(|value| value.0),
stored_input_hash: stored.as_ref().map(|value| value.1.clone()),
stored_created_at: stored.map(|value| value.2),
});
if issues.len() == limit {
break;
}
}
Ok(issues)
}
fn migrate(&mut self) -> Result<()> {
self.connection.execute_batch(
"PRAGMA foreign_keys = ON;
BEGIN IMMEDIATE;
CREATE TABLE IF NOT EXISTS sources(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, ingestion_key TEXT, revision INTEGER NOT NULL, kind TEXT NOT NULL, content TEXT NOT NULL, captured_at INTEGER NOT NULL, recorded_at INTEGER NOT NULL, deleted_at INTEGER);
CREATE INDEX IF NOT EXISTS sources_scope ON sources(tenant_id, person_id, id);
CREATE VIRTUAL TABLE IF NOT EXISTS source_fts USING fts5(source_id UNINDEXED, tenant_id UNINDEXED, person_id UNINDEXED, content, tokenize='unicode61');
CREATE TABLE IF NOT EXISTS evidence(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, source_id TEXT NOT NULL REFERENCES sources(id), source_revision INTEGER NOT NULL, quote TEXT NOT NULL, recorded_at INTEGER NOT NULL, deleted_at INTEGER);
CREATE INDEX IF NOT EXISTS evidence_scope ON evidence(tenant_id, person_id, source_id);
CREATE TABLE IF NOT EXISTS evidence_locators(tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, evidence_id TEXT NOT NULL REFERENCES evidence(id), device_id TEXT NOT NULL, provider TEXT NOT NULL, stream_id TEXT NOT NULL, segment_id TEXT NOT NULL, start_ms INTEGER NOT NULL, end_ms INTEGER NOT NULL, PRIMARY KEY(tenant_id, person_id, evidence_id));
CREATE TABLE IF NOT EXISTS claims(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, subject TEXT NOT NULL, predicate TEXT NOT NULL, value TEXT NOT NULL, valid_from INTEGER NOT NULL, valid_until INTEGER, recorded_from INTEGER NOT NULL, recorded_until INTEGER, status TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS claims_scope ON claims(tenant_id, person_id, status);
CREATE TABLE IF NOT EXISTS claim_evidence(tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, claim_id TEXT NOT NULL REFERENCES claims(id), evidence_id TEXT NOT NULL REFERENCES evidence(id), relation TEXT NOT NULL, confidence_basis_points INTEGER NOT NULL, PRIMARY KEY(tenant_id, person_id, claim_id, evidence_id));
CREATE TABLE IF NOT EXISTS daily_reviews(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, day TEXT NOT NULL, summary TEXT NOT NULL, evidence_ids TEXT NOT NULL, recorded_at INTEGER NOT NULL);
CREATE INDEX IF NOT EXISTS reviews_scope ON daily_reviews(tenant_id, person_id, day);
CREATE TABLE IF NOT EXISTS embeddings(tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, target_kind TEXT NOT NULL, target_id TEXT NOT NULL, model TEXT NOT NULL, version TEXT NOT NULL, dimension INTEGER NOT NULL, input_hash TEXT NOT NULL, target_revision INTEGER NOT NULL, created_at INTEGER NOT NULL, normalization TEXT NOT NULL, distance TEXT NOT NULL, vector TEXT NOT NULL, PRIMARY KEY(tenant_id, person_id, target_kind, target_id, model, version));
CREATE INDEX IF NOT EXISTS embeddings_scope ON embeddings(tenant_id, person_id, target_kind, target_id);
UPDATE claims SET valid_until = recorded_until WHERE status = 'superseded' AND valid_until IS NULL AND recorded_until > valid_from;
PRAGMA user_version = 3;
COMMIT;",
)?;
for (column, definition) in [
("target_revision", "INTEGER NOT NULL DEFAULT 0"),
("created_at", "INTEGER NOT NULL DEFAULT 0"),
] {
let exists: bool = self.connection.query_row(
"SELECT EXISTS(SELECT 1 FROM pragma_table_info('embeddings') WHERE name = ?1)",
[column],
|row| row.get(0),
)?;
if !exists {
self.connection.execute(
&format!("ALTER TABLE embeddings ADD COLUMN {column} {definition}"),
[],
)?;
}
}
let has_ingestion_key: bool = self.connection.query_row(
"SELECT EXISTS(SELECT 1 FROM pragma_table_info('sources') WHERE name = 'ingestion_key')",
[],
|row| row.get(0),
)?;
if !has_ingestion_key {
self.connection
.execute("ALTER TABLE sources ADD COLUMN ingestion_key TEXT", [])?;
}
self.connection.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS sources_ingestion_key ON sources(tenant_id, person_id, ingestion_key) WHERE ingestion_key IS NOT NULL",
[],
)?;
Ok(())
}
}
fn insert_claim(
transaction: &Transaction<'_>,
tenant_id: &TenantId,
person_id: &PersonId,
evidence_id: &EvidenceId,
claim: ClaimInput,
recorded_at: Timestamp,
) -> Result<ClaimId> {
require_text("claim subject", &claim.subject)?;
require_text("claim predicate", &claim.predicate)?;
require_text("claim value", &claim.value)?;
let id = ClaimId(new_id(transaction)?);
transaction.execute(
"INSERT INTO claims(id, tenant_id, person_id, subject, predicate, value, valid_from, recorded_from, status) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'accepted')",
params![id.0, tenant_id.0, person_id.0, claim.subject, claim.predicate, claim.value, claim.valid_from, recorded_at],
)?;
let relation = serde_json::to_string(&EvidenceRelation::Supports)?;
transaction.execute(
"INSERT INTO claim_evidence(tenant_id, person_id, claim_id, evidence_id, relation, confidence_basis_points) VALUES(?1, ?2, ?3, ?4, ?5, 10000)",
params![tenant_id.0, person_id.0, id.0, evidence_id.0, relation],
)?;
Ok(id)
}
fn validate_transcript_locator(locator: &TranscriptLocator) -> Result<()> {
for (field, value) in [
("locator.device_id", locator.device_id.as_str()),
("locator.provider", locator.provider.as_str()),
("locator.stream_id", locator.stream_id.as_str()),
("locator.segment_id", locator.segment_id.as_str()),
] {
require_text(field, value)?;
}
if locator.start_ms > locator.end_ms {
return Err(Error::Invalid(
"locator end_ms must not precede start_ms".to_owned(),
));
}
if locator.end_ms > i64::MAX as u64 {
return Err(Error::Invalid(
"locator end_ms exceeds the storage range".to_owned(),
));
}
Ok(())
}
fn new_id(transaction: &Transaction<'_>) -> Result<String> {
Ok(transaction.query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0))?)
}
fn require_scope(tenant_id: &TenantId, person_id: &PersonId) -> Result<()> {
require_text("tenant_id", &tenant_id.0)?;
require_text("person_id", &person_id.0)
}
fn require_text(field: &str, value: &str) -> Result<()> {
if value.trim().is_empty() {
return Err(Error::Invalid(format!("{field} must not be empty")));
}
Ok(())
}
fn validate_embedding(embedding: &Embedding) -> Result<()> {
require_text("embedding model", &embedding.model)?;
require_text("embedding version", &embedding.version)?;
require_text("embedding input_hash", &embedding.input_hash)?;
if embedding.vector.is_empty() || embedding.vector.iter().any(|value| !value.is_finite()) {
return Err(Error::Invalid(
"embedding vector must contain finite values".to_owned(),
));
}
Ok(())
}
fn input_hash(text: &str) -> String {
format!("sha256:{:x}", Sha256::digest(text.as_bytes()))
}
fn validate_dense_query(query: &DenseQuery) -> Result<()> {
require_text("query embedding model", &query.model)?;
require_text("query embedding version", &query.version)?;
if query.vector.is_empty() || query.vector.iter().any(|value| !value.is_finite()) {
return Err(Error::Invalid(
"query embedding vector must contain finite values".to_owned(),
));
}
Ok(())
}
fn vector_score(query: &[f32], stored: &[f32], distance: &VectorDistance) -> Result<f32> {
let dot = query
.iter()
.zip(stored)
.map(|(left, right)| left * right)
.sum::<f32>();
let score = match distance {
VectorDistance::Dot => dot,
VectorDistance::Euclidean => -query
.iter()
.zip(stored)
.map(|(left, right)| (left - right).powi(2))
.sum::<f32>()
.sqrt(),
VectorDistance::Cosine => {
let query_norm = query.iter().map(|value| value * value).sum::<f32>().sqrt();
let stored_norm = stored.iter().map(|value| value * value).sum::<f32>().sqrt();
if query_norm == 0.0 || stored_norm == 0.0 {
return Err(Error::Invalid(
"cosine embeddings must have non-zero magnitude".to_owned(),
));
}
dot / (query_norm * stored_norm)
}
};
if !score.is_finite() {
return Err(Error::Invalid(
"embedding similarity is not finite".to_owned(),
));
}
Ok(score)
}
fn reciprocal_rank_fusion(
lexical: &[RetrievalTarget],
dense: &[RetrievalTarget],
limit: usize,
) -> Vec<(RetrievalTarget, u16)> {
let mut scores = HashMap::<RetrievalTarget, u32>::new();
for ranking in [lexical, dense] {
let mut seen = HashSet::new();
for (offset, id) in ranking.iter().enumerate() {
if seen.insert(id) {
*scores.entry(id.clone()).or_default() += 1_000_000 / (61 + offset as u32);
}
}
}
let maximum = scores.values().copied().max().unwrap_or(1);
let mut ranked = scores
.into_iter()
.map(|(id, score)| (id, ((score as u64 * 10_000) / maximum as u64) as u16))
.collect::<Vec<_>>();
ranked.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
ranked.truncate(limit);
ranked
}
fn embedding_target_parts(target: &EmbeddingTarget) -> (&'static str, &str) {
match target {
EmbeddingTarget::Source(id) => ("source", &id.0),
EmbeddingTarget::Evidence(id) => ("evidence", &id.0),
EmbeddingTarget::Claim(id) => ("claim", &id.0),
}
}
fn embedding_target(kind: &str, id: &str) -> Result<EmbeddingTarget> {
Ok(match kind {
"source" => EmbeddingTarget::Source(SourceId(id.to_owned())),
"evidence" => EmbeddingTarget::Evidence(EvidenceId(id.to_owned())),
"claim" => EmbeddingTarget::Claim(ClaimId(id.to_owned())),
_ => {
return Err(Error::Invalid(
"stored embedding target is invalid".to_owned(),
));
}
})
}
const fn default_limit() -> u32 {
10
}
const fn bounded_limit(limit: u32) -> u32 {
if limit == 0 {
10
} else if limit > 100 {
100
} else {
limit
}
}
#[cfg(test)]
mod tests {
use super::*;
fn remember(tenant: &str, person: &str, value: &str) -> RememberInput {
RememberInput {
tenant_id: TenantId(tenant.into()),
person_id: PersonId(person.into()),
ingestion_key: None,
kind: SourceKind::Conversation,
text: format!("Sam works at {value}"),
captured_at: 10,
claim: Some(ClaimInput {
subject: "Sam".into(),
predicate: "employer".into(),
value: value.into(),
valid_from: 10,
}),
}
}
fn remember_raw(tenant: &str, person: &str, text: &str) -> RememberInput {
RememberInput {
tenant_id: TenantId(tenant.into()),
person_id: PersonId(person.into()),
ingestion_key: None,
kind: SourceKind::Conversation,
text: text.into(),
captured_at: 10,
claim: None,
}
}
fn hash_for(db: &MemoryDb, target: EmbeddingTarget) -> String {
db.projection_input(&TenantId("a".into()), &PersonId("sam".into()), target)
.unwrap()
.input_hash
}
#[test]
fn ingestion_keys_are_idempotent_within_scope() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let mut first = remember_raw("a", "sam", "Remember once");
first.ingestion_key = Some("turn-1".into());
let stored = db.remember(first).unwrap();
let mut replay = remember_raw("a", "sam", "Remember once");
replay.ingestion_key = Some("turn-1".into());
let replayed = db.remember(replay).unwrap();
assert_eq!(stored.source_id, replayed.source_id);
assert_eq!(stored.evidence_id, replayed.evidence_id);
assert_eq!(
db.connection
.query_row("SELECT count(*) FROM sources", [], |row| row
.get::<_, u64>(0))
.unwrap(),
1
);
let mut other_scope = remember_raw("b", "sam", "Remember once");
other_scope.ingestion_key = Some("turn-1".into());
assert_ne!(
stored.source_id,
db.remember(other_scope).unwrap().source_id
);
}
#[test]
fn ingestion_key_rejects_changed_payload() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let mut first = remember_raw("a", "sam", "Remember once");
first.ingestion_key = Some("turn-1".into());
db.remember(first).unwrap();
let changed_text = remember_raw("a", "sam", "Changed replay content");
let mut changed_kind = remember_raw("a", "sam", "Remember once");
changed_kind.kind = SourceKind::Screen;
let mut changed_time = remember_raw("a", "sam", "Remember once");
changed_time.captured_at = 11;
let mut changed_claim = remember_raw("a", "sam", "Remember once");
changed_claim.claim = Some(ClaimInput {
subject: "Sam".into(),
predicate: "status".into(),
value: "focused".into(),
valid_from: 10,
});
for mut changed in [changed_text, changed_kind, changed_time, changed_claim] {
changed.ingestion_key = Some("turn-1".into());
assert!(matches!(db.remember(changed), Err(Error::Invalid(_))));
}
let with_claim = |value: &str| {
let mut input = remember_raw("a", "sam", "Claim capture");
input.ingestion_key = Some("claim-turn".into());
input.claim = Some(ClaimInput {
subject: "Sam".into(),
predicate: "status".into(),
value: value.into(),
valid_from: 10,
});
input
};
let stored = db.remember(with_claim("focused")).unwrap();
assert_eq!(
stored.claim_id,
db.remember(with_claim("focused")).unwrap().claim_id
);
assert!(matches!(
db.remember(with_claim("distracted")),
Err(Error::Invalid(_))
));
}
#[test]
fn ingestion_key_rejects_deleted_memory() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let mut first = remember_raw("a", "sam", "Remember once");
first.ingestion_key = Some("turn-1".into());
let stored = db.remember(first).unwrap();
db.delete_source(DeleteInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
source_id: stored.source_id,
deleted_at: 20,
})
.unwrap();
let mut replay = remember_raw("a", "sam", "Remember once");
replay.ingestion_key = Some("turn-1".into());
assert!(matches!(db.remember(replay), Err(Error::NotFound)));
}
#[test]
fn raw_sources_are_scoped_cited_and_deleted_from_retrieval() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let raw = db
.remember(remember_raw("a", "sam", "The launch code is marigold"))
.unwrap();
db.remember(remember_raw("b", "sam", "Marigold belongs elsewhere"))
.unwrap();
let found = db
.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "marigold".into(),
limit: 5,
query_embedding: None,
})
.unwrap();
assert_eq!(found.items.len(), 1);
assert_eq!(
found.items[0].memory,
MemoryRef::Source(raw.source_id.clone())
);
assert_eq!(found.items[0].evidence_ids, vec![raw.evidence_id]);
db.delete_source(DeleteInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
source_id: raw.source_id.clone(),
deleted_at: 20,
})
.unwrap();
assert!(matches!(
db.get(GetInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target: EmbeddingTarget::Source(raw.source_id),
}),
Err(Error::NotFound)
));
assert!(
db.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "marigold".into(),
limit: 5,
query_embedding: None,
})
.unwrap()
.items
.is_empty()
);
}
#[test]
fn accepted_claim_replaces_its_source_in_retrieval() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let remembered = db.remember(remember("a", "sam", "Acme")).unwrap();
let target = EmbeddingTarget::Source(remembered.source_id);
let input_hash = hash_for(&db, target.clone());
db.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target,
embedding: Embedding {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
input_hash,
normalization: VectorNormalization::L2,
distance: VectorDistance::Cosine,
},
})
.unwrap();
let found = db
.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "Acme".into(),
limit: 5,
query_embedding: Some(DenseQuery {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
}),
})
.unwrap();
assert_eq!(found.items.len(), 1);
assert_eq!(
found.items[0].memory,
MemoryRef::Claim(remembered.claim_id.unwrap())
);
}
#[test]
fn dense_evidence_without_a_claim_is_retrievable() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let raw = db
.remember(remember_raw("a", "sam", "Quiet desk near a window"))
.unwrap();
let target = EmbeddingTarget::Evidence(raw.evidence_id.clone());
let input_hash = hash_for(&db, target.clone());
db.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target,
embedding: Embedding {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
input_hash,
normalization: VectorNormalization::L2,
distance: VectorDistance::Cosine,
},
})
.unwrap();
let found = db
.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "unmatched lexical phrase".into(),
limit: 5,
query_embedding: Some(DenseQuery {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
}),
})
.unwrap();
assert_eq!(found.items.len(), 1);
assert_eq!(found.items[0].memory, MemoryRef::Evidence(raw.evidence_id));
db.delete_source(DeleteInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
source_id: raw.source_id,
deleted_at: 20,
})
.unwrap();
assert!(
db.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "unmatched lexical phrase".into(),
limit: 5,
query_embedding: Some(DenseQuery {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
}),
})
.unwrap()
.items
.is_empty()
);
}
#[test]
fn lifecycle_is_scoped_cited_and_propagates_deletion() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let first = db.remember(remember("a", "sam", "Acme")).unwrap();
db.remember(remember("b", "sam", "Other")).unwrap();
let found = db
.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "Acme".into(),
limit: 5,
query_embedding: None,
})
.unwrap();
assert_eq!(found.items.len(), 1);
assert_eq!(found.items[0].evidence_ids, vec![first.evidence_id.clone()]);
let corrected = db
.correct(CorrectInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
claim_id: first.claim_id.unwrap(),
text: "I moved to Beta".into(),
value: "Beta".into(),
occurred_at: 20,
})
.unwrap();
let old_intervals = db
.connection
.query_row(
"SELECT valid_until, recorded_until FROM claims WHERE id = ?1",
[&corrected.superseded_claim_id.0],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
)
.unwrap();
assert_eq!(old_intervals, (20, 20));
let deleted = db
.delete_source(DeleteInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
source_id: corrected.source_id,
deleted_at: 30,
})
.unwrap();
assert_eq!((deleted.evidence_count, deleted.claim_count), (1, 1));
assert!(
db.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "Beta".into(),
limit: 5,
query_embedding: None,
})
.unwrap()
.items
.is_empty()
);
}
#[test]
fn reviews_require_live_same_scope_evidence() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let remembered = db.remember(remember("a", "sam", "Acme")).unwrap();
db.store_review(ReviewInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
day: "2026-07-21".into(),
summary: "Worked at Acme".into(),
evidence_ids: vec![remembered.evidence_id.clone()],
recorded_at: 20,
})
.unwrap();
assert_eq!(
db.reviews(ReviewsInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
limit: 1
})
.unwrap()
.len(),
1
);
db.delete_source(DeleteInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
source_id: remembered.source_id,
deleted_at: 30,
})
.unwrap();
assert!(
db.reviews(ReviewsInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
limit: 1
})
.unwrap()
.is_empty()
);
}
#[test]
fn embedding_projection_is_validated_and_scoped() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let remembered = db.remember(remember("a", "sam", "Acme")).unwrap();
let target = EmbeddingTarget::Evidence(remembered.evidence_id.clone());
let embedding = Embedding {
vector: vec![0.1, 0.2],
model: "provider/model".into(),
version: "1".into(),
input_hash: hash_for(&db, target.clone()),
normalization: VectorNormalization::L2,
distance: VectorDistance::Cosine,
};
let stored = db
.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target,
embedding: embedding.clone(),
})
.unwrap();
assert_eq!(stored.dimension, 2);
assert!(matches!(
db.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("b".into()),
person_id: PersonId("sam".into()),
target: EmbeddingTarget::Evidence(remembered.evidence_id),
embedding,
}),
Err(Error::NotFound)
));
}
#[test]
fn search_fuses_lexical_and_real_dense_ranks_deterministically() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let lexical = db.remember(remember("a", "sam", "Jazz Club")).unwrap();
let dense = db.remember(remember("a", "sam", "Music Venue")).unwrap();
for (claim_id, vector) in [
(lexical.claim_id.clone().unwrap(), vec![0.9, 0.1]),
(dense.claim_id.unwrap(), vec![1.0, 0.0]),
] {
let target = EmbeddingTarget::Claim(claim_id);
let input_hash = hash_for(&db, target.clone());
db.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target,
embedding: Embedding {
vector,
model: "test/model".into(),
version: "1".into(),
input_hash,
normalization: VectorNormalization::L2,
distance: VectorDistance::Cosine,
},
})
.unwrap();
}
let found = db
.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "Jazz Club".into(),
limit: 2,
query_embedding: Some(DenseQuery {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
}),
})
.unwrap();
assert_eq!(
found.items[0].memory,
MemoryRef::Claim(lexical.claim_id.unwrap())
);
assert_eq!(found.items.len(), 2);
assert!(!found.items[0].evidence_ids.is_empty());
}
#[test]
fn stale_projections_are_excluded_and_reported_with_current_inputs() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let raw = db
.remember(remember_raw("a", "sam", "A quiet desk"))
.unwrap();
let claimed = db.remember(remember("a", "sam", "Acme")).unwrap();
let other = db.remember(remember("b", "sam", "Other")).unwrap();
let targets = [
EmbeddingTarget::Source(raw.source_id.clone()),
EmbeddingTarget::Evidence(raw.evidence_id.clone()),
EmbeddingTarget::Claim(claimed.claim_id.clone().unwrap()),
];
for target in &targets {
db.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target: target.clone(),
embedding: Embedding {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
input_hash: hash_for(&db, target.clone()),
normalization: VectorNormalization::L2,
distance: VectorDistance::Cosine,
},
})
.unwrap();
}
db.connection
.execute(
"UPDATE sources SET content = 'A changed desk', revision = revision + 1 WHERE id = ?1",
[&raw.source_id.0],
)
.unwrap();
db.connection
.execute(
"UPDATE evidence SET quote = 'Changed evidence' WHERE id = ?1",
[&raw.evidence_id.0],
)
.unwrap();
db.connection
.execute(
"UPDATE claims SET value = 'Changed employer' WHERE id = ?1",
[&claimed.claim_id.as_ref().unwrap().0],
)
.unwrap();
let found = db
.search(SearchInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
query: "no lexical match".into(),
limit: 10,
query_embedding: Some(DenseQuery {
vector: vec![1.0, 0.0],
model: "test/model".into(),
version: "1".into(),
}),
})
.unwrap();
assert!(found.items.is_empty());
let issues = db
.projection_issues(ProjectionAuditInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
model: "test/model".into(),
version: "1".into(),
limit: 100,
})
.unwrap();
let stale = issues
.iter()
.filter(|issue| issue.state == ProjectionState::Stale)
.map(|issue| &issue.input.target)
.collect::<HashSet<_>>();
assert_eq!(stale, targets.iter().collect::<HashSet<_>>());
assert!(issues.iter().all(|issue| {
issue.input.target != EmbeddingTarget::Source(other.source_id.clone())
&& issue.input.target != EmbeddingTarget::Evidence(other.evidence_id.clone())
&& issue.input.target != EmbeddingTarget::Claim(other.claim_id.clone().unwrap())
}));
assert_eq!(
db.projection_issues(ProjectionAuditInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
model: "test/model".into(),
version: "1".into(),
limit: 2,
})
.unwrap()
.len(),
2
);
}
#[test]
fn embedding_rejects_input_hash_for_different_text() {
let mut db = MemoryDb {
connection: Connection::open_in_memory().unwrap(),
};
db.migrate().unwrap();
let remembered = db
.remember(remember_raw("a", "sam", "Current text"))
.unwrap();
let result = db.upsert_embedding(EmbeddingInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
target: EmbeddingTarget::Source(remembered.source_id),
embedding: Embedding {
vector: vec![1.0],
model: "test/model".into(),
version: "1".into(),
input_hash: input_hash("different text"),
normalization: VectorNormalization::L2,
distance: VectorDistance::Cosine,
},
});
assert!(matches!(result, Err(Error::Invalid(_))));
}
#[test]
fn migration_marks_existing_projections_for_lifecycle_revalidation() {
let connection = Connection::open_in_memory().unwrap();
connection
.execute_batch(
"CREATE TABLE embeddings(tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, target_kind TEXT NOT NULL, target_id TEXT NOT NULL, model TEXT NOT NULL, version TEXT NOT NULL, dimension INTEGER NOT NULL, input_hash TEXT NOT NULL, normalization TEXT NOT NULL, distance TEXT NOT NULL, vector TEXT NOT NULL, PRIMARY KEY(tenant_id, person_id, target_kind, target_id, model, version));
INSERT INTO embeddings VALUES('a', 'sam', 'source', 'old', 'model', '1', 1, 'sha256:old', '\"l2\"', '\"cosine\"', '[1.0]');",
)
.unwrap();
let mut db = MemoryDb { connection };
db.migrate().unwrap();
let lifecycle = db
.connection
.query_row(
"SELECT target_revision, created_at FROM embeddings WHERE target_id = 'old'",
[],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
)
.unwrap();
assert_eq!(lifecycle, (0, 0));
}
#[test]
fn migration_upgrades_v1_and_accepts_point_locators() {
let connection = Connection::open_in_memory().unwrap();
connection
.execute_batch("PRAGMA user_version = 1;")
.unwrap();
let mut db = MemoryDb { connection };
db.migrate().unwrap();
let version = db
.connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.unwrap();
assert_eq!(version, 3);
let remembered = db
.remember_with_locator(
remember_raw("a", "sam", "Untimed final transcript"),
Some(TranscriptLocator {
device_id: "omi-1".into(),
provider: "deepgram".into(),
stream_id: "stream-1".into(),
segment_id: "segment-1".into(),
start_ms: 1000,
end_ms: 1000,
}),
)
.unwrap();
assert_eq!(
db.evidence_locator(EvidenceLocatorInput {
tenant_id: TenantId("a".into()),
person_id: PersonId("sam".into()),
evidence_id: remembered.evidence_id,
})
.unwrap()
.unwrap()
.start_ms,
1000
);
}
#[test]
fn migration_closes_legacy_superseded_claim_validity() {
let connection = Connection::open_in_memory().unwrap();
connection
.execute_batch(
"CREATE TABLE claims(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, person_id TEXT NOT NULL, subject TEXT NOT NULL, predicate TEXT NOT NULL, value TEXT NOT NULL, valid_from INTEGER NOT NULL, valid_until INTEGER, recorded_from INTEGER NOT NULL, recorded_until INTEGER, status TEXT NOT NULL);
INSERT INTO claims VALUES('old', 'a', 'sam', 'sam', 'employer', 'Acme', 10, NULL, 10, 20, 'superseded');",
)
.unwrap();
let mut db = MemoryDb { connection };
db.migrate().unwrap();
let valid_until = db
.connection
.query_row(
"SELECT valid_until FROM claims WHERE id = 'old'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap();
assert_eq!(valid_until, 20);
}
}