zeph_memory/semantic/mod.rs
1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! High-level semantic memory orchestrator.
5//!
6//! [`SemanticMemory`] is the primary entry point used by `zeph-core`. It wires
7//! together [`crate::store::SqliteStore`] (relational persistence) and
8//! [`crate::embedding_store::EmbeddingStore`] (Qdrant vector index) into a single
9//! object with `remember` / `recall` / `summarize` operations.
10//!
11//! # Construction
12//!
13//! Use [`SemanticMemory::new`] for the default 0.7/0.3 vector/keyword weights, or
14//! [`SemanticMemory::with_qdrant_ops`] inside `AppBuilder` to share a single gRPC
15//! channel across all subsystems.
16//!
17//! # Hybrid recall
18//!
19//! Recall uses reciprocal-rank fusion of BM25 (`SQLite` FTS5) and cosine-similarity
20//! (Qdrant) results, with optional temporal decay, MMR diversity reranking, and
21//! per-tier score boosts.
22
23mod algorithms;
24mod corrections;
25mod cross_session;
26mod graph;
27pub(crate) mod importance;
28pub mod persona;
29mod recall;
30mod summarization;
31pub mod trajectory;
32pub mod tree_consolidation;
33
34#[cfg(test)]
35mod tests;
36
37use std::sync::Arc;
38use std::sync::Mutex;
39use std::sync::atomic::AtomicU64;
40use std::time::Instant;
41
42use tokio::sync::RwLock;
43use zeph_llm::any::AnyProvider;
44use zeph_llm::provider::LlmProvider as _;
45
46use crate::admission::AdmissionControl;
47use crate::embedding_store::EmbeddingStore;
48use crate::error::MemoryError;
49use crate::retrieval_failure_logger::RetrievalFailureLogger;
50use crate::store::SqliteStore;
51use crate::store::retrieval_failures::RetrievalFailureRecord;
52use crate::token_counter::TokenCounter;
53
54pub(crate) const SESSION_SUMMARIES_COLLECTION: &str = "zeph_session_summaries";
55pub(crate) const KEY_FACTS_COLLECTION: &str = "zeph_key_facts";
56pub(crate) const CORRECTIONS_COLLECTION: &str = "zeph_corrections";
57
58/// Progress state for embed backfill.
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub struct BackfillProgress {
61 /// Number of messages processed so far (including failures).
62 pub done: usize,
63 /// Total number of unembedded messages at backfill start.
64 pub total: usize,
65}
66
67pub use algorithms::{apply_mmr, apply_temporal_decay};
68pub use cross_session::SessionSummaryResult;
69pub use graph::{
70 ExtractionResult, ExtractionStats, GraphExtractionConfig, IngestBatchConfig, LinkingStats,
71 NoteLinkingConfig, PostExtractValidator, SharedPostExtractValidator, extract_and_store,
72 link_memory_notes,
73};
74pub use persona::{
75 PersonaExtractionConfig, contains_self_referential_language, extract_persona_facts,
76};
77pub use recall::{EmbedContext, RecalledMessage};
78pub use summarization::{StructuredSummary, SummarizeOutcome, Summary, build_summarization_prompt};
79pub use trajectory::{TrajectoryEntry, TrajectoryExtractionConfig, extract_trajectory_entries};
80pub use tree_consolidation::{
81 TreeConsolidationConfig, TreeConsolidationResult, run_tree_consolidation_sweep,
82 start_tree_consolidation_loop,
83};
84
85/// Cached profile centroid for query-bias correction (MM-F3, #3341).
86///
87/// Stored inside `SemanticMemory::profile_centroid` under an `RwLock`. Expires after
88/// `profile_centroid_ttl_secs` seconds; a miss is non-sticky (next call retries).
89#[derive(Debug, Clone)]
90pub(crate) struct CachedCentroid {
91 /// The centroid vector (unweighted mean of persona-fact embeddings).
92 pub vector: Vec<f32>,
93 /// Wall-clock instant when this centroid was computed.
94 pub computed_at: Instant,
95}
96
97/// Whether temporal decay is applied to recall scores.
98///
99/// When `Enabled`, older memories receive lower scores based on the configured
100/// half-life. When `Disabled`, all memories are scored equally regardless of age.
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
102#[non_exhaustive]
103pub enum TemporalDecay {
104 /// Apply exponential decay: older memories score lower.
105 Enabled,
106 /// No age-based score reduction.
107 #[default]
108 Disabled,
109}
110
111impl TemporalDecay {
112 /// Returns `true` when the variant is `Enabled`.
113 #[must_use]
114 #[inline]
115 pub fn is_enabled(self) -> bool {
116 self == Self::Enabled
117 }
118}
119
120impl From<bool> for TemporalDecay {
121 fn from(b: bool) -> Self {
122 if b { Self::Enabled } else { Self::Disabled }
123 }
124}
125
126/// Whether Maximal Marginal Relevance (MMR) diversity re-ranking is applied.
127///
128/// When `Enabled`, recall results are re-ranked to balance relevance and
129/// diversity using the configured lambda parameter.
130#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
131#[non_exhaustive]
132pub enum MmrReranking {
133 /// Apply MMR diversity re-ranking after initial vector search.
134 Enabled,
135 /// Return results in raw cosine-similarity order.
136 #[default]
137 Disabled,
138}
139
140impl MmrReranking {
141 /// Returns `true` when the variant is `Enabled`.
142 #[must_use]
143 #[inline]
144 pub fn is_enabled(self) -> bool {
145 self == Self::Enabled
146 }
147}
148
149impl From<bool> for MmrReranking {
150 fn from(b: bool) -> Self {
151 if b { Self::Enabled } else { Self::Disabled }
152 }
153}
154
155/// Whether write-time importance scoring influences recall ranking.
156///
157/// When `Enabled`, each stored message receives an importance score that
158/// is blended into the recall ranking with the configured weight.
159#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
160#[non_exhaustive]
161pub enum ImportanceScoring {
162 /// Blend importance scores into recall ranking.
163 Enabled,
164 /// Recall ranking uses only hybrid search scores.
165 #[default]
166 Disabled,
167}
168
169impl ImportanceScoring {
170 /// Returns `true` when the variant is `Enabled`.
171 #[must_use]
172 #[inline]
173 pub fn is_enabled(self) -> bool {
174 self == Self::Enabled
175 }
176}
177
178impl From<bool> for ImportanceScoring {
179 fn from(b: bool) -> Self {
180 if b { Self::Enabled } else { Self::Disabled }
181 }
182}
183
184/// Whether query-bias correction shifts first-person queries toward the user profile centroid.
185///
186/// When `Enabled`, queries containing first-person language are biased towards
187/// the stored user profile centroid to improve personalised recall.
188#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
189#[non_exhaustive]
190pub enum QueryBiasCorrection {
191 /// Shift first-person query embeddings towards the user profile centroid.
192 #[default]
193 Enabled,
194 /// Pass query embeddings through unchanged.
195 Disabled,
196}
197
198impl QueryBiasCorrection {
199 /// Returns `true` when the variant is `Enabled`.
200 #[must_use]
201 #[inline]
202 pub fn is_enabled(self) -> bool {
203 self == Self::Enabled
204 }
205}
206
207impl From<bool> for QueryBiasCorrection {
208 fn from(b: bool) -> Self {
209 if b { Self::Enabled } else { Self::Disabled }
210 }
211}
212
213/// Whether Hebbian edge-weight reinforcement is active.
214///
215/// When `Enabled`, each graph edge traversed during recall receives a small
216/// weight increment (`hebbian_lr`), strengthening frequently-used associations.
217#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
218#[non_exhaustive]
219pub enum HebbianReinforcement {
220 /// Increment edge weights after each recall traversal.
221 Enabled,
222 /// Edge weights remain unchanged after recall.
223 #[default]
224 Disabled,
225}
226
227impl HebbianReinforcement {
228 /// Returns `true` when the variant is `Enabled`.
229 #[must_use]
230 #[inline]
231 pub fn is_enabled(self) -> bool {
232 self == Self::Enabled
233 }
234}
235
236impl From<bool> for HebbianReinforcement {
237 fn from(b: bool) -> Self {
238 if b { Self::Enabled } else { Self::Disabled }
239 }
240}
241
242/// Classification of a user query's self-referential intent (MM-F3, #3341).
243///
244/// Used to decide whether query-bias correction should shift the embedding
245/// towards the user's profile centroid.
246#[derive(Debug, Clone, Copy, PartialEq, Eq)]
247pub(crate) enum QueryIntent {
248 /// Query contains first-person language — likely about the user themselves.
249 FirstPerson,
250 /// Query is about an external topic; no bias shift applied.
251 Other,
252}
253
254/// HL-F5 runtime wiring for spreading activation (mirror of `[memory.hebbian]` spread fields).
255///
256/// Built from config at bootstrap and attached via [`SemanticMemory::with_hebbian_spread`].
257#[derive(Debug, Clone)]
258pub struct HelaSpreadRuntime {
259 /// `true` when `[memory.hebbian] enabled = true` AND `spreading_activation = true`.
260 pub enabled: bool,
261 /// BFS hops, already clamped to `[1, 6]` by the caller.
262 pub depth: u32,
263 /// Soft upper bound on the visited-node set.
264 pub max_visited: usize,
265 /// MAGMA edge-type filter for BFS traversal.
266 pub edge_types: Vec<crate::graph::EdgeType>,
267 /// Per-step circuit-breaker duration.
268 pub step_budget: Option<std::time::Duration>,
269 /// Timeout for the initial query embedding call. `None` = no timeout.
270 pub embed_timeout: Option<std::time::Duration>,
271}
272
273impl Default for HelaSpreadRuntime {
274 fn default() -> Self {
275 Self {
276 enabled: false,
277 depth: 2,
278 max_visited: 200,
279 edge_types: Vec::new(),
280 step_budget: Some(std::time::Duration::from_millis(8)),
281 embed_timeout: Some(std::time::Duration::from_secs(5)),
282 }
283 }
284}
285
286/// High-level semantic memory orchestrator combining `SQLite` and Qdrant.
287///
288/// Instantiate via [`SemanticMemory::new`] or the `AppBuilder` integration.
289/// All fields are `pub(crate)` — callers interact through the inherent method API.
290pub struct SemanticMemory {
291 pub(crate) sqlite: SqliteStore,
292 pub(crate) qdrant: Option<Arc<EmbeddingStore>>,
293 pub(crate) provider: AnyProvider,
294 /// Dedicated provider for batch embedding calls (backfill, write-path embedding).
295 ///
296 /// When `Some`, all embedding I/O is routed through this provider instead of `provider`.
297 /// This prevents `embed_backfill` from saturating the main provider and causing guardrail
298 /// timeouts. When `None`, falls back to `provider`.
299 pub(crate) embed_provider: Option<AnyProvider>,
300 pub(crate) embedding_model: String,
301 pub(crate) vector_weight: f64,
302 pub(crate) keyword_weight: f64,
303 pub(crate) temporal_decay: TemporalDecay,
304 pub(crate) temporal_decay_half_life_days: u32,
305 pub(crate) mmr_reranking: MmrReranking,
306 pub(crate) mmr_lambda: f32,
307 pub(crate) importance_scoring: ImportanceScoring,
308 pub(crate) importance_weight: f64,
309 /// Multiplicative score boost for semantic-tier messages in recall ranking.
310 /// Default: `1.3`. Disabled when set to `1.0`.
311 pub(crate) tier_boost_semantic: f64,
312 pub token_counter: Arc<TokenCounter>,
313 pub graph_store: Option<Arc<crate::graph::GraphStore>>,
314 /// Experience store for tool-outcome telemetry and per-turn evolution sweeps.
315 ///
316 /// `Some` when `memory.graph.experience.enabled = true` at bootstrap.
317 pub experience: Option<Arc<crate::graph::experience::ExperienceStore>>,
318 /// `ReasoningBank` store for distilled reasoning strategies (#3342).
319 ///
320 /// `Some` when `memory.reasoning.enabled = true` at bootstrap.
321 pub reasoning: Option<Arc<crate::reasoning::ReasoningMemory>>,
322 pub(crate) community_detection_failures: Arc<AtomicU64>,
323 pub(crate) graph_extraction_count: Arc<AtomicU64>,
324 pub(crate) graph_extraction_failures: Arc<AtomicU64>,
325 pub(crate) last_qdrant_warn: Arc<AtomicU64>,
326 /// A-MAC admission control gate. When `Some`, each `remember()` call is evaluated.
327 pub(crate) admission_control: Option<Arc<AdmissionControl>>,
328 /// Write quality gate. When `Some`, evaluated in `remember()`/`remember_with_parts()`
329 /// after A-MAC admission and before persistence.
330 pub(crate) quality_gate: Option<Arc<crate::quality_gate::QualityGate>>,
331 /// Cosine similarity threshold for skipping near-duplicate key facts (0.0–1.0).
332 /// When a new fact's nearest neighbour in `zeph_key_facts` has score >= this value,
333 /// the fact is considered a duplicate and not inserted. Default: `0.95`.
334 pub(crate) key_facts_dedup_threshold: f32,
335 /// Bounded set of in-flight background embed tasks.
336 ///
337 /// Guarded by a `Mutex` because `SemanticMemory` is shared via `Arc` and
338 /// `JoinSet` requires `&mut self` for `spawn`. Capacity is capped at
339 /// `MAX_EMBED_BG_TASKS`; tasks that exceed the limit are dropped with a debug log.
340 pub(crate) embed_tasks: Mutex<tokio::task::JoinSet<()>>,
341 /// ANN candidate count fetched from the vector store before reranking (MM-F1, #3340).
342 ///
343 /// `0` = legacy behavior (`recall_limit * 2`). `≥ 1` = direct count.
344 pub(crate) retrieval_depth: u32,
345 /// Template applied to raw user queries before embedding (MM-F2, #3340).
346 ///
347 /// Empty string = identity (pass raw query through). Applied at query-side embed sites only;
348 /// never applied to stored content (summaries, documents).
349 pub(crate) search_prompt_template: String,
350 /// Fires `tracing::warn!` once per instance when `retrieval_depth < recall_limit`.
351 pub(crate) depth_below_limit_warned: Arc<std::sync::atomic::AtomicBool>,
352 /// Fires `tracing::warn!` once per instance when `search_prompt_template` has no `{query}`.
353 pub(crate) missing_placeholder_warned: Arc<std::sync::atomic::AtomicBool>,
354 /// Query-bias correction towards the user profile centroid (MM-F3, #3341).
355 pub(crate) query_bias_correction: QueryBiasCorrection,
356 /// Blend weight for query-bias correction (MM-F3, #3341). Clamped to `[0.0, 1.0]`.
357 pub(crate) query_bias_profile_weight: f32,
358 /// Cached profile centroid computed from persona-fact embeddings (MM-F3, #3341).
359 ///
360 /// Protected by `RwLock` to allow concurrent reads. Never holds the lock across `.await`
361 /// (await-discipline rule #4). TTL-bounded; miss is non-sticky.
362 pub(crate) profile_centroid: RwLock<Option<CachedCentroid>>,
363 /// Time-to-live for the profile centroid cache in seconds (MM-F3, #3341). Default: 300.
364 pub(crate) profile_centroid_ttl_secs: u64,
365 /// Opt-in master switch for Hebbian edge-weight reinforcement (HL-F2, #3344).
366 pub(crate) hebbian_reinforcement: HebbianReinforcement,
367 /// Weight increment applied per recall traversal when `hebbian_reinforcement` is `Enabled` (HL-F2, #3344).
368 pub(crate) hebbian_lr: f32,
369 /// HL-F5 spreading activation runtime config (#3346).
370 pub(crate) hebbian_spread: HelaSpreadRuntime,
371 /// `OmniMem` retrieval failure logger (issue #3576).
372 ///
373 /// `Some` when `memory.retrieval_failures.enabled = true` at bootstrap.
374 pub(crate) retrieval_failure_logger: Option<RetrievalFailureLogger>,
375 /// LLM call timeout for summarization, in seconds. Default: `60`.
376 pub(crate) summarization_llm_timeout_secs: u64,
377 /// PRISM: enable query-sensitive edge costing in A* graph recall.
378 ///
379 /// When `true`, A* edge cost is modulated by cosine similarity between the query
380 /// embedding and the target entity embedding. Mirrors [`GraphConfig::query_sensitive_cost`].
381 pub(crate) query_sensitive_cost: bool,
382 /// Five-signal SYNAPSE retrieval runtime (issue #4374).
383 ///
384 /// `Some` when `memory.five_signal.enabled = true` at bootstrap.
385 /// `None` guarantees zero overhead per NFR-005.
386 pub(crate) five_signal: Option<Arc<crate::five_signal::FiveSignalRuntime>>,
387 /// Per-call timeout applied to every `embed()` invocation in this instance.
388 ///
389 /// Configurable via `[memory.semantic] embed_timeout_secs`. Default: 5 s.
390 pub(crate) embed_timeout: std::time::Duration,
391 /// Cancellation tokens for all currently in-flight background graph-extraction tasks.
392 ///
393 /// Every call to [`SemanticMemory::spawn_graph_extraction`] pushes its token onto this
394 /// vec instead of overwriting a single slot, so tasks spawned before a previous one
395 /// finished remain reachable. Already-cancelled tokens are pruned opportunistically on
396 /// the next spawn. Call [`SemanticMemory::cancel_graph_extraction`] to signal
397 /// cooperative cancellation to every tracked task before hard-aborting via the
398 /// supervisor.
399 pub(crate) graph_cancel: Mutex<Vec<tokio_util::sync::CancellationToken>>,
400}
401
402impl SemanticMemory {
403 /// Build a `SemanticMemory` with every field set to its default value, varying only
404 /// the handful of parameters that differ across the public constructors.
405 ///
406 /// Shared by [`SemanticMemory::with_weights_and_pool_size`], [`SemanticMemory::with_qdrant_ops`],
407 /// [`SemanticMemory::from_parts`] and [`SemanticMemory::with_sqlite_backend_and_pool_size`]
408 /// to avoid duplicating the full field literal in each.
409 fn base(
410 sqlite: SqliteStore,
411 qdrant: Option<Arc<EmbeddingStore>>,
412 provider: AnyProvider,
413 embedding_model: String,
414 vector_weight: f64,
415 keyword_weight: f64,
416 token_counter: Arc<TokenCounter>,
417 ) -> Self {
418 Self {
419 sqlite,
420 qdrant,
421 provider,
422 embed_provider: None,
423 embedding_model,
424 vector_weight,
425 keyword_weight,
426 temporal_decay: TemporalDecay::Disabled,
427 temporal_decay_half_life_days: 30,
428 mmr_reranking: MmrReranking::Disabled,
429 mmr_lambda: 0.7,
430 importance_scoring: ImportanceScoring::Disabled,
431 importance_weight: 0.15,
432 tier_boost_semantic: 1.3,
433 token_counter,
434 graph_store: None,
435 experience: None,
436 reasoning: None,
437 community_detection_failures: Arc::new(AtomicU64::new(0)),
438 graph_extraction_count: Arc::new(AtomicU64::new(0)),
439 graph_extraction_failures: Arc::new(AtomicU64::new(0)),
440 last_qdrant_warn: Arc::new(AtomicU64::new(0)),
441 admission_control: None,
442 quality_gate: None,
443 key_facts_dedup_threshold: 0.95,
444 embed_tasks: std::sync::Mutex::new(tokio::task::JoinSet::new()),
445 retrieval_depth: 0,
446 search_prompt_template: String::new(),
447 depth_below_limit_warned: Arc::new(std::sync::atomic::AtomicBool::new(false)),
448 missing_placeholder_warned: Arc::new(std::sync::atomic::AtomicBool::new(false)),
449 query_bias_correction: QueryBiasCorrection::Enabled,
450 query_bias_profile_weight: 0.25,
451 profile_centroid: RwLock::new(None),
452 profile_centroid_ttl_secs: 300,
453 hebbian_reinforcement: HebbianReinforcement::Disabled,
454 hebbian_lr: 0.1,
455 hebbian_spread: HelaSpreadRuntime::default(),
456 retrieval_failure_logger: None,
457 summarization_llm_timeout_secs: 60,
458 query_sensitive_cost: false,
459 five_signal: None,
460 embed_timeout: std::time::Duration::from_secs(5),
461 graph_cancel: Mutex::new(Vec::new()),
462 }
463 }
464
465 /// Create a new `SemanticMemory` instance with default hybrid search weights (0.7/0.3).
466 ///
467 /// Qdrant connection is best-effort: if unavailable, semantic search is disabled.
468 ///
469 /// For `AppBuilder` bootstrap, prefer [`SemanticMemory::with_qdrant_ops`] to share
470 /// a single gRPC channel across all subsystems.
471 ///
472 /// # Errors
473 ///
474 /// Returns an error if `SQLite` cannot be initialized.
475 pub async fn new(
476 sqlite_path: &str,
477 qdrant_url: &str,
478 api_key: Option<&str>,
479 provider: AnyProvider,
480 embedding_model: &str,
481 ) -> Result<Self, MemoryError> {
482 Self::with_weights(
483 sqlite_path,
484 qdrant_url,
485 api_key,
486 provider,
487 embedding_model,
488 0.7,
489 0.3,
490 )
491 .await
492 }
493
494 /// Create a new `SemanticMemory` with custom vector/keyword weights for hybrid search.
495 ///
496 /// For `AppBuilder` bootstrap, prefer [`SemanticMemory::with_qdrant_ops`] to share
497 /// a single gRPC channel across all subsystems.
498 ///
499 /// # Errors
500 ///
501 /// Returns an error if `SQLite` cannot be initialized.
502 pub async fn with_weights(
503 sqlite_path: &str,
504 qdrant_url: &str,
505 api_key: Option<&str>,
506 provider: AnyProvider,
507 embedding_model: &str,
508 vector_weight: f64,
509 keyword_weight: f64,
510 ) -> Result<Self, MemoryError> {
511 Self::with_weights_and_pool_size(
512 sqlite_path,
513 qdrant_url,
514 api_key,
515 provider,
516 embedding_model,
517 vector_weight,
518 keyword_weight,
519 5,
520 )
521 .await
522 }
523
524 /// Create a new `SemanticMemory` with custom weights and configurable pool size.
525 ///
526 /// For `AppBuilder` bootstrap, prefer [`SemanticMemory::with_qdrant_ops`] to share
527 /// a single gRPC channel across all subsystems.
528 ///
529 /// # Errors
530 ///
531 /// Returns an error if `SQLite` cannot be initialized.
532 #[allow(clippy::too_many_arguments)]
533 pub async fn with_weights_and_pool_size(
534 sqlite_path: &str,
535 qdrant_url: &str,
536 api_key: Option<&str>,
537 provider: AnyProvider,
538 embedding_model: &str,
539 vector_weight: f64,
540 keyword_weight: f64,
541 pool_size: u32,
542 ) -> Result<Self, MemoryError> {
543 let sqlite = SqliteStore::with_pool_size(sqlite_path, pool_size).await?;
544 let db_instance_id = sqlite.db_instance_id().to_owned();
545 let pool = sqlite.pool().clone();
546
547 let qdrant = match EmbeddingStore::new(qdrant_url, api_key, pool) {
548 Ok(store) => Some(Arc::new(store.with_db_instance_id(db_instance_id))),
549 Err(e) => {
550 tracing::warn!("Qdrant unavailable, semantic search disabled: {e:#}");
551 None
552 }
553 };
554
555 Ok(Self::base(
556 sqlite,
557 qdrant,
558 provider,
559 embedding_model.into(),
560 vector_weight,
561 keyword_weight,
562 Arc::new(TokenCounter::new()),
563 ))
564 }
565
566 /// Create a `SemanticMemory` from a pre-built `QdrantOps` instance.
567 ///
568 /// Use this at bootstrap to share one `QdrantOps` (and thus one gRPC channel)
569 /// across all subsystems. The `ops` is consumed and wrapped inside `EmbeddingStore`.
570 ///
571 /// # Errors
572 ///
573 /// Returns an error if `SQLite` cannot be initialized.
574 pub async fn with_qdrant_ops(
575 sqlite_path: &str,
576 ops: crate::QdrantOps,
577 provider: AnyProvider,
578 embedding_model: &str,
579 vector_weight: f64,
580 keyword_weight: f64,
581 pool_size: u32,
582 ) -> Result<Self, MemoryError> {
583 let sqlite = SqliteStore::with_pool_size(sqlite_path, pool_size).await?;
584 let pool = sqlite.pool().clone();
585 let store = EmbeddingStore::with_store(Box::new(ops), pool)
586 .with_db_instance_id(sqlite.db_instance_id());
587
588 Ok(Self::base(
589 sqlite,
590 Some(Arc::new(store)),
591 provider,
592 embedding_model.into(),
593 vector_weight,
594 keyword_weight,
595 Arc::new(TokenCounter::new()),
596 ))
597 }
598
599 /// Attach a `GraphStore` for graph-aware retrieval.
600 ///
601 /// When set, `recall_graph` traverses the graph starting from entities
602 /// matched by the query.
603 #[must_use]
604 pub fn with_graph_store(mut self, store: Arc<crate::graph::GraphStore>) -> Self {
605 self.graph_store = Some(store);
606 self
607 }
608
609 /// Attach an [`ExperienceStore`](crate::graph::experience::ExperienceStore) for tool-outcome
610 /// telemetry and per-turn evolution sweeps.
611 ///
612 /// When set, the agent records one row per tool invocation in `experience_nodes` and
613 /// periodically runs `evolution_sweep` to prune low-confidence and self-loop edges.
614 #[must_use]
615 pub fn with_experience_store(
616 mut self,
617 store: Arc<crate::graph::experience::ExperienceStore>,
618 ) -> Self {
619 self.experience = Some(store);
620 self
621 }
622
623 /// Attach a [`ReasoningMemory`](crate::reasoning::ReasoningMemory) store for
624 /// distilled reasoning strategy storage and retrieval (#3342).
625 ///
626 /// When set, [`SemanticMemory::retrieve_reasoning_strategies`] uses this store for
627 /// embedding-similarity lookups. When `None`, retrieval returns an empty vec.
628 #[must_use]
629 pub fn with_reasoning(mut self, store: Arc<crate::reasoning::ReasoningMemory>) -> Self {
630 self.reasoning = Some(store);
631 self
632 }
633
634 /// Attach a [`RetrievalFailureLogger`] for `OmniMem` self-improvement data collection.
635 ///
636 /// When attached, [`SemanticMemory::log_retrieval_failure`] records events
637 /// asynchronously. When absent, `log_retrieval_failure` is a no-op.
638 #[must_use]
639 pub fn with_retrieval_failure_logger(mut self, logger: RetrievalFailureLogger) -> Self {
640 self.retrieval_failure_logger = Some(logger);
641 self
642 }
643
644 /// Log a retrieval failure event asynchronously.
645 ///
646 /// No-op when retrieval failure logging is disabled (`retrieval_failure_logger` is `None`).
647 /// On the hot path this method never blocks — records are sent via a bounded mpsc channel
648 /// and dropped silently when the channel is full.
649 pub fn log_retrieval_failure(&self, record: RetrievalFailureRecord) {
650 if let Some(logger) = &self.retrieval_failure_logger {
651 logger.log(record);
652 }
653 }
654
655 /// Returns the cumulative count of community detection failures since startup.
656 #[must_use]
657 pub fn community_detection_failures(&self) -> u64 {
658 use std::sync::atomic::Ordering;
659 self.community_detection_failures.load(Ordering::Relaxed)
660 }
661
662 /// Returns the cumulative count of successful graph extractions since startup.
663 #[must_use]
664 pub fn graph_extraction_count(&self) -> u64 {
665 use std::sync::atomic::Ordering;
666 self.graph_extraction_count.load(Ordering::Relaxed)
667 }
668
669 /// Returns the cumulative count of failed graph extractions since startup.
670 #[must_use]
671 pub fn graph_extraction_failures(&self) -> u64 {
672 use std::sync::atomic::Ordering;
673 self.graph_extraction_failures.load(Ordering::Relaxed)
674 }
675
676 /// Configure temporal decay and MMR re-ranking options.
677 #[must_use]
678 pub fn with_ranking_options(
679 mut self,
680 temporal_decay: TemporalDecay,
681 temporal_decay_half_life_days: u32,
682 mmr_reranking: MmrReranking,
683 mmr_lambda: f32,
684 ) -> Self {
685 self.temporal_decay = temporal_decay;
686 self.temporal_decay_half_life_days = temporal_decay_half_life_days;
687 self.mmr_reranking = mmr_reranking;
688 self.mmr_lambda = mmr_lambda;
689 self
690 }
691
692 /// Configure write-time importance scoring for memory retrieval.
693 #[must_use]
694 pub fn with_importance_options(mut self, scoring: ImportanceScoring, weight: f64) -> Self {
695 self.importance_scoring = scoring;
696 self.importance_weight = weight;
697 self
698 }
699
700 /// Configure the multiplicative score boost applied to semantic-tier messages during recall.
701 ///
702 /// Set to `1.0` to disable the boost. Default: `1.3`.
703 #[must_use]
704 pub fn with_tier_boost(mut self, boost: f64) -> Self {
705 self.tier_boost_semantic = boost;
706 self
707 }
708
709 /// Attach an A-MAC admission controller.
710 ///
711 /// When set, `remember()` and `remember_with_parts()` evaluate each message before persisting.
712 /// Messages below the admission threshold return `Ok(None)` without incrementing counts.
713 #[must_use]
714 pub fn with_admission_control(mut self, control: AdmissionControl) -> Self {
715 self.admission_control = Some(Arc::new(control));
716 self
717 }
718
719 /// Attach a write quality gate that scores each `remember()` call before persisting.
720 ///
721 /// When set, the gate is evaluated after A-MAC admission. A `Some(reason)` result from
722 /// [`crate::quality_gate::QualityGate::evaluate`] causes the write to be skipped
723 /// and `Ok(None)` / `Ok((None, false))` to be returned.
724 #[must_use]
725 pub fn with_quality_gate(mut self, gate: Arc<crate::quality_gate::QualityGate>) -> Self {
726 self.quality_gate = Some(gate);
727 self
728 }
729
730 /// Set the cosine similarity threshold used to skip near-duplicate key facts on insert.
731 ///
732 /// When a candidate fact's nearest neighbour in `zeph_key_facts` has a score ≥ this value,
733 /// the fact is not stored. Default: `0.95`.
734 #[must_use]
735 pub fn with_key_facts_dedup_threshold(mut self, threshold: f32) -> Self {
736 self.key_facts_dedup_threshold = threshold;
737 self
738 }
739
740 /// Set the LLM call timeout for summarization, in seconds.
741 ///
742 /// Applies to both structured and plain-text fallback summarization calls.
743 #[must_use]
744 pub fn with_summarization_timeout(mut self, timeout_secs: u64) -> Self {
745 self.summarization_llm_timeout_secs = timeout_secs;
746 self
747 }
748
749 /// Set the per-call timeout for every `embed()` invocation inside this instance.
750 ///
751 /// Configures the timeout applied at all embedding call sites: admission control,
752 /// quality gate, recall, summarization, graph retrieval, consolidation, and tree
753 /// consolidation. Must be non-zero; the minimum effective value is 1 s. Default: `5`.
754 #[must_use]
755 pub fn with_embed_timeout(mut self, timeout_secs: u64) -> Self {
756 let t = std::time::Duration::from_secs(timeout_secs.max(1));
757 self.embed_timeout = t;
758 self.hebbian_spread.embed_timeout = Some(t);
759 self
760 }
761
762 /// Configure query-bias correction (MM-F3, #3341).
763 ///
764 /// When `correction` is [`QueryBiasCorrection::Enabled`], first-person queries are biased
765 /// towards the user profile centroid. `profile_weight` controls the blend strength and is
766 /// clamped to `[0.0, 1.0]`. `centroid_ttl_secs` controls how long the centroid cache stays
767 /// valid.
768 #[must_use]
769 pub fn with_query_bias(
770 mut self,
771 correction: QueryBiasCorrection,
772 profile_weight: f32,
773 centroid_ttl_secs: u64,
774 ) -> Self {
775 self.query_bias_correction = correction;
776 self.query_bias_profile_weight = profile_weight.clamp(0.0, 1.0);
777 self.profile_centroid_ttl_secs = centroid_ttl_secs;
778 self
779 }
780
781 /// Configure HL-F5 spreading activation runtime parameters (HL-F5, #3346).
782 ///
783 /// Has no effect when `hebbian_spread.enabled = false` (the default).
784 /// Call this after `with_graph_store` and `with_hebbian` during bootstrap.
785 #[must_use]
786 pub fn with_hebbian_spread(mut self, runtime: HelaSpreadRuntime) -> Self {
787 self.hebbian_spread = runtime;
788 self
789 }
790
791 /// Configure Hebbian edge-weight reinforcement (HL-F2, #3344).
792 ///
793 /// When `reinforcement` is [`HebbianReinforcement::Enabled`], `lr` is added to the `weight`
794 /// column of each traversed edge after every recall. `lr = 0.0` with `Enabled` logs a WARN.
795 #[must_use]
796 pub fn with_hebbian(mut self, reinforcement: HebbianReinforcement, lr: f32) -> Self {
797 let lr = lr.max(0.0);
798 if reinforcement.is_enabled() && lr == 0.0 {
799 tracing::warn!("hebbian enabled with lr=0.0 — no reinforcement will occur");
800 }
801 self.hebbian_reinforcement = reinforcement;
802 self.hebbian_lr = lr;
803 self
804 }
805
806 /// Enable PRISM query-sensitive edge costing in A* graph recall (#4079).
807 ///
808 /// When enabled, edge cost is modulated by cosine similarity between the query embedding
809 /// and the target entity embedding, guiding A* toward semantically relevant paths.
810 #[must_use]
811 pub fn with_query_sensitive_cost(mut self, enabled: bool) -> Self {
812 self.query_sensitive_cost = enabled;
813 self
814 }
815
816 /// Attach the five-signal retrieval runtime (issue #4374).
817 ///
818 /// When attached, `recall_merge_and_rank` applies five-signal scoring after the
819 /// existing pipeline, gated by `!weights.is_baseline()`.
820 ///
821 /// # Examples
822 ///
823 /// ```no_run
824 /// use std::sync::Arc;
825 /// use zeph_memory::semantic::SemanticMemory;
826 /// use zeph_memory::five_signal::FiveSignalRuntime;
827 ///
828 /// # async fn example(mem: SemanticMemory, runtime: FiveSignalRuntime) {
829 /// let mem = mem.with_five_signal(Arc::new(runtime));
830 /// # }
831 /// ```
832 #[must_use]
833 pub fn with_five_signal(mut self, runtime: Arc<crate::five_signal::FiveSignalRuntime>) -> Self {
834 self.five_signal = Some(runtime);
835 self
836 }
837
838 /// Return the five-signal runtime, if one was attached via [`Self::with_five_signal`].
839 #[must_use]
840 pub fn five_signal_runtime(&self) -> Option<Arc<crate::five_signal::FiveSignalRuntime>> {
841 self.five_signal.clone()
842 }
843
844 /// Classify a query's intent for query-bias correction (MM-F3, #3341).
845 ///
846 /// Returns [`QueryIntent::FirstPerson`] when the query contains self-referential language
847 /// (first-person pronouns). Otherwise returns [`QueryIntent::Other`].
848 pub(crate) fn classify_query_intent(query: &str) -> QueryIntent {
849 if persona::contains_self_referential_language(query) {
850 QueryIntent::FirstPerson
851 } else {
852 QueryIntent::Other
853 }
854 }
855
856 /// Apply query-bias correction to an embedding (MM-F3, #3341).
857 ///
858 /// Returns the embedding unchanged if `query_bias_correction` is [`QueryBiasCorrection::Disabled`],
859 /// if the query is not first-person, or if the profile centroid is unavailable.
860 /// Logs a single WARN on dimension mismatch and returns the original embedding.
861 #[tracing::instrument(name = "memory.query_bias.apply", skip(self, embedding), fields(query_len = query.len()))]
862 pub(crate) async fn apply_query_bias(&self, query: &str, embedding: Vec<f32>) -> Vec<f32> {
863 if !self.query_bias_correction.is_enabled() {
864 tracing::debug!(reason = "disabled", "query-bias: skipping");
865 return embedding;
866 }
867 if Self::classify_query_intent(query) != QueryIntent::FirstPerson {
868 tracing::debug!(reason = "not_first_person", "query-bias: skipping");
869 return embedding;
870 }
871 let Some(centroid) = self.profile_centroid_cached().await else {
872 tracing::debug!(reason = "no_centroid", "query-bias: skipping");
873 return embedding;
874 };
875 if centroid.len() != embedding.len() {
876 tracing::warn!(
877 centroid_dim = centroid.len(),
878 query_dim = embedding.len(),
879 reason = "dim_mismatch",
880 "query-bias: dimension mismatch between profile centroid and query embedding — skipping bias"
881 );
882 return embedding;
883 }
884 let w = self.query_bias_profile_weight;
885 tracing::debug!(
886 intent = "first_person",
887 centroid_dim = centroid.len(),
888 weight = w,
889 "query-bias: applying profile bias"
890 );
891 embedding
892 .iter()
893 .zip(centroid.iter())
894 .map(|(&q, &c)| (1.0 - w) * q + w * c)
895 .collect()
896 }
897
898 /// Return the cached profile centroid, recomputing if stale or absent (MM-F3, #3341).
899 ///
900 /// Holds the read lock only to check freshness; releases it before any `.await`.
901 /// On compute failure, preserves the previous cache value (non-sticky miss).
902 #[tracing::instrument(name = "memory.query_bias.centroid", skip(self))]
903 pub(crate) async fn profile_centroid_cached(&self) -> Option<Vec<f32>> {
904 // Fast path: check freshness under read lock without holding it across await.
905 {
906 let guard = self.profile_centroid.read().await;
907 if let Some(c) = &*guard
908 && c.computed_at.elapsed().as_secs() < self.profile_centroid_ttl_secs
909 {
910 let ttl_remaining = self
911 .profile_centroid_ttl_secs
912 .saturating_sub(c.computed_at.elapsed().as_secs());
913 tracing::debug!(
914 centroid_dim = c.vector.len(),
915 ttl_remaining_secs = ttl_remaining,
916 "query-bias: centroid cache hit"
917 );
918 return Some(c.vector.clone());
919 }
920 }
921 // Slow path: recompute. Guard is dropped before this point.
922 let computed = self.compute_profile_centroid().await;
923 let mut guard = self.profile_centroid.write().await;
924 match computed {
925 Some(v) => {
926 tracing::debug!(centroid_dim = v.len(), "query-bias: centroid computed");
927 *guard = Some(CachedCentroid {
928 vector: v.clone(),
929 computed_at: Instant::now(),
930 });
931 Some(v)
932 }
933 None => {
934 // Do not overwrite a valid (but stale) cache on failure — serve stale over nothing.
935 guard.as_ref().map(|c| c.vector.clone())
936 }
937 }
938 }
939
940 /// Compute the profile centroid from persona-fact embeddings (MM-F3, #3341).
941 ///
942 /// Returns `None` when the persona table is empty or embedding fails.
943 /// Uses `load_persona_facts(0.0)` (all non-superseded facts) for the centroid basis.
944 async fn compute_profile_centroid(&self) -> Option<Vec<f32>> {
945 let facts = match self.sqlite.load_persona_facts(0.0).await {
946 Ok(f) => f,
947 Err(e) => {
948 tracing::warn!(error = %e, "query-bias: failed to load persona facts");
949 return None;
950 }
951 };
952 if facts.is_empty() {
953 return None;
954 }
955 let provider = self.effective_embed_provider();
956 let texts: Vec<String> = facts.iter().map(|f| f.content.clone()).collect();
957 let mut embeddings: Vec<Vec<f32>> = Vec::with_capacity(texts.len());
958 for text in &texts {
959 match tokio::time::timeout(self.embed_timeout, provider.embed(text)).await {
960 Ok(Ok(v)) => embeddings.push(v),
961 Ok(Err(e)) => {
962 tracing::warn!(error = %e, "query-bias: failed to embed persona fact — skipping");
963 }
964 Err(_) => {
965 tracing::warn!("query-bias: embed timed out for persona fact — skipping");
966 }
967 }
968 }
969 if embeddings.is_empty() {
970 return None;
971 }
972 let dim = embeddings[0].len();
973 let mut centroid = vec![0.0f32; dim];
974 for emb in &embeddings {
975 if emb.len() != dim {
976 tracing::warn!(
977 expected = dim,
978 got = emb.len(),
979 "query-bias: persona embedding dimension mismatch — skipping fact"
980 );
981 continue;
982 }
983 for (c, &v) in centroid.iter_mut().zip(emb.iter()) {
984 *c += v;
985 }
986 }
987 #[allow(clippy::cast_precision_loss)]
988 let n = embeddings.len() as f32;
989 for c in &mut centroid {
990 *c /= n;
991 }
992 Some(centroid)
993 }
994
995 /// Configure retrieval depth and search prompt template (MM-F1/F2, #3340).
996 ///
997 /// `depth` is the number of ANN candidates fetched from the vector store before keyword merge
998 /// and MMR re-ranking. `0` = legacy behavior (`recall_limit * 2`). `≥ 1` = exact count.
999 ///
1000 /// `search_prompt_template` is applied to the raw user query before embedding. Supports a
1001 /// single `{query}` placeholder. Empty string = identity.
1002 #[must_use]
1003 pub fn with_retrieval_options(
1004 mut self,
1005 depth: u32,
1006 search_prompt_template: impl Into<String>,
1007 ) -> Self {
1008 self.retrieval_depth = depth;
1009 self.search_prompt_template = search_prompt_template.into();
1010 self
1011 }
1012
1013 /// Effective ANN candidate count for a given requested final limit (MM-F1, #3340).
1014 ///
1015 /// - `retrieval_depth == 0`: legacy behavior, returns `limit * 2`.
1016 /// - `retrieval_depth >= 1`: returns the configured depth directly.
1017 ///
1018 /// When `retrieval_depth < limit`, a one-shot WARN fires because the ANN pool cannot
1019 /// saturate the requested top-k. When `limit <= retrieval_depth < limit * 2`, an INFO
1020 /// fires per call noting the smaller-than-legacy pool.
1021 pub(crate) fn effective_depth(&self, limit: usize) -> usize {
1022 use std::sync::atomic::Ordering;
1023
1024 let depth = self.retrieval_depth as usize;
1025 if depth == 0 {
1026 return limit.saturating_mul(2);
1027 }
1028 if depth < limit {
1029 if !self.depth_below_limit_warned.swap(true, Ordering::Relaxed) {
1030 tracing::warn!(
1031 retrieval_depth = depth,
1032 recall_limit = limit,
1033 "memory.retrieval.depth < recall_limit; ANN pool cannot saturate top-k — consider raising depth"
1034 );
1035 }
1036 } else if depth < limit.saturating_mul(2) {
1037 tracing::info!(
1038 retrieval_depth = depth,
1039 recall_limit = limit,
1040 legacy_default = limit.saturating_mul(2),
1041 "memory.retrieval.depth is below legacy limit*2; ANN pool will be smaller than pre-#3340"
1042 );
1043 } else {
1044 tracing::debug!(
1045 retrieval_depth = depth,
1046 recall_limit = limit,
1047 "recall: using configured ANN depth"
1048 );
1049 }
1050 depth
1051 }
1052
1053 /// Apply the configured search prompt template to a raw query (MM-F2, #3340).
1054 ///
1055 /// Returns `query` as-is when the template is empty or has no `{query}` placeholder.
1056 /// A one-shot WARN fires when the template is non-empty but missing the placeholder.
1057 pub(crate) fn apply_search_prompt(&self, query: &str) -> String {
1058 use std::sync::atomic::Ordering;
1059
1060 let template = &self.search_prompt_template;
1061 if template.is_empty() {
1062 return query.to_owned();
1063 }
1064 if !template.contains("{query}") {
1065 if !self
1066 .missing_placeholder_warned
1067 .swap(true, Ordering::Relaxed)
1068 {
1069 tracing::warn!(
1070 template = template.as_str(),
1071 "memory.retrieval.search_prompt_template has no {{query}} placeholder — \
1072 using raw query as-is"
1073 );
1074 }
1075 return query.to_owned();
1076 }
1077 template.replace("{query}", query)
1078 }
1079
1080 /// Attach a dedicated embedding provider for write-path and backfill operations.
1081 ///
1082 /// When set, all batch embedding calls (backfill, `remember`) route through this provider
1083 /// instead of the main `provider`. This prevents `embed_backfill` from saturating the main
1084 /// provider and causing guardrail timeouts due to rate-limit contention or Ollama model-lock.
1085 #[must_use]
1086 pub fn with_embedding_provider(mut self, embed_provider: AnyProvider) -> Self {
1087 self.embed_provider = Some(embed_provider);
1088 self
1089 }
1090
1091 /// Returns the provider to use for embedding calls.
1092 ///
1093 /// Returns the dedicated embed provider when configured, falling back to the main provider.
1094 pub fn effective_embed_provider(&self) -> &AnyProvider {
1095 self.embed_provider.as_ref().unwrap_or(&self.provider)
1096 }
1097
1098 /// Construct a `SemanticMemory` from pre-built parts.
1099 ///
1100 /// Intended for tests that need full control over the backing stores.
1101 #[must_use]
1102 pub fn from_parts(
1103 sqlite: SqliteStore,
1104 qdrant: Option<Arc<EmbeddingStore>>,
1105 provider: AnyProvider,
1106 embedding_model: impl Into<String>,
1107 vector_weight: f64,
1108 keyword_weight: f64,
1109 token_counter: Arc<TokenCounter>,
1110 ) -> Self {
1111 Self::base(
1112 sqlite,
1113 qdrant,
1114 provider,
1115 embedding_model.into(),
1116 vector_weight,
1117 keyword_weight,
1118 token_counter,
1119 )
1120 }
1121
1122 /// Create a `SemanticMemory` using the `SQLite`-embedded vector backend.
1123 ///
1124 /// # Errors
1125 ///
1126 /// Returns an error if `SQLite` cannot be initialized.
1127 pub async fn with_sqlite_backend(
1128 sqlite_path: &str,
1129 provider: AnyProvider,
1130 embedding_model: &str,
1131 vector_weight: f64,
1132 keyword_weight: f64,
1133 ) -> Result<Self, MemoryError> {
1134 Self::with_sqlite_backend_and_pool_size(
1135 sqlite_path,
1136 provider,
1137 embedding_model,
1138 vector_weight,
1139 keyword_weight,
1140 5,
1141 )
1142 .await
1143 }
1144
1145 /// Create a `SemanticMemory` using the `SQLite`-embedded vector backend with configurable pool size.
1146 ///
1147 /// # Errors
1148 ///
1149 /// Returns an error if `SQLite` cannot be initialized.
1150 pub async fn with_sqlite_backend_and_pool_size(
1151 sqlite_path: &str,
1152 provider: AnyProvider,
1153 embedding_model: &str,
1154 vector_weight: f64,
1155 keyword_weight: f64,
1156 pool_size: u32,
1157 ) -> Result<Self, MemoryError> {
1158 let sqlite = SqliteStore::with_pool_size(sqlite_path, pool_size).await?;
1159 let pool = sqlite.pool().clone();
1160 let store = EmbeddingStore::new_sqlite(pool).with_db_instance_id(sqlite.db_instance_id());
1161
1162 Ok(Self::base(
1163 sqlite,
1164 Some(Arc::new(store)),
1165 provider,
1166 embedding_model.into(),
1167 vector_weight,
1168 keyword_weight,
1169 Arc::new(TokenCounter::new()),
1170 ))
1171 }
1172
1173 /// Access the underlying `SqliteStore` for operations that don't involve semantics.
1174 #[must_use]
1175 pub fn sqlite(&self) -> &SqliteStore {
1176 &self.sqlite
1177 }
1178
1179 /// Return the per-call embed timeout configured for this instance.
1180 #[must_use]
1181 pub fn embed_timeout(&self) -> std::time::Duration {
1182 self.embed_timeout
1183 }
1184
1185 /// Check if the vector store backend is reachable.
1186 ///
1187 /// Performs a real health check (Qdrant gRPC ping or `SQLite` query)
1188 /// instead of just checking whether the client was created.
1189 pub async fn is_vector_store_connected(&self) -> bool {
1190 match self.qdrant.as_ref() {
1191 Some(store) => store.health_check().await,
1192 None => false,
1193 }
1194 }
1195
1196 /// Check if a vector store client is configured (may not be connected).
1197 #[must_use]
1198 pub fn has_vector_store(&self) -> bool {
1199 self.qdrant.is_some()
1200 }
1201
1202 /// Return a reference to the embedding store, if configured.
1203 #[must_use]
1204 pub fn embedding_store(&self) -> Option<&Arc<EmbeddingStore>> {
1205 self.qdrant.as_ref()
1206 }
1207
1208 /// Return a reference to the underlying LLM provider (used for RPE embedding).
1209 pub fn provider(&self) -> &AnyProvider {
1210 &self.provider
1211 }
1212
1213 /// Count messages in a conversation.
1214 ///
1215 /// # Errors
1216 ///
1217 /// Returns an error if the query fails.
1218 pub async fn message_count(
1219 &self,
1220 conversation_id: crate::types::ConversationId,
1221 ) -> Result<i64, MemoryError> {
1222 self.sqlite.count_messages(conversation_id).await
1223 }
1224
1225 /// Count messages not yet covered by any summary.
1226 ///
1227 /// # Errors
1228 ///
1229 /// Returns an error if the query fails.
1230 pub async fn unsummarized_message_count(
1231 &self,
1232 conversation_id: crate::types::ConversationId,
1233 ) -> Result<i64, MemoryError> {
1234 let after_id = self
1235 .sqlite
1236 .latest_summary_last_message_id(conversation_id)
1237 .await?
1238 .unwrap_or(crate::types::MessageId(0));
1239 self.sqlite
1240 .count_messages_after(conversation_id, after_id)
1241 .await
1242 }
1243
1244 /// Load recent episodic messages for the promotion-scan window.
1245 ///
1246 /// Returns up to `max_items` of the most recent non-deleted messages across all
1247 /// conversations, with their `conversation_id` for session-count heuristics.
1248 ///
1249 /// # Embedding note
1250 ///
1251 /// When Qdrant is configured, embeddings are populated by fetching `chunk_index = 0`
1252 /// vectors from the vector store via [`EmbeddingStore::get_vectors_for_messages`].
1253 /// Messages whose vector cannot be retrieved are still returned with `embedding: None`;
1254 /// the promotion engine skips those rows rather than re-embedding on the hot path.
1255 ///
1256 /// When Qdrant is not configured, all inputs carry `embedding: None`.
1257 ///
1258 /// Vectors whose dimension disagrees with the first non-empty vector in the batch
1259 /// are dropped with a single `WARN` log and treated as missing.
1260 ///
1261 /// # Errors
1262 ///
1263 /// Returns [`MemoryError`] if the underlying `SQLite` query or vector store fetch fails.
1264 pub async fn load_promotion_window(
1265 &self,
1266 max_items: usize,
1267 ) -> Result<Vec<crate::compression::promotion::PromotionInput>, MemoryError> {
1268 use zeph_db::sql;
1269
1270 let limit = i64::try_from(max_items).unwrap_or(i64::MAX);
1271 let rows: Vec<(
1272 crate::types::MessageId,
1273 crate::types::ConversationId,
1274 String,
1275 )> = zeph_db::query_as(sql!(
1276 "SELECT id, conversation_id, content \
1277 FROM messages \
1278 WHERE deleted_at IS NULL \
1279 ORDER BY id DESC \
1280 LIMIT ?"
1281 ))
1282 .bind(limit)
1283 .fetch_all(self.sqlite.pool())
1284 .await?;
1285
1286 let mut vectors = if let Some(qdrant) = &self.qdrant {
1287 let ids: Vec<_> = rows.iter().map(|(id, _, _)| *id).collect();
1288 let mut raw = qdrant.get_vectors_for_messages(&ids).await?;
1289
1290 // Dimension validation: find reference dim from the first non-empty vector.
1291 let ref_dim = raw.values().next().map(Vec::len);
1292 if let Some(ref_dim) = ref_dim {
1293 let mismatched: Vec<_> = raw
1294 .iter()
1295 .filter(|(_, v)| v.len() != ref_dim)
1296 .map(|(id, v)| (*id, v.len()))
1297 .collect();
1298 if !mismatched.is_empty() {
1299 tracing::warn!(
1300 expected_dim = ref_dim,
1301 dropped_count = mismatched.len(),
1302 "load_promotion_window: dimension mismatch — dropping mismatched vectors"
1303 );
1304 for (id, _) in mismatched {
1305 raw.remove(&id);
1306 }
1307 }
1308 }
1309 raw
1310 } else {
1311 std::collections::HashMap::new()
1312 };
1313
1314 Ok(rows
1315 .into_iter()
1316 .map(|(message_id, conversation_id, content)| {
1317 crate::compression::promotion::PromotionInput {
1318 message_id,
1319 conversation_id,
1320 content,
1321 embedding: vectors.remove(&message_id),
1322 }
1323 })
1324 .collect())
1325 }
1326
1327 /// Retrieve top-k reasoning strategies by embedding similarity to `query`.
1328 ///
1329 /// Returns an empty vec when reasoning memory is not attached, Qdrant is unavailable,
1330 /// or the provider does not support embeddings.
1331 ///
1332 /// This method is **pure** — it does not increment `use_count` or `last_used_at`.
1333 /// Call [`crate::reasoning::ReasoningMemory::mark_used`] with the ids of strategies
1334 /// actually injected into the prompt (after budget truncation).
1335 ///
1336 /// # Errors
1337 ///
1338 /// Returns an error if embedding generation or the vector search fails.
1339 pub async fn retrieve_reasoning_strategies(
1340 &self,
1341 query: &str,
1342 limit: usize,
1343 ) -> Result<Vec<crate::reasoning::ReasoningStrategy>, MemoryError> {
1344 let Some(reasoning) = &self.reasoning else {
1345 return Ok(Vec::new());
1346 };
1347 if !self.effective_embed_provider().supports_embeddings() {
1348 return Ok(Vec::new());
1349 }
1350 let embedding = match tokio::time::timeout(
1351 self.embed_timeout,
1352 self.effective_embed_provider().embed(query),
1353 )
1354 .await
1355 {
1356 Ok(Ok(v)) => v,
1357 Ok(Err(e)) => return Err(e.into()),
1358 Err(_) => {
1359 tracing::warn!("retrieve_reasoning_strategies: embed timed out, returning empty");
1360 return Ok(Vec::new());
1361 }
1362 };
1363 reasoning
1364 .retrieve_by_embedding(&embedding, limit as u64)
1365 .await
1366 }
1367}