use kimetsu_core::KimetsuResult;
use kimetsu_core::ids::new_id;
use kimetsu_core::memory::{MemoryScope, normalize_memory_text};
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use crate::embeddings::{Embedder, cosine_similarity, decode_embedding};
pub const DEFAULT_CONFLICT_THRESHOLD: f32 = 0.8;
pub const DEFAULT_TOP_K: u32 = 3;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConflictHit {
pub existing_memory_id: String,
pub existing_kind: String,
pub existing_text: String,
pub similarity: f32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConflictReport {
pub conflict_id: String,
pub new_memory_id: String,
pub new_text: String,
pub existing_memory_id: String,
pub existing_text: String,
pub scope: String,
pub kind: String,
pub similarity: f32,
pub detected_at: String,
pub resolved_at: Option<String>,
pub resolution: Option<String>,
}
pub fn find_potential_conflicts(
conn: &Connection,
scope: &MemoryScope,
new_text: &str,
embedder: &dyn Embedder,
top_k: u32,
threshold: f32,
) -> KimetsuResult<Vec<ConflictHit>> {
if embedder.is_noop() {
return Ok(Vec::new());
}
let new_vec = embedder
.embed(new_text)
.map_err(|e| format!("embedder failed during conflict scan: {e}"))?;
if new_vec.len() != embedder.dim() {
return Err(format!(
"embedder {} returned {} dims, expected {}",
embedder.model_id(),
new_vec.len(),
embedder.dim()
)
.into());
}
let new_normalized = normalize_memory_text(new_text);
let scope_label = scope.to_string();
let active_model = embedder.model_id();
let mut stmt = conn.prepare(
"
SELECT memory_id, kind, text, normalized_text, embedding, embedding_model
FROM memories
WHERE scope = ?1
AND invalidated_at IS NULL
AND embedding IS NOT NULL
AND embedding_model = ?2
",
)?;
let rows = stmt.query_map(params![scope_label, active_model], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Vec<u8>>(4)?,
))
})?;
let mut hits: Vec<ConflictHit> = Vec::new();
for row in rows {
let (existing_id, kind, text, normalized, bytes) = row?;
if normalized == new_normalized {
continue;
}
let Ok(existing_vec) = decode_embedding(&bytes, Some(new_vec.len())) else {
continue;
};
let sim = cosine_similarity(&new_vec, &existing_vec);
if sim >= threshold {
hits.push(ConflictHit {
existing_memory_id: existing_id,
existing_kind: kind,
existing_text: text,
similarity: sim,
});
}
}
hits.sort_by(|a, b| {
b.similarity
.partial_cmp(&a.similarity)
.unwrap_or(std::cmp::Ordering::Equal)
});
hits.truncate(top_k as usize);
Ok(hits)
}
pub fn record_conflict(
conn: &Connection,
new_memory_id: &str,
scope: &MemoryScope,
kind: &str,
hit: &ConflictHit,
) -> KimetsuResult<String> {
let existing: Option<String> = conn
.query_row(
"
SELECT conflict_id
FROM memory_conflicts
WHERE new_memory_id = ?1 AND existing_memory_id = ?2
",
params![new_memory_id, hit.existing_memory_id],
|row| row.get::<_, String>(0),
)
.optional()?;
if let Some(id) = existing {
return Ok(id);
}
let conflict_id = new_id().to_string();
let detected_at = OffsetDateTime::now_utc()
.format(&time::format_description::well_known::Rfc3339)
.map_err(|e| format!("timestamp format: {e}"))?;
conn.execute(
"
INSERT INTO memory_conflicts (
conflict_id, new_memory_id, existing_memory_id,
scope, kind, similarity, detected_at
)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
",
params![
conflict_id,
new_memory_id,
hit.existing_memory_id,
scope.to_string(),
kind,
hit.similarity as f64,
detected_at,
],
)?;
Ok(conflict_id)
}
pub fn detect_and_record(
conn: &Connection,
new_memory_id: &str,
scope: &MemoryScope,
kind: &str,
text: &str,
embedder: &dyn Embedder,
) -> usize {
let hits = match find_potential_conflicts(
conn,
scope,
text,
embedder,
DEFAULT_TOP_K,
DEFAULT_CONFLICT_THRESHOLD,
) {
Ok(h) => h,
Err(e) => {
eprintln!("kimetsu-brain: conflict scan skipped: {e}");
return 0;
}
};
let mut recorded = 0usize;
for hit in &hits {
match record_conflict(conn, new_memory_id, scope, kind, hit) {
Ok(_) => recorded += 1,
Err(e) => {
eprintln!(
"kimetsu-brain: failed to record conflict {} <-> {}: {e}",
new_memory_id, hit.existing_memory_id
);
}
}
}
recorded
}
pub fn list_unresolved_conflicts(
conn: &Connection,
limit: u32,
) -> KimetsuResult<Vec<ConflictReport>> {
let mut stmt = conn.prepare(
"
SELECT c.conflict_id, c.new_memory_id, mn.text, c.existing_memory_id,
me.text, c.scope, c.kind, c.similarity, c.detected_at,
c.resolved_at, c.resolution
FROM memory_conflicts c
LEFT JOIN memories mn ON mn.memory_id = c.new_memory_id
LEFT JOIN memories me ON me.memory_id = c.existing_memory_id
WHERE c.resolved_at IS NULL
ORDER BY c.detected_at DESC
LIMIT ?1
",
)?;
let rows = stmt.query_map(params![limit], |row| {
Ok(ConflictReport {
conflict_id: row.get(0)?,
new_memory_id: row.get(1)?,
new_text: row.get::<_, Option<String>>(2)?.unwrap_or_default(),
existing_memory_id: row.get(3)?,
existing_text: row.get::<_, Option<String>>(4)?.unwrap_or_default(),
scope: row.get(5)?,
kind: row.get(6)?,
similarity: row.get::<_, f64>(7)? as f32,
detected_at: row.get(8)?,
resolved_at: row.get(9)?,
resolution: row.get(10)?,
})
})?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
}
pub fn resolve_conflict(
conn: &Connection,
conflict_id: &str,
resolution: &str,
) -> KimetsuResult<bool> {
let resolution = resolution.trim();
if !matches!(resolution, "kept_new" | "kept_existing" | "kept_both") {
return Err(format!(
"invalid conflict resolution {resolution:?}; expected kept_new | kept_existing | kept_both"
)
.into());
}
let pair: Option<(String, String)> = conn
.query_row(
"
SELECT new_memory_id, existing_memory_id
FROM memory_conflicts
WHERE conflict_id = ?1 AND resolved_at IS NULL
",
params![conflict_id],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
)
.optional()?;
let Some((new_memory_id, existing_memory_id)) = pair else {
return Ok(false);
};
let now = OffsetDateTime::now_utc()
.format(&time::format_description::well_known::Rfc3339)
.map_err(|e| format!("timestamp format: {e}"))?;
let invalidation_reason = format!("v0.5.2 conflict {conflict_id} resolved as {resolution}");
if resolution == "kept_new" {
conn.execute(
"
UPDATE memories
SET invalidated_at = COALESCE(invalidated_at, ?2),
invalidated_reason = COALESCE(invalidated_reason, ?3)
WHERE memory_id = ?1
",
params![existing_memory_id, now, invalidation_reason],
)?;
} else if resolution == "kept_existing" {
conn.execute(
"
UPDATE memories
SET invalidated_at = COALESCE(invalidated_at, ?2),
invalidated_reason = COALESCE(invalidated_reason, ?3)
WHERE memory_id = ?1
",
params![new_memory_id, now, invalidation_reason],
)?;
}
let updated = conn.execute(
"
UPDATE memory_conflicts
SET resolved_at = ?2, resolution = ?3
WHERE conflict_id = ?1 AND resolved_at IS NULL
",
params![conflict_id, now, resolution],
)?;
Ok(updated > 0)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::embeddings::{NoopEmbedder, StubEmbedder, encode_embedding};
use kimetsu_core::memory::normalize_memory_text;
use rusqlite::Connection;
fn open_test_brain() -> Connection {
let conn = Connection::open_in_memory().expect("open in-memory");
crate::schema::initialize(&conn).expect("init schema");
conn
}
fn insert_memory(
conn: &Connection,
memory_id: &str,
scope: &str,
kind: &str,
text: &str,
embedder: &dyn Embedder,
) {
let normalized = normalize_memory_text(text);
let vec = embedder.embed(text).expect("embed test row");
let blob = encode_embedding(&vec);
conn.execute(
"
INSERT INTO memories (
memory_id, scope, kind, text, normalized_text, confidence,
source_event_id, provenance_snapshot_json, created_at,
use_count, usefulness_score, embedding, embedding_model
)
VALUES (?1, ?2, ?3, ?4, ?5, 1.0, NULL, '{}',
'2026-01-01T00:00:00Z', 0, 0.0, ?6, ?7)
",
params![
memory_id,
scope,
kind,
text,
normalized,
blob,
embedder.model_id(),
],
)
.expect("insert");
conn.execute(
"INSERT INTO memories_fts (memory_id, text, kind, scope)
VALUES (?1, ?2, ?3, ?4)",
params![memory_id, text, kind, scope],
)
.expect("fts");
}
#[test]
fn noop_embedder_returns_no_conflicts() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(
&conn,
"m_existing",
"global_user",
"fact",
"use thiserror for libraries",
&stub,
);
let hits = find_potential_conflicts(
&conn,
&MemoryScope::GlobalUser,
"use anyhow for libraries",
&NoopEmbedder,
DEFAULT_TOP_K,
DEFAULT_CONFLICT_THRESHOLD,
)
.expect("scan");
assert!(hits.is_empty(), "noop embedder should produce no hits");
}
#[test]
fn cross_model_rows_are_skipped() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(
&conn,
"m_xmodel",
"global_user",
"fact",
"use thiserror",
&stub,
);
conn.execute(
"UPDATE memories SET embedding_model = 'bge-small-en-v1.5' WHERE memory_id = 'm_xmodel'",
[],
)
.expect("force mismatch");
let hits = find_potential_conflicts(
&conn,
&MemoryScope::GlobalUser,
"use thiserror everywhere", &stub,
DEFAULT_TOP_K,
0.0,
)
.expect("scan");
assert!(
hits.is_empty(),
"cross-model rows must be skipped from conflict scan"
);
}
#[test]
fn exact_match_is_not_flagged_as_conflict() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(
&conn,
"m_exact",
"global_user",
"fact",
"Use ripgrep",
&stub,
);
let hits = find_potential_conflicts(
&conn,
&MemoryScope::GlobalUser,
"use ripgrep",
&stub,
DEFAULT_TOP_K,
0.0, )
.expect("scan");
assert!(
hits.is_empty(),
"exact normalized-text match should be dedup, not conflict"
);
}
#[test]
fn similar_but_different_text_is_flagged() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(
&conn,
"m_existing",
"global_user",
"fact",
"alpha beta gamma delta",
&stub,
);
let hits = find_potential_conflicts(
&conn,
&MemoryScope::GlobalUser,
"alpha beta gamma omega", &stub,
DEFAULT_TOP_K,
0.4,
)
.expect("scan");
assert!(
!hits.is_empty(),
"high-cosine + different-normalized text should flag a conflict"
);
assert_eq!(hits[0].existing_memory_id, "m_existing");
assert!(
hits[0].similarity >= 0.4,
"similarity should be >= threshold; got {}",
hits[0].similarity
);
}
#[test]
fn record_conflict_is_idempotent() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(&conn, "m_new", "global_user", "fact", "alpha", &stub);
insert_memory(&conn, "m_old", "global_user", "fact", "beta", &stub);
let hit = ConflictHit {
existing_memory_id: "m_old".to_string(),
existing_kind: "fact".to_string(),
existing_text: "beta".to_string(),
similarity: 0.85,
};
let id1 = record_conflict(&conn, "m_new", &MemoryScope::GlobalUser, "fact", &hit)
.expect("record 1");
let id2 = record_conflict(&conn, "m_new", &MemoryScope::GlobalUser, "fact", &hit)
.expect("record 2");
assert_eq!(id1, id2, "re-recording the same pair must return same id");
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM memory_conflicts", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(count, 1);
}
#[test]
fn list_unresolved_excludes_resolved_rows() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(
&conn,
"m_new1",
"global_user",
"fact",
"use thiserror",
&stub,
);
insert_memory(&conn, "m_old1", "global_user", "fact", "use anyhow", &stub);
insert_memory(
&conn,
"m_new2",
"global_user",
"fact",
"tabs over spaces",
&stub,
);
insert_memory(
&conn,
"m_old2",
"global_user",
"fact",
"spaces over tabs",
&stub,
);
let hit1 = ConflictHit {
existing_memory_id: "m_old1".to_string(),
existing_kind: "fact".to_string(),
existing_text: "use anyhow".to_string(),
similarity: 0.9,
};
let hit2 = ConflictHit {
existing_memory_id: "m_old2".to_string(),
existing_kind: "fact".to_string(),
existing_text: "spaces over tabs".to_string(),
similarity: 0.85,
};
let cid1 =
record_conflict(&conn, "m_new1", &MemoryScope::GlobalUser, "fact", &hit1).unwrap();
let _cid2 =
record_conflict(&conn, "m_new2", &MemoryScope::GlobalUser, "fact", &hit2).unwrap();
assert!(resolve_conflict(&conn, &cid1, "kept_both").unwrap());
let open = list_unresolved_conflicts(&conn, 50).unwrap();
assert_eq!(open.len(), 1, "only the unresolved conflict should list");
assert_eq!(open[0].new_memory_id, "m_new2");
assert_eq!(open[0].existing_memory_id, "m_old2");
assert_eq!(open[0].new_text, "tabs over spaces");
assert_eq!(open[0].existing_text, "spaces over tabs");
}
#[test]
fn resolve_conflict_invalidates_loser_side() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
for (mid, text) in [
("m_keep_new", "alpha"),
("m_old_loses", "beta"),
("m_new_loses", "gamma"),
("m_keep_existing", "delta"),
("m_both_a", "epsilon"),
("m_both_b", "zeta"),
] {
insert_memory(&conn, mid, "global_user", "fact", text, &stub);
}
let mk_hit = |old: &str| ConflictHit {
existing_memory_id: old.to_string(),
existing_kind: "fact".to_string(),
existing_text: "x".to_string(),
similarity: 0.9,
};
let c_kept_new = record_conflict(
&conn,
"m_keep_new",
&MemoryScope::GlobalUser,
"fact",
&mk_hit("m_old_loses"),
)
.unwrap();
let c_kept_existing = record_conflict(
&conn,
"m_new_loses",
&MemoryScope::GlobalUser,
"fact",
&mk_hit("m_keep_existing"),
)
.unwrap();
let c_both = record_conflict(
&conn,
"m_both_a",
&MemoryScope::GlobalUser,
"fact",
&mk_hit("m_both_b"),
)
.unwrap();
assert!(resolve_conflict(&conn, &c_kept_new, "kept_new").unwrap());
assert!(resolve_conflict(&conn, &c_kept_existing, "kept_existing").unwrap());
assert!(resolve_conflict(&conn, &c_both, "kept_both").unwrap());
let invalidated_at: Vec<(String, Option<String>)> = {
let mut stmt = conn
.prepare("SELECT memory_id, invalidated_at FROM memories ORDER BY memory_id")
.unwrap();
stmt.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?))
})
.unwrap()
.map(|r| r.unwrap())
.collect()
};
let map: std::collections::HashMap<_, _> = invalidated_at.into_iter().collect();
assert!(map["m_keep_new"].is_none(), "winner should stay active");
assert!(
map["m_old_loses"].is_some(),
"kept_new must invalidate the existing memory"
);
assert!(
map["m_keep_existing"].is_none(),
"winner (existing) should stay active"
);
assert!(
map["m_new_loses"].is_some(),
"kept_existing must invalidate the new memory"
);
assert!(
map["m_both_a"].is_none() && map["m_both_b"].is_none(),
"kept_both should leave both memories active"
);
}
#[test]
fn resolve_conflict_is_idempotent() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(&conn, "m_new", "global_user", "fact", "x", &stub);
insert_memory(&conn, "m_old", "global_user", "fact", "y", &stub);
let hit = ConflictHit {
existing_memory_id: "m_old".to_string(),
existing_kind: "fact".to_string(),
existing_text: "y".to_string(),
similarity: 0.95,
};
let cid = record_conflict(&conn, "m_new", &MemoryScope::GlobalUser, "fact", &hit).unwrap();
assert!(resolve_conflict(&conn, &cid, "kept_new").unwrap());
assert!(
!resolve_conflict(&conn, &cid, "kept_existing").unwrap(),
"second resolve must return false (already resolved)"
);
}
#[test]
fn detect_and_record_noop_writes_nothing() {
let conn = open_test_brain();
let stub = StubEmbedder::new();
insert_memory(
&conn,
"m_existing",
"global_user",
"fact",
"alpha beta",
&stub,
);
insert_memory(&conn, "m_new", "global_user", "fact", "alpha gamma", &stub);
let recorded = detect_and_record(
&conn,
"m_new",
&MemoryScope::GlobalUser,
"fact",
"alpha gamma",
&NoopEmbedder,
);
assert_eq!(recorded, 0);
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM memory_conflicts", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(count, 0);
}
#[test]
fn resolve_conflict_rejects_invalid_resolution_strings() {
let conn = open_test_brain();
let err = resolve_conflict(&conn, "ignored", "delete_them_all").unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("invalid conflict resolution"), "got: {msg}");
}
}