1use std::collections::{HashMap, HashSet};
17use std::sync::OnceLock;
18use std::time::{Instant, SystemTime, UNIX_EPOCH};
19#[allow(unused_imports)]
20use zeph_db::sql;
21
22use crate::embedding_store::EmbeddingStore;
23use crate::error::MemoryError;
24use crate::graph::store::GraphStore;
25use crate::graph::types::{Edge, EdgeType, edge_type_weight, evolved_weight};
26use crate::sqlite_time::parse_sqlite_datetime_to_unix;
27
28#[derive(Debug, Clone)]
30pub struct ActivatedNode {
31 pub entity_id: i64,
33 pub activation: f32,
35 pub depth: u32,
37}
38
39#[derive(Debug, Clone)]
41pub struct ActivatedFact {
42 pub edge: Edge,
44 pub activation_score: f32,
46 pub is_implicit_conflict: bool,
48 pub conflict_candidate_id: Option<i64>,
50}
51
52pub use zeph_common::memory::SpreadingActivationParams;
53
54#[derive(Debug, Clone)]
63pub struct HelaFact {
64 pub edge: Edge,
66 pub score: f32,
68 pub depth: u32,
71 pub path_weight: f32,
74 pub cosine: Option<f32>,
78}
79
80#[derive(Debug, Clone)]
90pub struct HelaSpreadParams {
91 pub spread_depth: u32,
93 pub edge_types: Vec<EdgeType>,
95 pub max_visited: usize,
97 pub step_budget: Option<std::time::Duration>,
103 pub embed_timeout: Option<std::time::Duration>,
106}
107
108impl Default for HelaSpreadParams {
109 fn default() -> Self {
110 Self {
111 spread_depth: 2,
112 edge_types: Vec::new(),
113 max_visited: 200,
114 step_budget: Some(std::time::Duration::from_millis(80)),
115 embed_timeout: Some(std::time::Duration::from_secs(5)),
116 }
117 }
118}
119
120static HELA_DIM_MISMATCH: OnceLock<String> = OnceLock::new();
129
130fn cosine(a: &[f32], b: &[f32]) -> f32 {
134 let dot: f32 = a.iter().zip(b.iter()).map(|(&x, &y)| x * y).sum();
135 let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
136 let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
137 let denom = (norm_a * norm_b).max(f32::EPSILON);
138 dot / denom
139}
140
141#[tracing::instrument(
168 name = "memory.graph.hela_spread",
169 skip_all,
170 fields(
171 depth = params.spread_depth,
172 limit,
173 anchor_id = tracing::field::Empty,
174 visited = tracing::field::Empty,
175 scored = tracing::field::Empty,
176 fallback = tracing::field::Empty,
177 )
178)]
179#[allow(clippy::too_many_arguments, clippy::too_many_lines)] pub async fn hela_spreading_recall(
181 store: &GraphStore,
182 embeddings: &EmbeddingStore,
183 provider: &zeph_llm::any::AnyProvider,
184 query: &str,
185 limit: usize,
186 params: &HelaSpreadParams,
187 hebbian_enabled: bool,
188 hebbian_lr: f32,
189) -> Result<Vec<HelaFact>, MemoryError> {
190 use zeph_llm::LlmProvider as _;
191
192 const ENTITY_COLLECTION: &str = "zeph_graph_entities";
193
194 if limit == 0 {
195 return Ok(Vec::new());
196 }
197
198 if HELA_DIM_MISMATCH.get().map(String::as_str) == Some(ENTITY_COLLECTION) {
201 tracing::debug!("hela: dim mismatch previously detected for collection, skipping");
202 return Ok(Vec::new());
203 }
204
205 let q_vec = if let Some(timeout) = params.embed_timeout {
207 tokio::time::timeout(timeout, provider.embed(query))
208 .await
209 .map_err(|_| {
210 tracing::warn!(timeout_ms = timeout.as_millis(), "hela: embed timed out");
211 MemoryError::Timeout("hela embed".into())
212 })??
213 } else {
214 provider.embed(query).await?
215 };
216
217 let t_anchor = Instant::now();
219 let anchor_results = match embeddings
220 .search_collection(ENTITY_COLLECTION, &q_vec, 1, None)
221 .await
222 {
223 Ok(r) => r,
224 Err(e) => {
225 let msg = e.to_string();
226 if msg.contains("wrong vector dimension")
227 || msg.contains("InvalidArgument")
228 || msg.contains("dimension")
229 {
230 let _ = HELA_DIM_MISMATCH.set(ENTITY_COLLECTION.to_owned());
231 tracing::warn!(
232 collection = ENTITY_COLLECTION,
233 error = %e,
234 "hela: vector dimension mismatch — HL-F5 disabled for this collection"
235 );
236 return Ok(Vec::new());
237 }
238 return Err(e);
239 }
240 };
241
242 if params.step_budget.is_some_and(|b| t_anchor.elapsed() > b) {
243 tracing::warn!(
244 elapsed_ms = t_anchor.elapsed().as_millis(),
245 "hela: anchor ANN over budget"
246 );
247 return Ok(Vec::new());
248 }
249
250 let Some(anchor_point) = anchor_results.first() else {
251 tracing::debug!("hela: no anchor found, returning empty");
252 return Ok(Vec::new());
253 };
254 let Some(anchor_entity_id) = anchor_point
255 .payload
256 .get("entity_id")
257 .and_then(serde_json::Value::as_i64)
258 else {
259 tracing::warn!("hela: anchor point missing entity_id payload");
260 return Ok(Vec::new());
261 };
262 let anchor_cosine = anchor_point.score;
263
264 tracing::Span::current().record("anchor_id", anchor_entity_id);
265 tracing::debug!(anchor_entity_id, anchor_cosine, "hela: anchor resolved");
266
267 let spread_depth = params.spread_depth.clamp(1, 6);
268
269 let mut visited: HashMap<i64, (u32, f32, Option<i64>)> = HashMap::new();
272 visited.insert(anchor_entity_id, (0, 1.0, None));
273
274 let mut edge_cache: HashMap<i64, Edge> = HashMap::new();
278 let mut frontier: Vec<i64> = vec![anchor_entity_id];
279
280 for hop in 0..spread_depth {
281 if frontier.is_empty() {
282 break;
283 }
284
285 tracing::debug!(hop, frontier_size = frontier.len(), "hela: starting hop");
286
287 let t_step = Instant::now();
288 let edges = store
289 .edges_for_entities(&frontier, ¶ms.edge_types)
290 .await?;
291 if params.step_budget.is_some_and(|b| t_step.elapsed() > b) {
292 tracing::warn!(
293 hop,
294 elapsed_ms = t_step.elapsed().as_millis(),
295 "hela: edge-fetch over budget"
296 );
297 return Ok(Vec::new());
298 }
299
300 let mut next_frontier: Vec<i64> = Vec::new();
301 let mut next_frontier_set: HashSet<i64> = HashSet::new();
302
303 for edge in &edges {
304 edge_cache.entry(edge.id).or_insert_with(|| edge.clone());
306
307 for &src_id in &frontier {
308 let neighbor = if edge.source_entity_id == src_id {
309 edge.target_entity_id
310 } else if edge.target_entity_id == src_id {
311 edge.source_entity_id
312 } else {
313 continue;
314 };
315
316 let parent_pw = visited.get(&src_id).map_or(1.0, |&(_, pw, _)| pw);
317 let new_pw = parent_pw * edge.weight;
318
319 let entry = visited
324 .entry(neighbor)
325 .or_insert((hop + 1, 0.0_f32, Some(edge.id)));
326 if new_pw > entry.1
328 || ((new_pw - entry.1).abs() < f32::EPSILON && hop + 1 < entry.0)
329 {
330 *entry = (hop + 1, new_pw, Some(edge.id));
331 if next_frontier_set.insert(neighbor) {
332 next_frontier.push(neighbor);
333 }
334 }
335
336 if visited.len() >= params.max_visited {
337 break;
338 }
339 }
340
341 if visited.len() >= params.max_visited {
342 break;
343 }
344 }
345
346 tracing::debug!(
347 hop,
348 edges_fetched = edges.len(),
349 visited = visited.len(),
350 next_frontier = next_frontier.len(),
351 "hela: hop complete"
352 );
353
354 frontier = next_frontier;
355 if visited.len() >= params.max_visited {
356 break;
357 }
358 }
359
360 if visited.len() == 1 {
363 tracing::Span::current().record("fallback", true);
364 tracing::debug!(
365 anchor_entity_id,
366 anchor_cosine,
367 "hela: anchor isolated, falling back to pure ANN"
368 );
369 let fact = HelaFact {
370 edge: Edge::synthetic_anchor(anchor_entity_id),
371 score: anchor_cosine,
372 depth: 0,
373 path_weight: 1.0,
374 cosine: Some(anchor_cosine.clamp(0.0, 1.0)),
375 };
376 return Ok(vec![fact]);
377 }
378
379 let entity_ids: Vec<i64> = visited.keys().copied().collect();
381 let point_id_map = store.qdrant_point_ids_for_entities(&entity_ids).await?;
382 let point_ids: Vec<String> = point_id_map.values().cloned().collect();
383
384 let t_vec = Instant::now();
385 let vec_map = embeddings
386 .get_vectors_from_collection(ENTITY_COLLECTION, &point_ids)
387 .await?;
388 if params.step_budget.is_some_and(|b| t_vec.elapsed() > b) {
389 tracing::warn!(
390 elapsed_ms = t_vec.elapsed().as_millis(),
391 "hela: vectors-batch over budget"
392 );
393 return Ok(Vec::new());
394 }
395
396 let mut facts: Vec<HelaFact> = Vec::with_capacity(visited.len().saturating_sub(1));
401 for (&entity_id, &(depth, path_weight, edge_id_opt)) in &visited {
402 if entity_id == anchor_entity_id {
403 continue;
404 }
405 let Some(edge_id) = edge_id_opt else {
406 continue;
407 };
408 let Some(point_id) = point_id_map.get(&entity_id) else {
409 continue;
410 };
411 let Some(node_vec) = vec_map.get(point_id) else {
412 continue;
413 };
414 if node_vec.len() != q_vec.len() {
415 continue;
417 }
418 let cosine_clamped = cosine(&q_vec, node_vec).max(0.0);
419 let fact_score = path_weight * cosine_clamped;
420 let Some(edge) = edge_cache.get(&edge_id).cloned() else {
421 continue;
422 };
423 facts.push(HelaFact {
424 edge,
425 score: fact_score,
426 depth,
427 path_weight,
428 cosine: Some(cosine_clamped),
429 });
430 }
431
432 facts.sort_by(|a, b| b.score.total_cmp(&a.score));
434 facts.truncate(limit);
435
436 if hebbian_enabled {
441 let edge_ids: Vec<i64> = facts
442 .iter()
443 .map(|f| f.edge.id)
444 .filter(|&id| id != 0) .collect();
446 if !edge_ids.is_empty()
447 && let Err(e) = store.apply_hebbian_increment(&edge_ids, hebbian_lr).await
448 {
449 tracing::warn!(error = %e, "hela: hebbian increment failed");
450 }
451 }
452
453 tracing::Span::current().record("visited", visited.len());
454 tracing::Span::current().record("scored", facts.len());
455
456 Ok(facts)
457}
458
459pub struct SpreadingActivation {
463 params: SpreadingActivationParams,
464}
465
466impl SpreadingActivation {
467 #[must_use]
472 pub fn new(params: SpreadingActivationParams) -> Self {
473 Self { params }
474 }
475
476 pub async fn spread(
492 &self,
493 store: &GraphStore,
494 seeds: HashMap<i64, f32>,
495 edge_types: &[EdgeType],
496 ) -> Result<(Vec<ActivatedNode>, Vec<ActivatedFact>), MemoryError> {
497 if seeds.is_empty() {
498 return Ok((Vec::new(), Vec::new()));
499 }
500
501 let now_secs: i64 = SystemTime::now()
504 .duration_since(UNIX_EPOCH)
505 .map_or(0, |d| d.as_secs().cast_signed());
506
507 let mut activation = self.initialize_seeds(&seeds);
508 let mut activated_facts: Vec<ActivatedFact> = Vec::new();
509
510 for hop in 0..self.params.max_hops {
511 let active_nodes: Vec<(i64, f32)> = activation
512 .iter()
513 .filter(|(_, (score, _))| *score >= self.params.activation_threshold)
514 .map(|(&id, &(score, _))| (id, score))
515 .collect();
516
517 if active_nodes.is_empty() {
518 break;
519 }
520
521 let node_ids: Vec<i64> = active_nodes.iter().map(|(id, _)| *id).collect();
522 let edges = store.edges_for_entities(&node_ids, edge_types).await?;
523 let edge_count = edges.len();
524
525 let next_activation =
526 self.propagate_one_hop(hop, &active_nodes, &edges, &activation, now_secs);
527
528 let pruned_count = self.merge_and_prune(&mut activation, next_activation);
529
530 tracing::debug!(
531 hop,
532 active_nodes = active_nodes.len(),
533 edges_fetched = edge_count,
534 after_merge = activation.len(),
535 pruned = pruned_count,
536 "spreading activation: hop complete"
537 );
538
539 self.collect_activated_facts(&edges, &activation, &mut activated_facts);
540 }
541
542 let result = self.finalize(activation);
543
544 tracing::info!(
545 activated = result.len(),
546 facts = activated_facts.len(),
547 "spreading activation: complete"
548 );
549
550 Ok((result, activated_facts))
551 }
552
553 fn initialize_seeds(&self, seeds: &HashMap<i64, f32>) -> HashMap<i64, (f32, u32)> {
555 let mut activation: HashMap<i64, (f32, u32)> = HashMap::new();
556 let mut seed_count = 0usize;
557 for (entity_id, match_score) in seeds {
559 if *match_score < self.params.activation_threshold {
560 tracing::debug!(
561 entity_id,
562 score = match_score,
563 threshold = self.params.activation_threshold,
564 "spreading activation: seed below threshold, skipping"
565 );
566 continue;
567 }
568 activation.insert(*entity_id, (*match_score, 0));
569 seed_count += 1;
570 }
571 tracing::debug!(
572 seeds = seed_count,
573 "spreading activation: initialized seeds"
574 );
575 activation
576 }
577
578 fn propagate_one_hop(
582 &self,
583 hop: u32,
584 active_nodes: &[(i64, f32)],
585 edges: &[Edge],
586 activation: &HashMap<i64, (f32, u32)>,
587 now_secs: i64,
588 ) -> HashMap<i64, (f32, u32)> {
589 let mut next_activation: HashMap<i64, (f32, u32)> = HashMap::new();
590
591 for edge in edges {
592 for &(active_id, node_score) in active_nodes {
593 let neighbor = if edge.source_entity_id == active_id {
594 edge.target_entity_id
595 } else if edge.target_entity_id == active_id {
596 edge.source_entity_id
597 } else {
598 continue;
599 };
600
601 let current_score = activation.get(&neighbor).map_or(0.0_f32, |&(s, _)| s);
606 let next_score = next_activation.get(&neighbor).map_or(0.0_f32, |&(s, _)| s);
607 if current_score >= self.params.inhibition_threshold
608 || next_score >= self.params.inhibition_threshold
609 {
610 continue;
611 }
612
613 let recency = self.recency_weight(&edge.valid_from, now_secs);
614 let blended = self.params.alpha * edge.confidence_fast
616 + (1.0 - self.params.alpha) * edge.confidence_slow;
617 let edge_weight = evolved_weight(edge.retrieval_count, blended);
618 let type_w = edge_type_weight(edge.edge_type);
619 let spread_value =
620 node_score * self.params.decay_lambda * edge_weight * recency * type_w;
621
622 if spread_value < self.params.activation_threshold {
623 continue;
624 }
625
626 let depth_at_max = hop + 1;
629 let entry = next_activation
630 .entry(neighbor)
631 .or_insert((0.0, depth_at_max));
632 let new_score = (entry.0 + spread_value).min(1.0);
633 if new_score > entry.0 {
634 entry.0 = new_score;
635 entry.1 = depth_at_max;
636 }
637 }
638 }
639
640 next_activation
641 }
642
643 fn merge_and_prune(
647 &self,
648 activation: &mut HashMap<i64, (f32, u32)>,
649 next_activation: HashMap<i64, (f32, u32)>,
650 ) -> usize {
651 for (node_id, (new_score, new_depth)) in next_activation {
652 let entry = activation.entry(node_id).or_insert((0.0, new_depth));
653 if new_score > entry.0 {
654 entry.0 = new_score;
655 entry.1 = new_depth;
656 }
657 }
658
659 if activation.len() > self.params.max_activated_nodes {
660 let before = activation.len();
661 let mut entries: Vec<(i64, (f32, u32))> = activation.drain().collect();
662 entries.sort_by(|(_, (a, _)), (_, (b, _))| b.total_cmp(a));
663 entries.truncate(self.params.max_activated_nodes);
664 *activation = entries.into_iter().collect();
665 before - self.params.max_activated_nodes
666 } else {
667 0
668 }
669 }
670
671 fn collect_activated_facts(
673 &self,
674 edges: &[Edge],
675 activation: &HashMap<i64, (f32, u32)>,
676 activated_facts: &mut Vec<ActivatedFact>,
677 ) {
678 for edge in edges {
679 let src_score = activation
680 .get(&edge.source_entity_id)
681 .map_or(0.0, |&(s, _)| s);
682 let tgt_score = activation
683 .get(&edge.target_entity_id)
684 .map_or(0.0, |&(s, _)| s);
685 if src_score >= self.params.activation_threshold
686 && tgt_score >= self.params.activation_threshold
687 {
688 let activation_score = src_score.max(tgt_score);
689 activated_facts.push(ActivatedFact {
690 edge: edge.clone(),
691 activation_score,
692 is_implicit_conflict: false,
693 conflict_candidate_id: None,
694 });
695 }
696 }
697 }
698
699 fn finalize(&self, activation: HashMap<i64, (f32, u32)>) -> Vec<ActivatedNode> {
701 let mut result: Vec<ActivatedNode> = activation
702 .into_iter()
703 .filter(|(_, (score, _))| *score >= self.params.activation_threshold)
704 .map(|(entity_id, (activation, depth))| ActivatedNode {
705 entity_id,
706 activation,
707 depth,
708 })
709 .collect();
710 result.sort_by(|a, b| b.activation.total_cmp(&a.activation));
711 result
712 }
713
714 #[allow(clippy::cast_precision_loss)]
720 fn recency_weight(&self, valid_from: &str, now_secs: i64) -> f32 {
721 if self.params.temporal_decay_rate <= 0.0 {
722 return 1.0;
723 }
724 let Some(valid_from_secs) = parse_sqlite_datetime_to_unix(valid_from) else {
725 return 1.0;
726 };
727 let age_secs = (now_secs - valid_from_secs).max(0);
728 let age_days = age_secs as f64 / 86_400.0;
729 let weight = 1.0_f64 / (1.0 + age_days * self.params.temporal_decay_rate);
730 #[allow(clippy::cast_possible_truncation)]
732 let w = weight as f32;
733 w
734 }
735}
736
737#[cfg(test)]
738mod tests {
739 use super::*;
740 use crate::graph::GraphStore;
741 use crate::graph::types::EntityType;
742 use crate::store::SqliteStore;
743
744 async fn setup_store() -> GraphStore {
745 let store = SqliteStore::new(":memory:").await.unwrap();
746 GraphStore::new(store.pool().clone())
747 }
748
749 fn default_params() -> SpreadingActivationParams {
750 SpreadingActivationParams {
751 decay_lambda: 0.85,
752 max_hops: 3,
753 activation_threshold: 0.1,
754 inhibition_threshold: 0.8,
755 max_activated_nodes: 50,
756 temporal_decay_rate: 0.0,
757 seed_structural_weight: 0.4,
758 seed_community_cap: 3,
759 alpha: 0.3,
760 }
761 }
762
763 #[tokio::test]
766 async fn spread_empty_graph_no_edges_no_facts() {
767 let store = setup_store().await;
768 let sa = SpreadingActivation::new(default_params());
769 let seeds = HashMap::from([(1_i64, 1.0_f32)]);
770 let (nodes, facts) = sa.spread(&store, seeds, &[]).await.unwrap();
771 assert_eq!(nodes.len(), 1, "seed must be in activated nodes");
773 assert_eq!(nodes[0].entity_id, 1);
774 assert!((nodes[0].activation - 1.0).abs() < 1e-6);
775 assert!(
777 facts.is_empty(),
778 "expected no activated facts on empty graph"
779 );
780 }
781
782 #[tokio::test]
784 async fn spread_empty_seeds_returns_empty() {
785 let store = setup_store().await;
786 let sa = SpreadingActivation::new(default_params());
787 let (nodes, facts) = sa.spread(&store, HashMap::new(), &[]).await.unwrap();
788 assert!(nodes.is_empty());
789 assert!(facts.is_empty());
790 }
791
792 #[tokio::test]
794 async fn spread_single_seed_no_edges_returns_seed() {
795 let store = setup_store().await;
796 let alice = store
797 .upsert_entity("Alice", "Alice", EntityType::Person, None, None)
798 .await
799 .unwrap()
800 .0;
801
802 let sa = SpreadingActivation::new(default_params());
803 let seeds = HashMap::from([(alice, 1.0_f32)]);
804 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
805 assert_eq!(nodes.len(), 1);
806 assert_eq!(nodes[0].entity_id, alice);
807 assert_eq!(nodes[0].depth, 0);
808 assert!((nodes[0].activation - 1.0).abs() < 1e-6);
809 }
810
811 #[tokio::test]
813 async fn spread_linear_chain_all_activated_with_decay() {
814 let store = setup_store().await;
815 let a = store
816 .upsert_entity("A", "A", EntityType::Person, None, None)
817 .await
818 .unwrap()
819 .0;
820 let b = store
821 .upsert_entity("B", "B", EntityType::Person, None, None)
822 .await
823 .unwrap()
824 .0;
825 let c = store
826 .upsert_entity("C", "C", EntityType::Person, None, None)
827 .await
828 .unwrap()
829 .0;
830 store
831 .insert_edge(a, b, "knows", "A knows B", 1.0, None, None)
832 .await
833 .unwrap();
834 store
835 .insert_edge(b, c, "knows", "B knows C", 1.0, None, None)
836 .await
837 .unwrap();
838
839 let mut cfg = default_params();
840 cfg.max_hops = 3;
841 cfg.decay_lambda = 0.9;
842 let sa = SpreadingActivation::new(cfg);
843 let seeds = HashMap::from([(a, 1.0_f32)]);
844 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
845
846 let ids: Vec<i64> = nodes.iter().map(|n| n.entity_id).collect();
847 assert!(ids.contains(&a), "A (seed) must be activated");
848 assert!(ids.contains(&b), "B (hop 1) must be activated");
849 assert!(ids.contains(&c), "C (hop 2) must be activated");
850
851 let score_a = nodes.iter().find(|n| n.entity_id == a).unwrap().activation;
853 let score_b = nodes.iter().find(|n| n.entity_id == b).unwrap().activation;
854 let score_c = nodes.iter().find(|n| n.entity_id == c).unwrap().activation;
855 assert!(
856 score_a > score_b,
857 "seed A should have higher activation than hop-1 B"
858 );
859 assert!(
860 score_b > score_c,
861 "hop-1 B should have higher activation than hop-2 C"
862 );
863 }
864
865 #[tokio::test]
867 async fn spread_linear_chain_max_hops_limits_reach() {
868 let store = setup_store().await;
869 let a = store
870 .upsert_entity("A", "A", EntityType::Person, None, None)
871 .await
872 .unwrap()
873 .0;
874 let b = store
875 .upsert_entity("B", "B", EntityType::Person, None, None)
876 .await
877 .unwrap()
878 .0;
879 let c = store
880 .upsert_entity("C", "C", EntityType::Person, None, None)
881 .await
882 .unwrap()
883 .0;
884 store
885 .insert_edge(a, b, "knows", "A knows B", 1.0, None, None)
886 .await
887 .unwrap();
888 store
889 .insert_edge(b, c, "knows", "B knows C", 1.0, None, None)
890 .await
891 .unwrap();
892
893 let mut cfg = default_params();
894 cfg.max_hops = 1;
895 let sa = SpreadingActivation::new(cfg);
896 let seeds = HashMap::from([(a, 1.0_f32)]);
897 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
898
899 let ids: Vec<i64> = nodes.iter().map(|n| n.entity_id).collect();
900 assert!(ids.contains(&a), "A must be activated (seed)");
901 assert!(ids.contains(&b), "B must be activated (hop 1)");
902 assert!(!ids.contains(&c), "C must NOT be activated with max_hops=1");
903 }
904
905 #[tokio::test]
909 async fn spread_diamond_graph_convergence() {
910 let store = setup_store().await;
911 let a = store
912 .upsert_entity("A", "A", EntityType::Person, None, None)
913 .await
914 .unwrap()
915 .0;
916 let b = store
917 .upsert_entity("B", "B", EntityType::Person, None, None)
918 .await
919 .unwrap()
920 .0;
921 let c = store
922 .upsert_entity("C", "C", EntityType::Person, None, None)
923 .await
924 .unwrap()
925 .0;
926 let d = store
927 .upsert_entity("D", "D", EntityType::Person, None, None)
928 .await
929 .unwrap()
930 .0;
931 store
932 .insert_edge(a, b, "rel", "A-B", 1.0, None, None)
933 .await
934 .unwrap();
935 store
936 .insert_edge(a, c, "rel", "A-C", 1.0, None, None)
937 .await
938 .unwrap();
939 store
940 .insert_edge(b, d, "rel", "B-D", 1.0, None, None)
941 .await
942 .unwrap();
943 store
944 .insert_edge(c, d, "rel", "C-D", 1.0, None, None)
945 .await
946 .unwrap();
947
948 let mut cfg = default_params();
949 cfg.max_hops = 3;
950 cfg.decay_lambda = 0.9;
951 cfg.inhibition_threshold = 0.95; let sa = SpreadingActivation::new(cfg);
953 let seeds = HashMap::from([(a, 1.0_f32)]);
954 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
955
956 let ids: Vec<i64> = nodes.iter().map(|n| n.entity_id).collect();
957 assert!(ids.contains(&d), "D must be activated via diamond paths");
958
959 let node_d = nodes.iter().find(|n| n.entity_id == d).unwrap();
961 assert_eq!(node_d.depth, 2, "D should be at depth 2");
962 }
963
964 #[tokio::test]
966 async fn spread_inhibition_prevents_runaway() {
967 let store = setup_store().await;
968 let hub = store
970 .upsert_entity("Hub", "Hub", EntityType::Concept, None, None)
971 .await
972 .unwrap()
973 .0;
974
975 for i in 0..5 {
976 let leaf = store
977 .upsert_entity(
978 &format!("Leaf{i}"),
979 &format!("Leaf{i}"),
980 EntityType::Concept,
981 None,
982 None,
983 )
984 .await
985 .unwrap()
986 .0;
987 store
988 .insert_edge(
989 hub,
990 leaf,
991 "has",
992 &format!("Hub has Leaf{i}"),
993 1.0,
994 None,
995 None,
996 )
997 .await
998 .unwrap();
999 store
1001 .insert_edge(
1002 leaf,
1003 hub,
1004 "part_of",
1005 &format!("Leaf{i} part_of Hub"),
1006 1.0,
1007 None,
1008 None,
1009 )
1010 .await
1011 .unwrap();
1012 }
1013
1014 let mut cfg = default_params();
1016 cfg.inhibition_threshold = 0.8;
1017 cfg.max_hops = 3;
1018 let sa = SpreadingActivation::new(cfg);
1019 let seeds = HashMap::from([(hub, 1.0_f32)]);
1020 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
1021
1022 let hub_node = nodes.iter().find(|n| n.entity_id == hub);
1024 assert!(hub_node.is_some(), "hub must be in results");
1025 assert!(
1026 hub_node.unwrap().activation <= 1.0,
1027 "activation must not exceed 1.0"
1028 );
1029 }
1030
1031 #[tokio::test]
1033 async fn spread_max_activated_nodes_cap_enforced() {
1034 let store = setup_store().await;
1035 let root = store
1036 .upsert_entity("Root", "Root", EntityType::Person, None, None)
1037 .await
1038 .unwrap()
1039 .0;
1040
1041 for i in 0..20 {
1043 let leaf = store
1044 .upsert_entity(
1045 &format!("Node{i}"),
1046 &format!("Node{i}"),
1047 EntityType::Concept,
1048 None,
1049 None,
1050 )
1051 .await
1052 .unwrap()
1053 .0;
1054 store
1055 .insert_edge(
1056 root,
1057 leaf,
1058 "has",
1059 &format!("Root has Node{i}"),
1060 0.9,
1061 None,
1062 None,
1063 )
1064 .await
1065 .unwrap();
1066 }
1067
1068 let max_nodes = 5;
1069 let cfg = SpreadingActivationParams {
1070 max_activated_nodes: max_nodes,
1071 max_hops: 2,
1072 ..default_params()
1073 };
1074 let sa = SpreadingActivation::new(cfg);
1075 let seeds = HashMap::from([(root, 1.0_f32)]);
1076 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
1077
1078 assert!(
1079 nodes.len() <= max_nodes,
1080 "activation must be capped at {max_nodes} nodes, got {}",
1081 nodes.len()
1082 );
1083 }
1084
1085 #[tokio::test]
1087 async fn spread_temporal_decay_recency_effect() {
1088 let store = setup_store().await;
1089 let src = store
1090 .upsert_entity("Src", "Src", EntityType::Person, None, None)
1091 .await
1092 .unwrap()
1093 .0;
1094 let recent = store
1095 .upsert_entity("Recent", "Recent", EntityType::Tool, None, None)
1096 .await
1097 .unwrap()
1098 .0;
1099 let old = store
1100 .upsert_entity("Old", "Old", EntityType::Tool, None, None)
1101 .await
1102 .unwrap()
1103 .0;
1104
1105 store
1107 .insert_edge(src, recent, "uses", "Src uses Recent", 1.0, None, None)
1108 .await
1109 .unwrap();
1110
1111 zeph_db::query(
1113 sql!("INSERT INTO graph_edges (source_entity_id, target_entity_id, relation, fact, confidence, valid_from)
1114 VALUES (?1, ?2, 'uses', 'Src uses Old', 1.0, '1970-01-01 00:00:00')"),
1115 )
1116 .bind(src)
1117 .bind(old)
1118 .execute(store.pool())
1119 .await
1120 .unwrap();
1121
1122 let mut cfg = default_params();
1123 cfg.max_hops = 2;
1124 let sa = SpreadingActivation::new(SpreadingActivationParams {
1126 temporal_decay_rate: 0.5,
1127 ..cfg
1128 });
1129 let seeds = HashMap::from([(src, 1.0_f32)]);
1130 let (nodes, _) = sa.spread(&store, seeds, &[]).await.unwrap();
1131
1132 let score_recent = nodes
1133 .iter()
1134 .find(|n| n.entity_id == recent)
1135 .map_or(0.0, |n| n.activation);
1136 let score_old = nodes
1137 .iter()
1138 .find(|n| n.entity_id == old)
1139 .map_or(0.0, |n| n.activation);
1140
1141 assert!(
1142 score_recent > score_old,
1143 "recent edge ({score_recent}) must produce higher activation than old edge ({score_old})"
1144 );
1145 }
1146
1147 #[tokio::test]
1149 async fn spread_edge_type_filter_excludes_other_types() {
1150 let store = setup_store().await;
1151 let a = store
1152 .upsert_entity("A", "A", EntityType::Person, None, None)
1153 .await
1154 .unwrap()
1155 .0;
1156 let b_semantic = store
1157 .upsert_entity("BSemantic", "BSemantic", EntityType::Tool, None, None)
1158 .await
1159 .unwrap()
1160 .0;
1161 let c_causal = store
1162 .upsert_entity("CCausal", "CCausal", EntityType::Concept, None, None)
1163 .await
1164 .unwrap()
1165 .0;
1166
1167 store
1169 .insert_edge(a, b_semantic, "uses", "A uses BSemantic", 1.0, None, None)
1170 .await
1171 .unwrap();
1172
1173 zeph_db::query(
1175 sql!("INSERT INTO graph_edges (source_entity_id, target_entity_id, relation, fact, confidence, valid_from, edge_type)
1176 VALUES (?1, ?2, 'caused', 'A caused CCausal', 1.0, datetime('now'), 'causal')"),
1177 )
1178 .bind(a)
1179 .bind(c_causal)
1180 .execute(store.pool())
1181 .await
1182 .unwrap();
1183
1184 let cfg = default_params();
1185 let sa = SpreadingActivation::new(cfg);
1186
1187 let seeds = HashMap::from([(a, 1.0_f32)]);
1189 let (nodes, _) = sa
1190 .spread(&store, seeds, &[EdgeType::Semantic])
1191 .await
1192 .unwrap();
1193
1194 let ids: Vec<i64> = nodes.iter().map(|n| n.entity_id).collect();
1195 assert!(
1196 ids.contains(&b_semantic),
1197 "BSemantic must be activated via semantic edge"
1198 );
1199 assert!(
1200 !ids.contains(&c_causal),
1201 "CCausal must NOT be activated when filtering to semantic only"
1202 );
1203 }
1204
1205 #[tokio::test]
1207 async fn spread_large_seed_list() {
1208 let store = setup_store().await;
1209 let mut seeds = HashMap::new();
1210
1211 for i in 0..100i64 {
1213 let id = store
1214 .upsert_entity(
1215 &format!("Entity{i}"),
1216 &format!("entity{i}"),
1217 EntityType::Concept,
1218 None,
1219 None,
1220 )
1221 .await
1222 .unwrap()
1223 .0;
1224 seeds.insert(id, 1.0_f32);
1225 }
1226
1227 let cfg = default_params();
1228 let sa = SpreadingActivation::new(cfg);
1229 let result = sa.spread(&store, seeds, &[]).await;
1231 assert!(
1232 result.is_ok(),
1233 "large seed list must not error: {:?}",
1234 result.err()
1235 );
1236 }
1237
1238 #[test]
1241 fn hela_cosine_identical_vectors() {
1242 let v = vec![1.0_f32, 0.0, 0.0];
1243 assert!(
1244 (cosine(&v, &v) - 1.0).abs() < 1e-6,
1245 "identical vectors → cosine 1.0"
1246 );
1247 }
1248
1249 #[test]
1250 fn hela_cosine_orthogonal_vectors() {
1251 let a = vec![1.0_f32, 0.0];
1252 let b = vec![0.0_f32, 1.0];
1253 assert!(
1254 cosine(&a, &b).abs() < 1e-6,
1255 "orthogonal vectors → cosine 0.0"
1256 );
1257 }
1258
1259 #[test]
1260 fn hela_cosine_anti_correlated() {
1261 let a = vec![1.0_f32, 0.0];
1262 let b = vec![-1.0_f32, 0.0];
1263 assert!(
1264 cosine(&a, &b) < 0.0,
1265 "anti-correlated vectors → negative cosine"
1266 );
1267 }
1268
1269 #[test]
1270 fn hela_cosine_zero_vector_no_panic() {
1271 let a = vec![0.0_f32, 0.0];
1272 let b = vec![1.0_f32, 0.0];
1273 let result = cosine(&a, &b);
1275 assert!(
1276 result.is_finite(),
1277 "zero-norm vector must yield finite cosine"
1278 );
1279 }
1280
1281 #[test]
1282 fn hela_spread_params_default_depth_is_two() {
1283 let p = HelaSpreadParams::default();
1284 assert_eq!(p.spread_depth, 2);
1285 assert!(p.step_budget.is_some());
1286 assert!(p.edge_types.is_empty());
1287 assert_eq!(p.max_visited, 200);
1288 }
1289
1290 #[test]
1291 fn hela_spread_params_default_embed_timeout_is_some() {
1292 let p = HelaSpreadParams::default();
1293 assert!(
1294 p.embed_timeout.is_some(),
1295 "default embed_timeout must be Some (5 s)"
1296 );
1297 }
1298
1299 #[tokio::test]
1302 async fn hela_spreading_recall_embed_timeout_returns_error() {
1303 use std::time::Duration;
1304 use zeph_llm::any::AnyProvider;
1305 use zeph_llm::mock::MockProvider;
1306
1307 use crate::embedding_store::EmbeddingStore;
1308 use crate::error::MemoryError;
1309 use crate::in_memory_store::InMemoryVectorStore;
1310
1311 let store = setup_store().await;
1312
1313 let mock = MockProvider::default().with_embed_delay(500);
1315 let provider = AnyProvider::Mock(mock);
1316
1317 let sqlite = crate::store::SqliteStore::with_pool_size(":memory:", 1)
1318 .await
1319 .unwrap();
1320 let embeddings =
1321 EmbeddingStore::with_store(Box::new(InMemoryVectorStore::new()), sqlite.pool().clone());
1322
1323 let params = HelaSpreadParams {
1324 embed_timeout: Some(Duration::from_millis(50)),
1325 ..Default::default()
1326 };
1327
1328 let result = hela_spreading_recall(
1329 &store,
1330 &embeddings,
1331 &provider,
1332 "test query",
1333 5,
1334 ¶ms,
1335 false,
1336 0.0,
1337 )
1338 .await;
1339
1340 assert!(
1341 matches!(result, Err(MemoryError::Timeout(_))),
1342 "expected Err(MemoryError::Timeout), got {result:?}"
1343 );
1344 }
1345
1346 #[tokio::test]
1349 async fn hela_spreading_recall_no_timeout_does_not_wrap() {
1350 use zeph_llm::any::AnyProvider;
1351 use zeph_llm::mock::MockProvider;
1352
1353 use crate::embedding_store::EmbeddingStore;
1354 use crate::in_memory_store::InMemoryVectorStore;
1355
1356 let store = setup_store().await;
1357
1358 let mock = MockProvider::default().with_embed_delay(0);
1359 let provider = AnyProvider::Mock(mock);
1360
1361 let sqlite = crate::store::SqliteStore::with_pool_size(":memory:", 1)
1362 .await
1363 .unwrap();
1364 let embeddings =
1365 EmbeddingStore::with_store(Box::new(InMemoryVectorStore::new()), sqlite.pool().clone());
1366
1367 let params = HelaSpreadParams {
1368 embed_timeout: None,
1369 ..Default::default()
1370 };
1371
1372 let result = hela_spreading_recall(
1375 &store,
1376 &embeddings,
1377 &provider,
1378 "test query",
1379 5,
1380 ¶ms,
1381 false,
1382 0.0,
1383 )
1384 .await;
1385
1386 assert!(
1389 !matches!(result, Err(crate::error::MemoryError::Timeout(_))),
1390 "embed_timeout: None must not produce a Timeout error, got {result:?}"
1391 );
1392 }
1393
1394 #[test]
1395 fn hela_synthetic_anchor_edge_id_is_zero() {
1396 let edge = Edge::synthetic_anchor(42);
1397 assert_eq!(
1398 edge.id, 0,
1399 "synthetic anchor must have id = 0 to be excluded from Hebbian"
1400 );
1401 assert_eq!(edge.source_entity_id, 42);
1402 assert_eq!(edge.target_entity_id, 42);
1403 }
1404
1405 #[test]
1406 fn hela_negative_cosine_clamped_to_zero_in_score() {
1407 let anti = vec![-1.0_f32, 0.0];
1409 let query = vec![1.0_f32, 0.0];
1410 let cosine_raw = cosine(&query, &anti);
1411 assert!(cosine_raw < 0.0);
1412 let clamped = cosine_raw.max(0.0);
1413 let fact_score = 0.9_f32 * clamped;
1414 assert!(
1415 fact_score < f32::EPSILON,
1416 "anti-correlated score must be 0.0"
1417 );
1418 }
1419
1420 #[test]
1421 fn hela_path_weight_multiplicative() {
1422 let w1 = 0.8_f32;
1424 let w2 = 0.5_f32;
1425 let expected = w1 * w2;
1426 assert!((expected - 0.4).abs() < 1e-6);
1427 }
1428
1429 #[test]
1430 fn hela_max_path_weight_on_multipath() {
1431 let pw_a = 0.9_f32; let pw_b = 0.3_f32; let kept = pw_a.max(pw_b);
1435 assert!(
1436 (kept - 0.9).abs() < 1e-6,
1437 "multi-path resolution must keep maximum path_weight"
1438 );
1439 }
1440
1441 #[test]
1442 fn hela_fact_score_formula() {
1443 let path_weight = 0.8_f32;
1444 let cosine_clamped = 0.75_f32;
1445 let expected = path_weight * cosine_clamped;
1446 assert!((expected - 0.6).abs() < 1e-5);
1448 }
1449
1450 #[tokio::test]
1455 async fn synapse_blend_uses_alpha_not_raw_confidence() {
1456 let store = SqliteStore::new(":memory:").await.unwrap();
1457 let gs = GraphStore::new(store.pool().clone()).with_benna_rates(0.5, 0.05);
1458
1459 let alpha = 0.3_f32;
1460 let params = SpreadingActivationParams {
1461 alpha,
1462 ..default_params()
1463 };
1464
1465 let src = gs
1466 .upsert_entity("Blend_src", "Blend_src", EntityType::Person, None, None)
1467 .await
1468 .unwrap()
1469 .0;
1470 let tgt = gs
1471 .upsert_entity("Blend_tgt", "Blend_tgt", EntityType::Concept, None, None)
1472 .await
1473 .unwrap()
1474 .0;
1475
1476 gs.insert_edge_typed(
1478 src,
1479 tgt,
1480 "knows",
1481 "Blend_src knows Blend_tgt",
1482 0.5,
1483 None,
1484 crate::graph::types::EdgeType::Semantic,
1485 None,
1486 None,
1487 )
1488 .await
1489 .unwrap();
1490 gs.insert_edge_typed(
1492 src,
1493 tgt,
1494 "knows",
1495 "Blend_src knows Blend_tgt",
1496 0.9,
1497 None,
1498 crate::graph::types::EdgeType::Semantic,
1499 None,
1500 None,
1501 )
1502 .await
1503 .unwrap();
1504
1505 let sa = SpreadingActivation::new(params);
1506 let seeds = HashMap::from([(src, 1.0_f32)]);
1507 let (_nodes, facts) = sa.spread(&gs, seeds, &[]).await.unwrap();
1508
1509 assert!(!facts.is_empty(), "spread must return at least one fact");
1514 let raw_score = facts[0].activation_score;
1515 assert!(
1517 (raw_score - 0.9_f32).abs() > 0.05,
1518 "blend score {raw_score} must differ from raw confidence 0.9 — alpha blend not applied"
1519 );
1520 }
1521}