use rusqlite::{params, Connection};
use crate::graph_index::GraphIndex;
const MAX_ANCHOR_ENTITIES: usize = 4;
const MAX_CLAIMS_PER_ENTITY: usize = 24;
const MAX_PATH_SEEDS: usize = 8;
const MAX_PATH_PER_SEED: usize = 4;
const MAX_PATH_CANDIDATES: usize = 16;
const DIRECT_CLAIM_LEX: f64 = 1.0;
const PATH_CLAIM_LEX: f64 = 0.9;
const PATH_MARKER: &str = "(path via ";
const CHAIN_DENY_RELS: &[&str] = &["co_occurs_with", "related_to", "mentions"];
pub(crate) fn chain_traversable(rel_type: &str) -> bool {
!CHAIN_DENY_RELS.contains(&rel_type)
}
pub(crate) struct ClaimCandidate {
pub rid: String,
pub why: String,
pub hops: u8,
}
pub(crate) fn claims_lex_strength(why: &[String]) -> Option<f64> {
let w = why.iter().find(|w| w.starts_with("claims_match"))?;
Some(if w.contains(PATH_MARKER) {
PATH_CLAIM_LEX
} else {
DIRECT_CLAIM_LEX
})
}
fn claims_touching(
conn: &Connection,
entity: &str,
namespace: Option<&str>,
) -> Vec<(String, String, String, String, i64)> {
let sql = format!(
"SELECT src, rel_type, dst, source_memory_rid, polarity, extractor FROM claims \
WHERE (src = ?1 OR dst = ?1) AND tombstoned = 0 \
AND source_memory_rid IS NOT NULL {} \
ORDER BY created_at DESC LIMIT {}",
if namespace.is_some() {
"AND namespace = ?2"
} else {
""
},
MAX_CLAIMS_PER_ENTITY,
);
let Ok(mut stmt) = conn.prepare_cached(&sql) else {
return Vec::new(); };
let mapper =
|row: &rusqlite::Row| -> rusqlite::Result<(String, String, String, String, i64, String)> {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
))
};
let rows: Vec<(String, String, String, String, i64, String)> = if let Some(ns) = namespace {
stmt.query_map(params![entity, ns], mapper)
.map(|r| r.filter_map(|x| x.ok()).collect())
.unwrap_or_default()
} else {
stmt.query_map(params![entity], mapper)
.map(|r| r.filter_map(|x| x.ok()).collect())
.unwrap_or_default()
};
rows.into_iter()
.filter(|(src, _, dst, _, _, _extractor)| {
!(crate::graph::is_rejected_entity_name(src)
|| crate::graph::is_rejected_entity_name(dst))
})
.map(|(src, rel, dst, rid, polarity, _)| (src, rel, dst, rid, polarity))
.collect()
}
pub(crate) fn claims_candidates(
conn: &Connection,
graph_index: &GraphIndex,
query_tokens: &[String],
namespace: Option<&str>,
) -> Vec<ClaimCandidate> {
let mut anchors = graph_index.entity_matches_query(query_tokens);
if anchors.is_empty() {
return Vec::new();
}
anchors.sort_by(|a, b| b.2.cmp(&a.2).then_with(|| a.0.cmp(&b.0)));
anchors.truncate(MAX_ANCHOR_ENTITIES);
let anchor_names: std::collections::HashSet<&str> =
anchors.iter().map(|(name, _, _)| name.as_str()).collect();
let mut out: Vec<ClaimCandidate> = Vec::new();
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut path_seeds: Vec<(String, String, String)> = Vec::new();
for (entity, _etype, _mentions) in &anchors {
for (src, rel, dst, rid, polarity) in claims_touching(conn, entity, namespace) {
let neg = if polarity < 0 { "NOT " } else { "" };
let far = if src == *entity { &dst } else { &src };
if chain_traversable(&rel)
&& !anchor_names.contains(far.as_str())
&& path_seeds.len() < MAX_PATH_SEEDS
&& !path_seeds.iter().any(|(seed, _, _)| seed == far)
{
path_seeds.push((
far.clone(),
entity.clone(),
format!("{src} -{neg}{rel}-> {dst}"),
));
}
if !seen.insert(rid.clone()) {
continue;
}
out.push(ClaimCandidate {
why: format!("claims_match: {src} -{neg}{rel}-> {dst} (anchor {entity})"),
rid,
hops: 1,
});
}
}
let mut path_admitted = 0usize;
for (seed, anchor, hop1) in &path_seeds {
if path_admitted >= MAX_PATH_CANDIDATES {
break;
}
let rows = claims_touching(conn, seed, namespace);
if rows.len() >= MAX_CLAIMS_PER_ENTITY {
continue;
}
let mut per_seed = 0usize;
for (src, rel, dst, rid, polarity) in rows {
if per_seed >= MAX_PATH_PER_SEED || path_admitted >= MAX_PATH_CANDIDATES {
break;
}
if !chain_traversable(&rel) {
continue; }
if !seen.insert(rid.clone()) {
continue; }
let neg = if polarity < 0 { "NOT " } else { "" };
out.push(ClaimCandidate {
why: format!(
"claims_match: {hop1} ; {src} -{neg}{rel}-> {dst} \
{PATH_MARKER}{seed}, anchor {anchor})"
),
rid,
hops: 2,
});
per_seed += 1;
path_admitted += 1;
}
}
out
}
impl super::YantrikDB {
#[allow(clippy::too_many_arguments)]
#[allow(clippy::too_many_arguments)]
pub(crate) fn apply_claims_lane(
&self,
scored: &mut Vec<crate::types::RecallResult>,
query_embedding: &[f32],
query_text: Option<&str>,
namespace: Option<&str>,
time_window: Option<(f64, f64)>,
include_consolidated: bool,
memory_type: Option<&str>,
domain: Option<&str>,
source: Option<&str>,
certainty_min: Option<f64>,
event_allow: Option<&std::collections::HashSet<String>>,
learned_weights: &crate::types::LearnedWeights,
ts: f64,
query_sentiment: f64,
) -> crate::error::Result<()> {
use crate::base::scoring;
let Some(qt) = query_text else {
return Ok(());
};
let cands = {
let gi = self.graph_index.read();
let tokens = crate::graph::tokenize(qt);
let conn = self.read_conn();
claims_candidates(&conn, &gi, &tokens, namespace)
};
if cands.is_empty() {
return Ok(());
}
let mut by_rid: std::collections::HashMap<&str, &str> = cands
.iter()
.map(|c| (c.rid.as_str(), c.why.as_str()))
.collect();
for result in scored.iter_mut() {
if let Some(why) = by_rid.remove(result.rid.as_str()) {
if !result
.why_retrieved
.iter()
.any(|w| w.starts_with("claims_match"))
{
result.why_retrieved.push(why.to_string());
}
}
}
let new_rids: Vec<(&str, &str)> = cands
.iter()
.filter(|c| by_rid.contains_key(c.rid.as_str()))
.map(|c| (c.rid.as_str(), c.why.as_str()))
.collect();
if new_rids.is_empty() {
return Ok(());
}
let rid_refs: Vec<&str> = new_rids.iter().map(|(r, _)| *r).collect();
let emb_map = self.fetch_embeddings_by_rids(&rid_refs)?;
let cache = self.scoring_cache.read();
for (rid, claim_why) in new_rids {
let Some(row) = cache.get(rid) else { continue };
if !crate::engine::recall::passes_recall_filters(
rid,
row,
include_consolidated,
memory_type,
time_window,
namespace,
domain,
source,
certainty_min,
event_allow,
) {
continue;
}
let Some(emb_blob) = emb_map.get(rid) else {
continue;
};
let mem_emb = crate::serde_helpers::deserialize_f32(emb_blob);
let sim_score = crate::consolidate::cosine_similarity(query_embedding, &mem_emb) as f64;
let decay = scoring::ranking_decay(row.importance, row.created_at, ts);
let age = ts - row.created_at;
let recency = scoring::recency_score(age);
let composite = scoring::adaptive_composite_score(
sim_score,
decay,
recency,
row.importance,
row.valence,
query_sentiment,
learned_weights,
);
let mut why = scoring::build_why(sim_score, recency, decay, row.valence);
why.push(claim_why.to_string());
let contributions = scoring::adaptive_contributions(
sim_score,
decay,
recency,
row.importance,
learned_weights,
);
let valence_multiplier = scoring::query_valence_boost(row.valence, query_sentiment);
scored.push(crate::types::RecallResult {
rid: rid.to_string(),
memory_type: row.memory_type.clone(),
text: String::new(),
created_at: row.created_at,
importance: row.importance,
valence: row.valence,
score: composite,
scores: crate::types::ScoreBreakdown {
similarity: sim_score,
decay,
recency,
importance: row.importance,
graph_proximity: 0.0,
contributions,
valence_multiplier,
},
why_retrieved: why,
metadata: serde_json::Value::Null,
namespace: row.namespace.clone(),
certainty: row.certainty,
domain: row.domain.clone(),
source: row.source.clone(),
emotional_state: row.emotional_state.clone(),
current_status: Default::default(),
superseded_by: None,
disputed_with: Vec::new(),
aged_last_verified: None,
best_span: None,
pack: None,
});
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
fn seeded_store() -> Connection {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"CREATE TABLE entities (name TEXT PRIMARY KEY, entity_type TEXT, \
first_seen REAL, last_seen REAL, mention_count INTEGER);
CREATE TABLE memory_entities (memory_rid TEXT, entity_name TEXT);
CREATE TABLE claims (claim_id TEXT PRIMARY KEY, src TEXT NOT NULL, \
dst TEXT NOT NULL, rel_type TEXT NOT NULL, weight REAL DEFAULT 1.0, \
created_at REAL NOT NULL, tombstoned INTEGER NOT NULL DEFAULT 0, \
polarity INTEGER NOT NULL DEFAULT 1, \
extractor TEXT NOT NULL DEFAULT 'manual', source_memory_rid TEXT, \
namespace TEXT NOT NULL DEFAULT 'default');
CREATE VIEW edges AS SELECT src, dst, weight, tombstoned FROM claims;",
)
.unwrap();
for (name, etype, mc) in [
("DB", "tech", 5),
("Postgres", "tech", 3),
("THE", "unknown", 10),
] {
conn.execute(
"INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
VALUES (?1, ?2, 0.0, 0.0, ?3)",
params![name, etype, mc],
)
.unwrap();
}
for (cid, src, dst, rel, ts, extractor, rid) in [
("c1", "DB", "THE", "leads", 3.0, "heuristic_v1", "m1"),
("c2", "DB", "Postgres", "uses", 2.0, "heuristic_v1", "m2"),
("c3", "THE", "DB", "leads", 1.0, "manual", "m3"),
] {
conn.execute(
"INSERT INTO claims (claim_id, src, dst, rel_type, created_at, \
extractor, source_memory_rid) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![cid, src, dst, rel, ts, extractor, rid],
)
.unwrap();
}
conn
}
#[test]
fn stopword_endpoint_claims_are_suppressed_at_read() {
let conn = seeded_store();
let gi = GraphIndex::build_from_db(&conn).unwrap();
let tokens = crate::graph::tokenize("what does DB use");
let cands = claims_candidates(&conn, &gi, &tokens, None);
let rids: Vec<&str> = cands.iter().map(|c| c.rid.as_str()).collect();
assert!(
!rids.contains(&"m1"),
"heuristic claim with stopword dst must be suppressed, got {rids:?}"
);
assert!(
rids.contains(&"m2"),
"legitimate claim must still match, got {rids:?}"
);
assert!(
!rids.contains(&"m3"),
"migration-backfilled 'manual' label must NOT exempt a phantom-anchored claim in this lane, got {rids:?}"
);
}
#[test]
fn numeric_endpoint_claims_are_suppressed_at_read() {
let conn = seeded_store();
conn.execute(
"INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
VALUES ('15', 'unknown', 0.0, 0.0, 4)",
[],
)
.unwrap();
conn.execute(
"INSERT INTO claims (claim_id, src, dst, rel_type, created_at, extractor, \
source_memory_rid) VALUES ('c15', '15', 'LOG', 'leads', 0.0, 'heuristic_v1', 'm9')",
[],
)
.unwrap();
conn.execute(
"INSERT INTO memory_entities (memory_rid, entity_name) VALUES ('m9', '15')",
[],
)
.unwrap();
let gi = GraphIndex::build_from_db(&conn).unwrap();
let tokens = crate::graph::tokenize("release 15 log architecture");
let cands = claims_candidates(&conn, &gi, &tokens, None);
let rids: Vec<&str> = cands.iter().map(|c| c.rid.as_str()).collect();
assert!(
!rids.contains(&"m9"),
"bare-number-anchored heuristic claim must be suppressed, got {rids:?}"
);
}
#[test]
fn protected_stopword_anchor_surfaces_only_manual_claims() {
let conn = seeded_store();
let gi = GraphIndex::build_from_db(&conn).unwrap();
assert!(
!gi.entity_matches_query(&[String::from("the")]).is_empty(),
"fixture must reproduce the protected-anchor precondition"
);
let tokens = crate::graph::tokenize("the database leads");
let cands = claims_candidates(&conn, &gi, &tokens, None);
assert!(
!cands.iter().any(|c| c.why.contains("DB -leads-> THE")),
"the exact live phantom why must never be emitted, got {:?}",
cands.iter().map(|c| c.why.as_str()).collect::<Vec<_>>()
);
assert!(
!cands.iter().any(|c| c.rid == "m3"),
"no phantom-anchored claim survives, whatever its extractor label — the migration backfill made 'manual' meaningless for lane rows"
);
}
#[test]
fn direction_provenance_is_spelled_out() {
let c = ClaimCandidate {
rid: "r".into(),
why: format!(
"claims_match: {} -{}{}-> {} (anchor {})",
"Taylor", "", "reports_to", "Carol", "Taylor"
),
hops: 1,
};
assert_eq!(
c.why,
"claims_match: Taylor -reports_to-> Carol (anchor Taylor)"
);
}
fn chain_store() -> Connection {
let conn = seeded_store();
for (name, etype) in [
("Alice Moreau", "person"),
("Fennwick Labs", "org"),
("Berlin", "place"),
] {
conn.execute(
"INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
VALUES (?1, ?2, 0.0, 0.0, 2)",
params![name, etype],
)
.unwrap();
}
for (cid, src, dst, rel, ts, rid) in [
(
"cA",
"Alice Moreau",
"Fennwick Labs",
"works_at",
10.0,
"mA",
),
(
"cB",
"Fennwick Labs",
"Berlin",
"headquartered_in",
11.0,
"mB",
),
("cC", "Fennwick Labs", "THE", "leads", 12.0, "mC"),
] {
conn.execute(
"INSERT INTO claims (claim_id, src, dst, rel_type, created_at, \
extractor, source_memory_rid) VALUES (?1, ?2, ?3, ?4, ?5, 'heuristic_v1', ?6)",
params![cid, src, dst, rel, ts, rid],
)
.unwrap();
}
for (rid, name) in [
("mA", "Alice Moreau"),
("mA", "Fennwick Labs"),
("mB", "Fennwick Labs"),
("mB", "Berlin"),
] {
conn.execute(
"INSERT INTO memory_entities (memory_rid, entity_name) VALUES (?1, ?2)",
params![rid, name],
)
.unwrap();
}
conn
}
#[test]
fn chain_admits_the_second_hop_with_full_path_provenance() {
let conn = chain_store();
let gi = GraphIndex::build_from_db(&conn).unwrap();
let tokens = crate::graph::tokenize("which city does Alice Moreau work in");
let cands = claims_candidates(&conn, &gi, &tokens, None);
let direct = cands
.iter()
.find(|c| c.rid == "mA")
.expect("hop-1 record admitted");
assert_eq!(direct.hops, 1);
let hop2 = cands
.iter()
.find(|c| c.rid == "mB")
.expect("hop-2 record admitted");
assert_eq!(hop2.hops, 2);
assert_eq!(
hop2.why,
"claims_match: Alice Moreau -works_at-> Fennwick Labs ; \
Fennwick Labs -headquartered_in-> Berlin (path via Fennwick Labs, anchor Alice Moreau)"
);
assert!(
!cands.iter().any(|c| c.rid == "mC"),
"phantom endpoint on the seed must be suppressed at hop 2 too"
);
let hops: Vec<u8> = cands.iter().map(|c| c.hops).collect();
assert!(hops.windows(2).all(|w| w[0] <= w[1]), "got {hops:?}");
}
#[test]
fn chain_never_readmits_hop_one_provenance_from_the_seed_side() {
let conn = chain_store();
let gi = GraphIndex::build_from_db(&conn).unwrap();
let tokens = crate::graph::tokenize("Alice Moreau");
let cands = claims_candidates(&conn, &gi, &tokens, None);
assert_eq!(cands.iter().filter(|c| c.rid == "mA").count(), 1);
}
#[test]
fn hub_seed_is_not_traversed() {
let conn = chain_store();
for i in 0..MAX_CLAIMS_PER_ENTITY {
let (cid, dst, rid) = (format!("hub{i}"), format!("Partner{i}"), format!("mh{i}"));
conn.execute(
"INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
VALUES (?1, 'org', 0.0, 0.0, 1)",
params![dst],
)
.unwrap();
conn.execute(
"INSERT INTO claims (claim_id, src, dst, rel_type, created_at, \
extractor, source_memory_rid) VALUES (?1, 'Fennwick Labs', ?2, 'partners_with', 20.0, \
'heuristic_v1', ?3)",
params![cid, dst, rid],
)
.unwrap();
}
let gi = GraphIndex::build_from_db(&conn).unwrap();
let tokens = crate::graph::tokenize("which city does Alice Moreau work in");
let cands = claims_candidates(&conn, &gi, &tokens, None);
assert!(
cands.iter().all(|c| c.hops == 1),
"hub must not be traversed: {:?}",
cands.iter().map(|c| &c.why).collect::<Vec<_>>()
);
}
#[test]
fn chain_never_follows_a_denied_relation() {
let conn = chain_store();
conn.execute(
"INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
VALUES ('Lisbon', 'place', 0.0, 0.0, 1)",
[],
)
.unwrap();
conn.execute(
"INSERT INTO claims (claim_id, src, dst, rel_type, created_at, \
extractor, source_memory_rid) VALUES ('cX', 'Fennwick Labs', 'Lisbon', \
'co_occurs_with', 30.0, 'heuristic_v1', 'mX')",
[],
)
.unwrap();
let gi = GraphIndex::build_from_db(&conn).unwrap();
let tokens = crate::graph::tokenize("which city does Alice Moreau work in");
let cands = claims_candidates(&conn, &gi, &tokens, None);
assert!(
cands.iter().any(|c| c.rid == "mB"),
"real hop-2 still admitted"
);
assert!(
!cands.iter().any(|c| c.rid == "mX"),
"co_occurs_with hop must not be followed"
);
assert!(!chain_traversable("co_occurs_with") && chain_traversable("works_at"));
}
#[test]
fn path_reserve_strength_ranks_below_a_direct_claim() {
let direct = vec!["claims_match: A -works_at-> B (anchor A)".to_string()];
let path = vec![
"recent".to_string(),
"claims_match: A -works_at-> B ; B -headquartered_in-> C (path via B, anchor A)"
.to_string(),
];
assert_eq!(claims_lex_strength(&direct), Some(DIRECT_CLAIM_LEX));
assert_eq!(claims_lex_strength(&path), Some(PATH_CLAIM_LEX));
assert!(PATH_CLAIM_LEX < DIRECT_CLAIM_LEX);
assert_eq!(claims_lex_strength(&["keyword_match".to_string()]), None);
}
}