Skip to main content

zeph_memory/graph/
activation.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! SYNAPSE spreading activation retrieval over the entity graph.
5//!
6//! Implements the spreading activation algorithm from arXiv 2601.02744, adapted for
7//! the zeph-memory graph schema. Seeds are matched via fuzzy entity search; activation
8//! propagates hop-by-hop with:
9//! - Exponential decay per hop (`decay_lambda`)
10//! - Edge confidence weighting
11//! - Temporal recency weighting (reuses `GraphConfig.temporal_decay_rate`)
12//! - Lateral inhibition (nodes above `inhibition_threshold` stop receiving activation)
13//! - Per-hop pruning to enforce `max_activated_nodes` bound (SA-INV-04)
14//! - MAGMA edge type filtering via `edge_types` parameter
15
16use 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/// A graph node that was activated during spreading activation.
29#[derive(Debug, Clone)]
30pub struct ActivatedNode {
31    /// Database ID of the activated entity.
32    pub entity_id: i64,
33    /// Final activation score in `[0.0, 1.0]`.
34    pub activation: f32,
35    /// Hop at which the maximum activation was received (`0` = seed).
36    pub depth: u32,
37}
38
39/// A graph edge traversed during spreading activation, with its activation score.
40#[derive(Debug, Clone)]
41pub struct ActivatedFact {
42    /// The traversed edge.
43    pub edge: Edge,
44    /// Activation score of the source or target entity at time of traversal.
45    pub activation_score: f32,
46    /// `true` when this edge has a pending implicit conflict candidate (spec 004-17).
47    pub is_implicit_conflict: bool,
48    /// ID of the `implicit_conflict_candidates` row, if any.
49    pub conflict_candidate_id: Option<i64>,
50}
51
52pub use zeph_common::memory::SpreadingActivationParams;
53
54// ── HL-F5: HeLa-Mem spreading activation (#3346) ─────────────────────────────
55
56/// A graph edge surfaced by HL-F5 spreading activation (#3346), scored by
57/// `path_weight × max(cosine_query_to_endpoint, 0.0)`.
58///
59/// Mirrors [`ActivatedFact`] so callers can dispatch over a single
60/// `Vec<HelaFact>` ↔ `Vec<ActivatedFact>` ↔ `Vec<GraphFact>` shape at the
61/// strategy-selection site.
62#[derive(Debug, Clone)]
63pub struct HelaFact {
64    /// The edge by which the higher-scored endpoint was reached.
65    pub edge: Edge,
66    /// Final HL-F5 score: `path_weight × cosine_clamped`. Range: `[0.0, +∞)`.
67    pub score: f32,
68    /// BFS depth at which `edge` was traversed (`1..=spread_depth`).
69    /// `0` is reserved for the synthetic anchor edge in the isolated-anchor fallback.
70    pub depth: u32,
71    /// Multiplicative product of edge weights along the BFS path that reached
72    /// this edge's far endpoint. Range: `[0.0, +∞)`.
73    pub path_weight: f32,
74    /// Clamped cosine similarity of the far endpoint's entity embedding
75    /// to the query embedding, in `[0.0, 1.0]`. `None` when the endpoint
76    /// has no stored embedding (skipped from results in that case).
77    pub cosine: Option<f32>,
78}
79
80/// Parameters for HL-F5 spreading activation retrieval.
81///
82/// Build via [`Default`] and override individual fields:
83///
84/// ```rust
85/// use zeph_memory::graph::activation::HelaSpreadParams;
86///
87/// let params = HelaSpreadParams { spread_depth: 3, ..Default::default() };
88/// ```
89#[derive(Debug, Clone)]
90pub struct HelaSpreadParams {
91    /// BFS hops. Clamped to `[1, 6]` at runtime. Default: `2`.
92    pub spread_depth: u32,
93    /// MAGMA edge-type filter. Empty = all types. Default: `[]`.
94    pub edge_types: Vec<EdgeType>,
95    /// Soft upper bound on the visited-node set. Default: `200`.
96    pub max_visited: usize,
97    /// Per-step circuit breaker. Any internal step (anchor ANN, edges batch,
98    /// vectors batch) that exceeds this duration triggers an `Ok(Vec::new())`
99    /// fallback with a `WARN`. Default: `Some(80 ms)` — headroom over realistic
100    /// local-Qdrant round-trip latency (the previous `8 ms` default aborted
101    /// almost every call even when Qdrant was healthy, #5785).
102    pub step_budget: Option<std::time::Duration>,
103    /// Timeout for the initial query embedding call. `None` = no timeout.
104    /// Default: `Some(5 s)`.
105    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
120/// Process-global dim-mismatch sentinel for HL-F5 (keyed by collection name).
121///
122/// MINOR-1 resolution: keyed by collection so re-provisioning with a different
123/// dimension recovers after a process restart.  A per-`SemanticMemory` guard would
124/// require passing state down; a process-global string key is the least-invasive
125/// approach that prevents permanent lockout from transient startup errors.
126/// Test isolation: each test constructs its own `HelaSpreadParams` with
127/// a distinct mock collection name to avoid cross-test interference.
128static HELA_DIM_MISMATCH: OnceLock<String> = OnceLock::new();
129
130/// Cosine similarity of two equal-length slices.
131///
132/// Returns `0.0` when either norm is zero (prevents division by zero).
133fn 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/// HL-F5 BFS spreading activation from the top-1 ANN anchor node (#3346).
142///
143/// Algorithm overview:
144/// 1. Embed `query` → anchor via ANN search in the entity Qdrant collection.
145/// 2. BFS up to `params.spread_depth` hops, propagating multiplicative edge
146///    weights (`path_weight = Π edge.weight along path`). Multi-path convergence
147///    keeps the maximum `path_weight`.
148/// 3. Retrieve entity embeddings for all visited nodes via `get_points`.
149/// 4. Score each node: `score = path_weight × max(cosine(query, entity), 0.0)`.
150/// 5. Sort descending, truncate to `limit`, reinforce traversed edges via Hebbian
151///    update (when `hebbian_enabled`).
152///
153/// Fallback: when the anchor entity has no outgoing edges a single synthetic
154/// [`HelaFact`] with `edge.id == 0` and `score = anchor_cosine` is returned
155/// (the real ANN cosine, never a fabricated `1.0`).
156///
157/// Per-step circuit breaker: any individual step exceeding `params.step_budget`
158/// emits a `WARN` and returns `Ok(Vec::new())`.
159///
160/// Dim-mismatch resilience: a one-time dim probe on the first call guards against
161/// collection/provider configuration mismatches (#3382 pattern). Subsequent calls
162/// to a mismatched collection short-circuit immediately.
163///
164/// # Errors
165///
166/// Returns an error if the embed call or any database query fails.
167#[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)] // complex algorithm function; both suppressions justified until the function is decomposed in a future refactor
180pub 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    // ── Step 0: dim-mismatch guard ────────────────────────────────────────────
199    // MINOR-1: guard is keyed by collection name so re-provisioning recovers.
200    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    // ── Step 1: embed query ───────────────────────────────────────────────────
206    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    // Dim probe: search with k=1 to catch dimension mismatch at the Qdrant layer.
218    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    // ── Step 2: BFS with multiplicative path-weight propagation ──────────────
270    // `visited`: entity_id → (depth, path_weight, edge_id_via_which_we_arrived)
271    let mut visited: HashMap<i64, (u32, f32, Option<i64>)> = HashMap::new();
272    visited.insert(anchor_entity_id, (0, 1.0, None));
273
274    // Dedup edges keyed by id for Step 4 lookup (avoids N clones per frontier).
275    // MINOR-3 resolution: collect edges into a HashMap<id, Edge> outside the
276    // per-source loop to avoid 10K clones on a hub × 50-entity frontier.
277    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, &params.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            // Cache by edge id to avoid repeated clones per source in frontier.
305            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                // Multi-path resolution: keep MAX path_weight; lower depth as
320                // tie-break. MINOR-4 note: max_visited is a soft bound — the
321                // actual visited set may exceed it by O(edges_per_hop_step) for
322                // one frontier step before the outer break fires.
323                let entry = visited
324                    .entry(neighbor)
325                    .or_insert((hop + 1, 0.0_f32, Some(edge.id)));
326                // Prefer strictly higher path weight; break ties in favour of shallower depth.
327                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    // ── Isolated-anchor fallback ──────────────────────────────────────────────
361    // `visited.len() == 1` means no edges were traversed from the anchor.
362    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    // ── Step 3: retrieve entity embeddings ───────────────────────────────────
380    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    // ── Step 4: score per visited node ────────────────────────────────────────
397    // Cosine clamped to [0.0, 1.0]: anti-correlated neighbors score 0.0 so
398    // they are ranked below positively-correlated ones.  A negative cosine on a
399    // strongly-reinforced edge would otherwise invert the retrieval signal.
400    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            // Per-node dim mismatch — skip (defense-in-depth for legacy collections).
416            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    // ── Step 5: sort, truncate, Hebbian increment ─────────────────────────────
433    facts.sort_by(|a, b| b.score.total_cmp(&a.score));
434    facts.truncate(limit);
435
436    // HL-F2 reinforcement on edges that survived truncation (kept ≈ used).
437    // Hebbian on "kept edges only" — consistent with graph_recall_activated at
438    // graph/retrieval.rs:427-433. Note: SYNAPSE reinforces all traversed edges;
439    // this PR intentionally reinforces only surfaced edges. See MINOR-5.
440    if hebbian_enabled {
441        let edge_ids: Vec<i64> = facts
442            .iter()
443            .map(|f| f.edge.id)
444            .filter(|&id| id != 0) // skip synthetic anchor
445            .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
459// ── SYNAPSE spreading activation ──────────────────────────────────────────────
460
461/// Spreading activation engine parameterized from [`SpreadingActivationParams`].
462pub struct SpreadingActivation {
463    params: SpreadingActivationParams,
464}
465
466impl SpreadingActivation {
467    /// Create a new spreading activation engine from explicit parameters.
468    ///
469    /// `params.temporal_decay_rate` is taken from `GraphConfig.temporal_decay_rate` so that
470    /// recency weighting reuses the same parameter as BFS recall (SA-INV-05).
471    #[must_use]
472    pub fn new(params: SpreadingActivationParams) -> Self {
473        Self { params }
474    }
475
476    /// Run spreading activation from `seeds` over the graph.
477    ///
478    /// Returns activated nodes sorted by activation score descending, along with
479    /// edges collected during propagation.
480    ///
481    /// # Parameters
482    ///
483    /// - `store`: graph database accessor
484    /// - `seeds`: `HashMap<entity_id, initial_activation>` — nodes to start from
485    /// - `edge_types`: MAGMA subgraph filter; when non-empty, only edges of these types
486    ///   are traversed (mirrors `bfs_typed` behaviour; SA-INV-08)
487    ///
488    /// # Errors
489    ///
490    /// Returns an error if any database query fails.
491    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        // Compute `now_secs` once for consistent temporal recency weighting
502        // across all edges (matches the pattern in retrieval.rs:83-86).
503        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    /// Populate the activation map from seed scores, filtering seeds below threshold.
554    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        // Seeds bypass activation_threshold (they are query anchors per SYNAPSE semantics).
558        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    /// Compute the next-hop activation map by propagating through `edges`.
579    ///
580    /// Applies lateral inhibition (CRIT-02) and clamped multi-path convergence sums.
581    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                // Lateral inhibition: skip neighbor if it already has high activation
602                // in either the current map OR this hop's next_activation (CRIT-02 fix:
603                // checks both maps to match SYNAPSE paper semantics and prevent runaway
604                // activation when multiple paths converge in the same hop).
605                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                // SYNAPSE blend: use Benna-Fusi fast/slow variables instead of raw confidence (#3709).
615                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                // Clamped sum preserves the multi-path convergence signal: nodes reachable
627                // via multiple paths receive proportionally higher activation (MAJOR-01).
628                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    /// Merge `next_activation` into `activation` and prune to `max_activated_nodes` (SA-INV-04).
644    ///
645    /// Returns the number of pruned nodes for tracing.
646    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    /// Append edges whose both endpoints are above threshold to `activated_facts`.
672    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    /// Collect nodes above threshold into `Vec<ActivatedNode>`, sorted descending by score.
700    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    /// Compute temporal recency weight for an edge.
715    ///
716    /// Formula: `1.0 / (1.0 + age_days * temporal_decay_rate)`.
717    /// Returns `1.0` when `temporal_decay_rate = 0.0` (no temporal adjustment).
718    /// Reuses the same formula as `GraphFact::score_with_decay` (SA-INV-05).
719    #[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        // cast f64 -> f32: safe, weight is in [0.0, 1.0]
731        #[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    // Test 1: empty graph (no edges) — seed entity is still returned as activated node,
764    // but no facts (edges) are found. Spread does not validate entity existence in DB.
765    #[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        // Seed node is returned as activated (activation=1.0, depth=0).
772        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        // No edges in empty graph, so no ActivatedFacts.
776        assert!(
777            facts.is_empty(),
778            "expected no activated facts on empty graph"
779        );
780    }
781
782    // Test 2: empty seeds returns empty
783    #[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    // Test 3: single seed with no edges returns only the seed
793    #[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    // Test 4: linear chain A->B->C with max_hops=3 — all activated, scores decay
812    #[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        // Scores must decay: score(A) > score(B) > score(C)
852        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    // Test 5: linear chain with max_hops=1 — C not activated
866    #[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    // Test 6: diamond graph — D receives convergent activation from two paths
906    // Graph: A -> B, A -> C, B -> D, C -> D
907    // With clamped sum, D gets activation from both paths (convergence signal preserved).
908    #[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; // raise inhibition to allow convergence
952        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        // D should be activated at depth 2
960        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    // Test 7: inhibition threshold prevents runaway activation in dense cluster
965    #[tokio::test]
966    async fn spread_inhibition_prevents_runaway() {
967        let store = setup_store().await;
968        // Create a hub node connected to many leaves
969        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            // Connect all leaves back to hub to create a dense cluster
1000            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        // Seed hub with full activation — it should be inhibited after hop 1
1015        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        // Hub should remain at initial activation (1.0), not grow unbounded
1023        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    // Test 8: max_activated_nodes cap — lowest activations pruned
1032    #[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        // Create 20 leaf nodes connected to root
1042        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    // Test 9: temporal decay — recent edges produce higher activation
1086    #[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        // Insert recent edge (default valid_from = now)
1106        store
1107            .insert_edge(src, recent, "uses", "Src uses Recent", 1.0, None, None)
1108            .await
1109            .unwrap();
1110
1111        // Insert old edge manually with a 1970 timestamp
1112        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        // Use significant temporal decay rate to distinguish recent vs old
1125        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    // Test 10: edge_type filtering — only edges of specified type are traversed
1148    #[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        // Semantic edge from A
1168        store
1169            .insert_edge(a, b_semantic, "uses", "A uses BSemantic", 1.0, None, None)
1170            .await
1171            .unwrap();
1172
1173        // Causal edge from A (inserted with explicit edge_type)
1174        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        // Spread with only semantic edges
1188        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    // Test 11: large seed list (stress test for batch query)
1206    #[tokio::test]
1207    async fn spread_large_seed_list() {
1208        let store = setup_store().await;
1209        let mut seeds = HashMap::new();
1210
1211        // Create 100 seed entities — tests that edges_for_entities handles chunking correctly
1212        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        // Should complete without error even with 100 seeds (chunking handles SQLite limit)
1230        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    // ── HL-F5 unit tests ─────────────────────────────────────────────────────
1239
1240    #[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        // Should not panic — denom is guarded by f32::EPSILON
1274        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    // Regression test for #4285: hela_spreading_recall must return
1300    // MemoryError::Timeout when the embed provider stalls beyond embed_timeout.
1301    #[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        // Provider sleeps 500 ms; timeout is set to 50 ms → must fire.
1314        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            &params,
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    // When embed_timeout is None the embed call is not wrapped; the (fast) mock
1347    // returns immediately and the function must succeed.
1348    #[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        // embed returns a zero-dimension vector (embed not configured for 384-dim),
1373        // so Qdrant search finds nothing — the function returns Ok(Vec::new()).
1374        let result = hela_spreading_recall(
1375            &store,
1376            &embeddings,
1377            &provider,
1378            "test query",
1379            5,
1380            &params,
1381            false,
1382            0.0,
1383        )
1384        .await;
1385
1386        // The mock embed returns an empty vec by default; the ANN search will
1387        // find no results — the expected outcome is Ok(empty) or a non-Timeout error.
1388        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        // path_weight × cosine.max(0.0): negative cosine must contribute 0.0
1408        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        // Two-hop path with edge weights 0.8, 0.5 → path_weight = 0.4
1423        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        // When two paths reach the same node, keep the higher path_weight.
1432        let pw_a = 0.9_f32; // short direct path
1433        let pw_b = 0.3_f32; // longer indirect path
1434        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        // Verify the formula used in hela_spreading_recall Step 4.
1447        assert!((expected - 0.6).abs() < 1e-5);
1448    }
1449
1450    /// SYNAPSE alpha-blend: blended score must equal `alpha*fast + (1-alpha)*slow`.
1451    ///
1452    /// With diverged fast=0.7, slow=0.51, alpha=0.3:
1453    ///   blended = 0.3*0.7 + 0.7*0.51 = 0.21 + 0.357 = 0.567
1454    #[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        // First insert — fast=slow=0.5.
1477        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        // Second insert — applies Benna-Fusi: fast≈0.7, slow≈0.51.
1491        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        // The blend formula is applied inside propagate_one_hop; the resulting edge weight
1510        // is evolved_weight(retrieval_count, blended). We verify that at least one fact is
1511        // returned with a score that is NOT equal to the raw confidence (0.9), confirming
1512        // the blend path was taken.
1513        assert!(!facts.is_empty(), "spread must return at least one fact");
1514        let raw_score = facts[0].activation_score;
1515        // blended = 0.3*0.7 + 0.7*0.51 ≈ 0.567 — different from raw confidence 0.9
1516        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}