diff --git a/core/continuum-core/src/cognition/dream_consolidation.rs b/core/continuum-core/src/cognition/dream_consolidation.rs index 3a73c53e3..01dff9bee 100644 --- a/core/continuum-core/src/cognition/dream_consolidation.rs +++ b/core/continuum-core/src/cognition/dream_consolidation.rs @@ -329,6 +329,19 @@ impl SemanticDistiller { persona_id: persona_id.map(|id| id.to_string()), }; + // Take a NON-directed serving lane before the dream's model call. Dream + // distillation is InferenceHeavy background work, but UNGUARDED it called + // the adapter directly and grabbed a physical llama decode lane WITHOUT + // counting against the non-directed budget — silently bypassing the #139 + // reservation that always holds one lane for a directed chat turn. A + // directed reply then queued INSIDE llama behind the dream, so an idle + // persona looked comatose and unresponsive (glass-boxed live 2026-07-26: + // 4 personas dreamed continuously 00:00→05:03, no reply to direct chat). + // Acquiring the permit here caps ALL dream inference at MAX_LANES-1 and + // guarantees a waiting chat turn wins the lane — the same guarantee the + // turn path already relies on (llm_deliberation_faculty acquires it too). + // Held across the generate call, released on drop. + let _lane = crate::cognition::resource_admission::acquire_serving_lane(false).await; let response = self .adapter .generate_text(request) @@ -468,6 +481,22 @@ const REVIEW_ONLY_BATCH: usize = 6; /// re-judging a conclusion made minutes ago is churn, not hygiene. Mechanical /// age gate (selection), never judgment. const REVIEW_MIN_AGE_MS: u64 = 10 * 60 * 1000; + +/// Minimum wall-clock rest between two quiet-day REVIEW-ONLY passes for the SAME +/// persona. Belief hygiene is a gentle background trickle, not a grind: without +/// this, a persona with a large belief store (Atlas held 4,336 engrams) drains +/// `REVIEW_ONLY_BATCH` at a time with NO pause, so the governor re-ticks the +/// drained-but-not-empty queue every few seconds and the review runs back-to-back +/// for HOURS — a single-lane inference storm that starves reactive responding and +/// makes an idle persona look comatose (glass-boxed live 2026-07-26: 4 personas +/// dreamed continuously 00:00→05:03, unresponsive to chat; compounded by the +/// in-memory `reviewed` set resetting on each of ~6 dev restarts → full re-review +/// every boot). This gate does NOT touch consolidation of FRESH experience (that +/// stays ungated — real learning is never throttled); it only paces the +/// quiet-day hygiene review so the lane stays free for the being to respond and +/// act. Code, not env (concurrency guide). [[the-fade-is-necessary...]] — the +/// fade is right, but it must trickle, never flood. +const REVIEW_ONLY_COOLDOWN_MS: u64 = 5 * 60 * 1000; /// Default minimum cluster size — below this an episode is not yet a pattern /// worth generalizing into a fact. const DEFAULT_MIN_CLUSTER: usize = 2; @@ -539,6 +568,10 @@ pub struct DreamConsolidationRegion { /// re-reviews from the oldest — harmless (demotion is idempotent, admission /// dedups) and self-healing. reviewed: Arc>>>, + /// Per-persona wall-clock ms of the last quiet-day review-only pass. The + /// [`REVIEW_ONLY_COOLDOWN_MS`] rest gate reads this so belief hygiene trickles + /// instead of grinding the serving lane back-to-back (see that const's doc). + last_review_ms: Arc>>, } /// Process-global handle to the ONE live dream region, installed at the ipc @@ -568,6 +601,7 @@ impl DreamConsolidationRegion { consolidated: Arc::new(Mutex::new(HashMap::new())), in_flight: Arc::new(Mutex::new(HashSet::new())), reviewed: Arc::new(Mutex::new(HashMap::new())), + last_review_ms: Arc::new(Mutex::new(HashMap::new())), } } @@ -709,6 +743,18 @@ impl DreamConsolidationRegion { reflector: &PersonaReflector, persona_id: Uuid, ) -> Option { + // Rest gate: belief hygiene trickles, never grinds. If this persona ran a + // review-only pass within the cooldown, sleep instead of launching another + // — the lane stays free for reactive responding + real work rather than + // a back-to-back review storm (see REVIEW_ONLY_COOLDOWN_MS). The first + // pass (no prior timestamp) is always allowed. + let now = now_ms(); + if let Some(&last) = self.last_review_ms.lock().unwrap().get(&persona_id) { + if now.saturating_sub(last) < REVIEW_ONLY_COOLDOWN_MS { + return None; + } + } + let already = self .reviewed .lock() @@ -716,7 +762,7 @@ impl DreamConsolidationRegion { .get(&persona_id) .cloned() .unwrap_or_default(); - let cutoff = now_ms().saturating_sub(REVIEW_MIN_AGE_MS); + let cutoff = now.saturating_sub(REVIEW_MIN_AGE_MS); let beliefs = reflector.admission.semantic_beliefs_oldest_excluding( &already, cutoff, diff --git a/core/continuum-core/src/cognition/embedding.rs b/core/continuum-core/src/cognition/embedding.rs index 3da1a6fb3..9bbe76bbc 100644 --- a/core/continuum-core/src/cognition/embedding.rs +++ b/core/continuum-core/src/cognition/embedding.rs @@ -279,6 +279,140 @@ impl EmbeddingCache { pub fn is_empty(&self) -> bool { self.map.is_empty() } + + /// Serialize the cache to a compact, dep-free binary snapshot at `path` + /// (atomic: write a sibling temp file, then rename). This is what lets a + /// content's vector survive a core restart — the difference between + /// re-embedding 375 memories (and re-fighting for VRAM) on every boot and a + /// warm cache that embeds each unique content ONCE, ever. Format, all + /// little-endian: `[u64 count]` then per entry `[u64 key][u32 dim][dim × f32]`. + /// Best-effort: the caller treats a failure as a warn, never fatal — a lost + /// snapshot just means the cache rebuilds by re-embedding. + pub fn snapshot_to(&self, path: &std::path::Path) -> std::io::Result { + use std::io::Write; + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + let tmp = path.with_extension("bin.tmp"); + let mut w = std::io::BufWriter::new(std::fs::File::create(&tmp)?); + // Snapshot the keys first so the count matches the bytes even if the map + // grows during the write (a concurrent insert simply lands in the next + // snapshot). Iterating clones under DashMap's per-shard locks — brief. + let entries: Vec<(u64, Vec)> = + self.map.iter().map(|e| (*e.key(), e.value().clone())).collect(); + w.write_all(&(entries.len() as u64).to_le_bytes())?; + for (key, vec) in &entries { + w.write_all(&key.to_le_bytes())?; + w.write_all(&(vec.len() as u32).to_le_bytes())?; + for f in vec { + w.write_all(&f.to_le_bytes())?; + } + } + w.flush()?; + drop(w); + std::fs::rename(&tmp, path)?; + Ok(entries.len()) + } + + /// Load a snapshot written by [`snapshot_to`] into the map (additive; honors + /// the [`EMBEDDING_CACHE_MAX`] cap; skips a truncated tail rather than + /// failing). A missing file is `Ok(0)` — first boot, nothing to warm from. + /// Returns the count loaded. + pub fn load_from(&self, path: &std::path::Path) -> std::io::Result { + let bytes = match std::fs::read(path) { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0), + Err(e) => return Err(e), + }; + let mut pos = 0usize; + let take = |pos: &mut usize, n: usize| -> Option<&[u8]> { + if *pos + n > bytes.len() { + return None; + } + let s = &bytes[*pos..*pos + n]; + *pos += n; + Some(s) + }; + let count = match take(&mut pos, 8) { + Some(b) => u64::from_le_bytes(b.try_into().unwrap()) as usize, + None => return Ok(0), + }; + let mut loaded = 0usize; + for _ in 0..count { + if self.map.len() >= EMBEDDING_CACHE_MAX { + break; + } + let key = match take(&mut pos, 8) { + Some(b) => u64::from_le_bytes(b.try_into().unwrap()), + None => break, // truncated tail — keep what parsed + }; + let dim = match take(&mut pos, 4) { + Some(b) => u32::from_le_bytes(b.try_into().unwrap()) as usize, + None => break, + }; + let raw = match take(&mut pos, dim * 4) { + Some(b) => b, + None => break, + }; + let vec: Vec = raw + .chunks_exact(4) + .map(|c| f32::from_le_bytes(c.try_into().unwrap())) + .collect(); + self.map.insert(key, vec); + loaded += 1; + } + Ok(loaded) + } +} + +/// Cadence for the embedding-cache snapshot — a background consolidator, so the +/// slower end of the ladder (concurrency guide: 30s–5min). A crash loses at most +/// this window of newly-embedded vectors, which simply re-embed on next use. +const EMBEDDING_CACHE_FLUSH_SECS: u64 = 60; + +/// Warm the process-global embedding cache from `path` NOW (boot warm-start), then +/// snapshot it every [`EMBEDDING_CACHE_FLUSH_SECS`] on its OWN tokio task — the +/// RTOS shape (own task + `tokio::time::interval`, off every hot path; the file +/// write itself is `spawn_blocking`). Best-effort throughout: a failed load or +/// flush is a warn, never fatal. Call once at boot, inside the tokio runtime. +pub fn spawn_embedding_cache_persistence(cache: Arc, path: std::path::PathBuf) { + match cache.load_from(&path) { + Ok(n) => crate::probe!( + class = "embedding.cache.loaded", + vectors = n, + "warmed embedding cache from durable snapshot — no re-embed for cached content" + ), + Err(e) => tracing::warn!( + target = "embedding_cache", + "embedding cache warm-load failed ({e}) — starting cold, will re-embed on demand" + ), + } + tokio::spawn(async move { + let mut ticker = + tokio::time::interval(std::time::Duration::from_secs(EMBEDDING_CACHE_FLUSH_SECS)); + ticker.tick().await; // consume the immediate first tick + loop { + ticker.tick().await; + let cache = cache.clone(); + let path = path.clone(); + let result = tokio::task::spawn_blocking(move || cache.snapshot_to(&path)).await; + match result { + Ok(Ok(n)) => crate::probe!( + class = "embedding.cache.flushed", + vectors = n, + "snapshotted embedding cache to durable store" + ), + Ok(Err(e)) => tracing::warn!( + target = "embedding_cache", + "embedding cache flush failed ({e}) — cache stays in-memory, retries next tick" + ), + Err(join) => tracing::warn!( + target = "embedding_cache", + "embedding cache flush task panicked ({join}) — skipping this tick" + ), + } + } + }); } /// Process-global content-addressed embedding cache — the one place a message's @@ -609,11 +743,150 @@ pub async fn resolve_recall_embedder(adapter: Arc) -> Arc Arc::new(CachingEmbeddingProvider::new(Arc::new(LexicalEmbedder::new()))) } +/// Resolve the recall embedder WITHOUT a chat adapter — for the GLOBAL memory +/// manager (agent-memory bridge + hydrated corpora), which has no per-persona +/// chat gateway to offer. Tries the dedicated in-process embed model (GPU, no +/// HTTP hop), then the lexical floor. Same process-stable, probe-gated, +/// LOUD-on-degrade contract as [`resolve_recall_embedder`]; it simply omits the +/// chat-adapter `/v1/embeddings` rung (rung 2) that only a persona's own adapter +/// can provide. Always returns a usable embedder — never errors, never panics. +pub async fn resolve_recall_embedder_local() -> Arc { + let model = crate::config_env::read("UNSLOTH_EMBED_MODEL") + .filter(|m| !m.trim().is_empty()) + .unwrap_or_else(|| CANONICAL_EMBED_MODEL.to_string()); + + if let Some((embed_adapter, gguf_id)) = local_embed_adapter() { + if let Some(provider) = try_neural_embedder(embed_adapter, &model).await { + crate::probe!( + class = "recall.embedder.resolved", + kind = "neural", + source = "in-process-llamacpp-global", + model = %model, + gguf = %gguf_id, + "global recall embedder = NEURAL (dedicated in-process Qwen3-Embedding)" + ); + return provider; + } + } + + crate::probe!( + class = "recall.embedder.resolved", + kind = "lexical", + source = "fallback-global", + model = %model, + "global recall embedder = LEXICAL — no in-process embed model serving; semantic recall DEGRADED to word-overlap" + ); + Arc::new(CachingEmbeddingProvider::new(Arc::new(LexicalEmbedder::new()))) +} + +/// A recall embedder that resolves its real backend LAZILY on first use, off the +/// boot critical path. The global memory manager is constructed during boot, +/// where NOTHING may gate the IPC socket bind on a GPU/gateway probe (the +/// concurrency guide's non-negotiable — a probe here previously wedged boot). +/// So the manager holds THIS: trivially cheap to construct, and on the first +/// `embed` it resolves the dedicated in-process neural embedder via +/// [`resolve_recall_embedder_local`] (probe-gated, LOUD-on-degrade), caches it +/// **process-stably** (one embedding space for the whole process — a query and +/// the stored vectors must live in the SAME space), and delegates. Every later +/// call reuses the resolved provider; the resolution's cost (probe + ~16 +/// calibration embeds) is paid once, on the first real recall, never at boot. +pub struct LazyRecallEmbedder { + resolved: tokio::sync::OnceCell>, +} + +impl LazyRecallEmbedder { + pub fn new() -> Self { + Self { + resolved: tokio::sync::OnceCell::new(), + } + } + + /// The resolved backend, resolving (once) on first call. + async fn backend(&self) -> &Arc { + self.resolved + .get_or_init(resolve_recall_embedder_local) + .await + } +} + +impl Default for LazyRecallEmbedder { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl EmbeddingProvider for LazyRecallEmbedder { + /// The resolved space id once known; a stable placeholder until first embed. + /// Only informational (degrade logging) — cache identity lives in the + /// resolved provider's own `CachingEmbeddingProvider`/`NeuralEmbeddingProvider`. + fn id(&self) -> &str { + self.resolved + .get() + .map(|p| p.id()) + .unwrap_or("recall-embedder(resolving)") + } + + fn dim(&self) -> usize { + self.resolved.get().map(|p| p.dim()).unwrap_or(0) + } + + async fn embed(&self, text: &str) -> Vec { + self.backend().await.embed(text).await + } + + fn unrelated_null(&self) -> Option<(f32, f32)> { + self.resolved.get().and_then(|p| p.unrelated_null()) + } +} + #[cfg(test)] mod tests { use super::*; use std::sync::atomic::{AtomicUsize, Ordering}; + // what this catches: the lazy recall embedder is safe to CONSTRUCT at boot — + // no GPU/gateway probe fires until the first `embed` — and reports a stable + // "resolving" identity + dim 0 until then, so nothing on the boot path can + // wedge the IPC socket bind (the concurrency-guide non-negotiable this exists + // to honor). Resolution itself needs the model registry, exercised live. + #[test] + fn lazy_recall_embedder_is_boot_safe_before_resolution() { + let e = LazyRecallEmbedder::new(); + assert_eq!(e.id(), "recall-embedder(resolving)"); + assert_eq!(e.dim(), 0); + assert!(e.unrelated_null().is_none()); + } + + // what this catches: the persistent snapshot survives a "restart" — vectors + // written by snapshot_to load back byte-identical via load_from, so cached + // content never re-embeds (nor re-fights the serving lane for VRAM) across a + // core restart. A missing file warms as Ok(0), never an error (first boot). + #[test] + fn embedding_cache_snapshot_round_trips() { + let src = EmbeddingCache::new(); + let ka = EmbeddingCache::key("qwen3-embedding-0.6b", "the deploy went red"); + let kb = EmbeddingCache::key("qwen3-embedding-0.6b", "a heron in the shallows"); + let va = vec![0.1f32, -0.2, 0.3, 0.4]; + let vb = vec![1.0f32, 2.0, 3.0]; + src.map.insert(ka, va.clone()); + src.map.insert(kb, vb.clone()); + + let path = std::env::temp_dir().join("continuum-embcache-roundtrip.bin"); + let _ = std::fs::remove_file(&path); + assert_eq!(src.snapshot_to(&path).unwrap(), 2); + + // A fresh cache (the "restarted" process) warms from the snapshot. + let dst = EmbeddingCache::new(); + assert_eq!(dst.load_from(&path).unwrap(), 2); + assert_eq!(dst.map.get(&ka).map(|e| e.clone()), Some(va)); + assert_eq!(dst.map.get(&kb).map(|e| e.clone()), Some(vb)); + + // Missing file (first boot) → Ok(0), not an error. + let _ = std::fs::remove_file(&path); + assert_eq!(dst.load_from(&path).unwrap(), 0, "missing snapshot warms as Ok(0)"); + } + // what this catches: the cosine of identical text is ~1; orthogonal // (no shared vocabulary) is ~0. The relevance primitive is sane. #[tokio::test] diff --git a/core/continuum-core/src/inference/llamacpp_adapter.rs b/core/continuum-core/src/inference/llamacpp_adapter.rs index d6feef635..d9b43d3c1 100644 --- a/core/continuum-core/src/inference/llamacpp_adapter.rs +++ b/core/continuum-core/src/inference/llamacpp_adapter.rs @@ -1185,9 +1185,87 @@ impl AIProviderAdapter for LlamaCppAdapter { }); } let start = Instant::now(); + + // ── Governed GPU admission (mirrors cognition/eval.rs::acquire_eval_lane_slot) ── + // The embedding forward pass allocates a bounded Metal context (a 2048-token + // embedding context: ~1.2 GiB compute buffer — quadratic in n_ubatch=2048 — + // plus ~224 MiB KV). UNGOVERNED, it grabbed that context behind the governor's + // back and competed with the serving lane for VRAM; under pressure the Metal + // command buffer OOM'd and the decode returned an ALL-ZERO vector — the + // "degenerate embedding" that silently broke semantic recall. Lease the VRAM + // from the ResourceGovernor FIRST: granted ⇒ the bytes are reserved for the + // life of the guard and the decode has room to succeed; refused ⇒ fail LOUD + // here instead of allocating a doomed context that emits garbage. Embeddings + // are GPU-only ([[gpu-is-non-negotiable-every-component-no-cpu-fallback]]), so + // unlike the eval lane there is NO CPU spill — the honest degrade is a named + // refusal the caller ([`NeuralEmbeddingProvider::embed`]) already surfaces as + // "no signal", never a zero vector. + // + // Const, not env-tunable — substrate policy lives in code (concurrency guide). + // Only the per-call CONTEXT is leased here; registering the embed model's + // resident WEIGHTS (~600 MiB) as a ResourceConsumer (mirroring ServingConsumer) + // is the residency-accounting follow-up, not this admission-control slice. + const EMBED_LANE_CONSUMER_ID: &str = "embed"; + const EMBED_LANE_VRAM_BYTES: u64 = 1792 * 1024 * 1024; // ~1.75 GiB (ctx + headroom) + const EMBED_LANE_LEASE_TTL_MS: u64 = 60_000; // SIGKILL backstop; the RAII guard frees on drop + let _vram_lease = { + use crate::resources::{ + LeaseError, LeaseRequest, ReclaimPolicy, ResourceDaemon, ResourceKind, + }; + match ResourceDaemon::global() { + Some(daemon) => { + let req = LeaseRequest { + consumer_id: EMBED_LANE_CONSUMER_ID.to_string(), + kind: ResourceKind::Vram, + bytes: EMBED_LANE_VRAM_BYTES, + // A bounded, sub-second forward pass is not yanked mid-embed; + // the guard returns the bytes the instant it finishes. + ttl_ms: EMBED_LANE_LEASE_TTL_MS, + reclaim_policy: ReclaimPolicy::Pinned, + }; + match daemon.acquire_guarded(&req) { + Ok(guard) => { + crate::probe!( + class = "embed.vram.leased", + consumer = EMBED_LANE_CONSUMER_ID, + bytes = EMBED_LANE_VRAM_BYTES, + texts = texts.len(), + "embedding lane acquired a governed VRAM lease" + ); + Some(guard) + } + Err(LeaseError::InsufficientCapacity { available, .. }) => { + crate::probe!( + class = "embed.vram.refused", + requested = EMBED_LANE_VRAM_BYTES, + available = available, + "embedding VRAM lease refused — failing loud, NOT emitting degenerate zeros" + ); + return Err(format!( + "embed: VRAM lease refused — {} MiB requested, {} MiB governed-available. \ + Refusing to allocate an OOM-doomed embedding context (it would decode to \ + degenerate zeros). Free VRAM (tier down serving) and retry.", + EMBED_LANE_VRAM_BYTES / (1024 * 1024), + available / (1024 * 1024), + )); + } + Err(e) => { + return Err(format!( + "embed: VRAM lease error ({e:?}) — refusing to embed ungoverned" + )); + } + } + } + // Ungoverned node (no ResourceDaemon::global()): behavior unchanged — + // the backend's own new_context allocation is the only gate, as before. + None => None, + } + }; + let embeddings = tokio::task::spawn_blocking(move || backend.embed(&texts)) .await .map_err(|e| format!("embedding task join failed: {e}"))??; + // `_vram_lease` drops here → the VRAM is returned to the governor's board. Ok(EmbeddingResponse { embeddings, model: model_id, diff --git a/core/continuum-core/src/main.rs b/core/continuum-core/src/main.rs index 93a44217c..3563d1e74 100644 --- a/core/continuum-core/src/main.rs +++ b/core/continuum-core/src/main.rs @@ -325,28 +325,45 @@ async fn main() -> Result<(), Box> { // Initialize Hippocampus memory subsystem (task #40). Embedding is async + // adapter-routed (never an in-process ONNX model). The GLOBAL manager here - // bootstraps with the LEXICAL embedder — instant, no network on the boot - // critical path. This is non-negotiable per the concurrency guide: NOTHING - // may gate the IPC socket bind on a gateway probe (a hanging `/v1/models` - // call here previously wedged boot before the socket bound). The LIVE - // per-persona recall path resolves its own NEURAL embedder lazily on spawn - // (`supervisor` → `cognition::embedding::resolve_recall_embedder`) and logs - // gateway availability there — that is where real semantic recall lives. - // The lexical embedder is an explicit, logged, non-neural relevance embedder - // (allowed; NOT a silent ONNX fallback). Wiring neural embedding for the - // global manager (against the dedicated embed server, not the chat gateway) - // is a separate addressing follow-up. + // now holds the LAZY recall embedder: still constructed with ZERO boot-path + // cost (no GPU/gateway probe), but on the FIRST real recall it resolves the + // dedicated in-process NEURAL embedder (Qwen3-Embedding-0.6B GGUF) via + // `resolve_recall_embedder_local` — probe-gated, process-stable, LOUD if it + // has to fall to the lexical floor. This is non-negotiable per the + // concurrency guide: NOTHING may gate the IPC socket bind on a gateway probe + // (a hanging `/v1/models` call here previously wedged boot before the socket + // bound) — the lazy embedder keeps that guarantee while giving the + // agent-memory bridge + hydrated corpora real SEMANTIC recall, not the + // lexical word-overlap floor they were silently stuck on. (This is the + // "separate addressing follow-up" the old comment promised.) The per-persona + // recall path still resolves its own neural embedder on spawn. info!( - "🧠 Initializing Hippocampus (lexical bootstrap embedder; per-persona recall \ - resolves neural on spawn) — no gateway probe on the boot path" - ); - let embedding_provider: Arc = Arc::new( - continuum_core::cognition::embedding::CachingEmbeddingProvider::new(Arc::new( - continuum_core::cognition::embedding::LexicalEmbedder::new(), - )), + "🧠 Initializing Hippocampus (lazy neural recall embedder; resolves \ + in-process Qwen3-Embedding on first recall) — no gateway probe on the boot path" ); + let embedding_provider: Arc = + Arc::new(continuum_core::cognition::embedding::LazyRecallEmbedder::new()); let memory_manager = Arc::new(PersonaMemoryManager::new(embedding_provider)); + // Persist the process-global embedding cache across restarts. This is the ONE + // cache every persona AND agent shares — their recall embedders all wrap + // `global_embedding_cache()` — so warming it here brings back the whole + // citizenry's vectors at once: each unique content embeds ONCE, ever, instead + // of re-embedding (and re-fighting the serving lane for VRAM) on every boot. + // Own background task, snapshot on a slow cadence, best-effort (a lost snapshot + // just re-embeds). Off the boot critical path — the load is a fast file read, + // no gateway/GPU probe. + if let Some(home) = dirs::home_dir() { + let cache_path = home + .join(".continuum") + .join("cache") + .join("embedding-cache.bin"); + continuum_core::cognition::embedding::spawn_embedding_cache_persistence( + continuum_core::cognition::embedding::global_embedding_cache(), + cache_path, + ); + } + // Capture tokio runtime handle for async operations from IPC thread let rt_handle = tokio::runtime::Handle::current(); diff --git a/core/continuum-core/src/memory/mod.rs b/core/continuum-core/src/memory/mod.rs index 6e1e51cbc..bd0234b96 100644 --- a/core/continuum-core/src/memory/mod.rs +++ b/core/continuum-core/src/memory/mod.rs @@ -205,6 +205,29 @@ impl PersonaMemoryManager { let had_query_embedding = query_embedding.is_some(); + // Candidate-vector backfill (only when the query itself embedded — a real + // signal to rank by). The semantic layer scores the query against STORED + // memory vectors, but agent-authored memories are written WITHOUT one (the + // embedding is computed here, not at write) and a cold hydrate carries + // none — so `memories_with_embeddings()` is empty and SemanticRecallLayer + // no-ops, leaving recall in importance/recency order that IGNORES the query. + // That is the exact "same off-topic memories for every query" bug the + // agent-memory bridge hit (2026-07). Embed the missing vectors NOW — async, + // content-addressed-cached, OUTSIDE the sync recall and outside any lock — + // so the semantic layer has vectors to rank against. First recall per + // persona pays it; the cache makes repeats free. + if had_query_embedding { + let embedded = self.ensure_memory_embeddings(&corpus_lock).await; + if embedded > 0 { + crate::log_info!( + "module", + "memory_recall", + "backfilled {embedded} candidate memory embeddings for {persona_id} (embedder={}) — semantic layer can now rank", + self.embedding.id() + ); + } + } + // Phase 1: recall with read lock (pure sync compute on in-memory corpus) let (response, memories_with_vectors) = { let corpus = corpus_lock.read().map_err(|e| { @@ -252,6 +275,84 @@ impl PersonaMemoryManager { Ok(response) } + /// Ensure candidate memories carry an embedding so [`recall::SemanticRecallLayer`] + /// can rank them. Snapshots `(id, content)` for every memory MISSING a vector + /// (read lock), embeds each via the manager's embedder (async, content-addressed + /// cache — OUTSIDE any lock, never held across the await), then writes the vectors + /// back (write lock). Returns the number newly embedded. + /// + /// In-memory only: the durable store round-trips `CorpusMemory.embedding`, but + /// agent memories are currently written with `None`, so this is the on-demand + /// backfill that gives the semantic layer something to score until embed-on-write + /// lands. A no-op once every memory has a vector (a cheap read-lock scan), so it + /// is safe to call on every recall. Content is immutable, so a computed vector + /// never goes stale. + async fn ensure_memory_embeddings( + &self, + corpus_lock: &Arc>, + ) -> usize { + // Bound the work per recall so a SessionStart hook stays responsive: + // - MAX_PER_CALL caps how many missing memories we embed in one recall, so a + // large cold corpus is embedded incrementally across several recalls + // (amortized) instead of blocking one call on hundreds of GPU forward + // passes. Content is immutable, so partial progress never goes stale. + // - FAIL_FAST: if the first handful of embeds all come back empty, the + // embedder is DOWN/degenerate (e.g. the Qwen3 GGUF returning zero vectors + // under GPU-OOM). Bail immediately rather than grind every remaining + // candidate through a failing forward pass — the loud-degrade signal in + // the caller already reports the resulting non-semantic recall. + const MAX_PER_CALL: usize = 64; + const FAIL_FAST: usize = 3; + + // 1. Snapshot the memories that lack a vector (read lock, dropped before await). + let missing: Vec<(String, String)> = { + let corpus = match corpus_lock.read() { + Ok(c) => c, + Err(_) => return 0, + }; + corpus + .memories + .iter() + .filter(|m| !corpus.memory_embeddings.contains_key(&m.id)) + .take(MAX_PER_CALL) + .map(|m| (m.id.clone(), m.content.clone())) + .collect() + }; + if missing.is_empty() { + return 0; + } + + // 2. Embed each missing memory's content (async, cached) — no lock held. + // An empty vector = "no signal" (embedder down / degenerate); skip it so + // the memory is retried on a later recall rather than cached as zeros. + // Fail fast if the embedder is clearly down (leading run of empties). + let mut embedded: Vec<(String, Vec)> = Vec::with_capacity(missing.len()); + let mut consecutive_empty = 0usize; + for (id, content) in missing { + let v = self.embedding.embed(&content).await; + if v.is_empty() { + consecutive_empty += 1; + if embedded.is_empty() && consecutive_empty >= FAIL_FAST { + break; // embedder is down — stop grinding failing forward passes + } + } else { + consecutive_empty = 0; + embedded.push((id, v)); + } + } + + // 3. Write the vectors back (write lock). + let n = embedded.len(); + if n > 0 { + if let Ok(mut corpus) = corpus_lock.write() { + for (id, v) in embedded { + corpus.memory_embeddings.insert(id, v); + } + } + } + n + } + // ─── Consciousness Context ──────────────────────────────────────────────── /// Build consciousness context (temporal + cross-context + intentions). @@ -465,6 +566,95 @@ mod tests { assert!(resp.load_time_ms >= 0.0); } + // what this catches: memories written WITHOUT a vector (the agent-memory + // bridge's `remember` path sets embedding: None) get embedded on demand so + // SemanticRecallLayer has something to rank against, instead of no-op'ing into + // non-semantic importance/recency order — the exact "same off-topic memories + // for every query" bug the bridge hit (2026-07). Idempotent once populated. + #[tokio::test] + async fn ensure_memory_embeddings_backfills_missing_vectors() { + let manager = test_manager(); + let mut m_none1 = make_corpus_memory("m1", "alpha lesson", 0.5); + m_none1.embedding = None; + let mut m_none2 = make_corpus_memory("m2", "beta lesson", 0.5); + m_none2.embedding = None; + let m_has = make_corpus_memory("m3", "gamma lesson", 0.5); // already Some(vec) + manager.load_corpus("p1", vec![m_none1, m_none2, m_has], vec![]); + + let corpus_lock = manager.get_corpus("p1").unwrap(); + // Precondition: only the pre-embedded memory carries a vector. + assert_eq!( + corpus_lock.read().unwrap().memories_with_embeddings().len(), + 1 + ); + + // Backfill embeds the two missing (StubEmbeddingProvider yields a real vec). + let n = manager.ensure_memory_embeddings(&corpus_lock).await; + assert_eq!(n, 2, "both unembedded memories are backfilled"); + assert_eq!( + corpus_lock.read().unwrap().memories_with_embeddings().len(), + 3, + "every memory now carries a vector the semantic layer can rank" + ); + + // Idempotent: a second pass finds nothing missing (cheap read-lock scan). + let n2 = manager.ensure_memory_embeddings(&corpus_lock).await; + assert_eq!(n2, 0, "no re-embedding once every memory has a vector"); + } + + // A stand-in for a DOWN / degenerate embedder (e.g. the Qwen3 GGUF returning + // zero vectors under GPU-OOM): every embed comes back empty. Counts calls so + // the test can prove the backfill bails instead of grinding the whole corpus. + struct DownEmbeddingProvider { + calls: Arc, + } + #[async_trait] + impl EmbeddingProvider for DownEmbeddingProvider { + fn id(&self) -> &str { + "down" + } + fn dim(&self) -> usize { + 384 + } + async fn embed(&self, _text: &str) -> Vec { + self.calls + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Vec::new() // "no signal" + } + } + + // what this catches: when the embedder is down (all embeds empty), the backfill + // FAILS FAST after a few attempts instead of grinding every candidate through a + // failing forward pass — so a SessionStart recall over a large cold corpus stays + // responsive rather than hanging the hook. (The caller's loud-degrade signal + // still reports the resulting non-semantic recall.) + #[tokio::test] + async fn ensure_memory_embeddings_fails_fast_when_embedder_down() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let manager = PersonaMemoryManager::new(Arc::new(DownEmbeddingProvider { + calls: calls.clone(), + })); + // 20 memories, none embedded. + let mems: Vec<_> = (0..20) + .map(|i| { + let mut m = make_corpus_memory(&format!("m{i}"), "content", 0.5); + m.embedding = None; + m + }) + .collect(); + manager.load_corpus("p1", mems, vec![]); + let corpus_lock = manager.get_corpus("p1").unwrap(); + + let n = manager.ensure_memory_embeddings(&corpus_lock).await; + assert_eq!(n, 0, "nothing embedded when the embedder is down"); + // Bailed after the fail-fast threshold — did NOT attempt all 20. + let attempted = calls.load(std::sync::atomic::Ordering::SeqCst); + assert!( + attempted <= 4, + "should bail after ~3 empty embeds, not grind all 20 (attempted {attempted})" + ); + } + #[tokio::test] async fn test_multi_layer_recall() { let manager = test_manager();