use std::collections::{HashMap, HashSet};
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};
use lunaris_consolidate::Consolidator;
use lunaris_core::{
Embedder, HlcClock, KeywordPort, Lsn, LunarisError, Scope, StorageError, StoragePort,
};
use lunaris_core::activation::RefSignal;
use lunaris_core::keyspace::activation_key;
use lunaris_core::keyspace::verify_agenda_key;
use lunaris_ingest::{BakoffConfig, TokenCounter, make_token_counter};
use serde::{Deserialize, Serialize};
use ulid::Ulid;
use crate::episode_builder::EpisodeBuilder;
use lunaris_extract::{Extractor, NoopExtractor};
use lunaris_rerank::{NoopReranker, Reranker};
use lunaris_storage_moon::MoonStorage;
use lunaris_verify::{
BOOST_DELTA, NoopReflectSupervisor, NoopVerifier, ReflectInput, ReflectOutput,
ReflectSupervisor, Verifier, apply_reflect_boost, apply_reflect_invalidate,
boost_cache_capacity,
};
use crate::consolidator_pipeline::ConsolidatorPipelineHandle;
use crate::graph_pipeline::GraphPipelineHandle;
use crate::verify_pipeline::VerifierPipelineHandle;
#[derive(Clone)]
pub struct Lunaris {
pub(crate) storage: Arc<dyn StoragePort>,
pub(crate) keyword: Arc<dyn KeywordPort>,
pub(crate) embedder: Arc<dyn Embedder>,
pub(crate) clock: Arc<HlcClock>,
pub(crate) moon_storage: Option<Arc<MoonStorage>>,
pub(crate) reranker: Arc<dyn Reranker>,
pub(crate) recall_rerank: crate::recall_rerank::RecallRerankConfig,
pub(crate) graph_pipeline: Arc<GraphPipelineHandle>,
pub(crate) verify_pipeline: Arc<VerifierPipelineHandle>,
pub(crate) consolidator_pipeline: Arc<ConsolidatorPipelineHandle>,
pub(crate) reflect_supervisor: Arc<dyn ReflectSupervisor>,
pub(crate) boost_cache: Arc<parking_lot::RwLock<lru::LruCache<(Scope, Ulid), f32>>>,
pub(crate) warm_up_semaphore: Arc<tokio::sync::Semaphore>,
pub(crate) token_counter: Arc<dyn TokenCounter + Send + Sync>,
pub(crate) bakeoff_config: Option<Arc<BakoffConfig>>,
}
struct CachedEmbedder {
inner: Arc<dyn Embedder>,
cache: parking_lot::RwLock<lru::LruCache<String, Vec<f32>>>,
hits: AtomicUsize,
misses: AtomicUsize,
}
impl CachedEmbedder {
fn new(inner: Arc<dyn Embedder>, capacity: NonZeroUsize) -> Self {
Self {
inner,
cache: parking_lot::RwLock::new(lru::LruCache::new(capacity)),
hits: AtomicUsize::new(0),
misses: AtomicUsize::new(0),
}
}
async fn embed_batch_with(
&self,
inputs: &[&str],
lowpri: bool,
) -> Result<Vec<Vec<f32>>, LunarisError> {
let mut out: Vec<Option<Vec<f32>>> = vec![None; inputs.len()];
let mut missing: HashMap<String, Vec<usize>> = HashMap::new();
{
let cache = self.cache.read();
for (idx, input) in inputs.iter().enumerate() {
if let Some(cached) = cache.peek(*input) {
out[idx] = Some(cached.clone());
self.hits.fetch_add(1, Ordering::Relaxed);
} else {
missing.entry((*input).to_string()).or_default().push(idx);
}
}
}
if !missing.is_empty() {
let keys: Vec<String> = missing.keys().cloned().collect();
let refs: Vec<&str> = keys.iter().map(String::as_str).collect();
let embedded = if lowpri {
self.inner.embed_batch_lowpri(&refs).await?
} else {
self.inner.embed_batch(&refs).await?
};
if embedded.len() != keys.len() {
return Err(LunarisError::Storage(StorageError::Backend(format!(
"cached embedder inner returned {} rows for {} inputs",
embedded.len(),
keys.len()
))));
}
let mut cache = self.cache.write();
for (key, embedding) in keys.into_iter().zip(embedded.into_iter()) {
self.misses.fetch_add(1, Ordering::Relaxed);
cache.put(key.clone(), embedding.clone());
if let Some(indices) = missing.remove(&key) {
for idx in indices {
out[idx] = Some(embedding.clone());
}
}
}
}
out.into_iter()
.map(|row| {
row.ok_or_else(|| {
LunarisError::Storage(StorageError::Backend(
"cached embedder failed to fill an output row".into(),
))
})
})
.collect()
}
}
impl std::fmt::Debug for CachedEmbedder {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CachedEmbedder")
.field("dim", &self.inner.dim())
.field("cache_len", &self.cache.read().len())
.field("hits", &self.hits.load(Ordering::Relaxed))
.field("misses", &self.misses.load(Ordering::Relaxed))
.finish()
}
}
#[async_trait::async_trait]
impl Embedder for CachedEmbedder {
fn dim(&self) -> usize {
self.inner.dim()
}
async fn embed_batch(&self, inputs: &[&str]) -> Result<Vec<Vec<f32>>, LunarisError> {
self.embed_batch_with(inputs, false).await
}
async fn embed_batch_lowpri(&self, inputs: &[&str]) -> Result<Vec<Vec<f32>>, LunarisError> {
self.embed_batch_with(inputs, true).await
}
}
impl std::fmt::Debug for Lunaris {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Lunaris")
.field("backend_capabilities", &self.storage.capabilities())
.field("embedder_dim", &self.embedder.dim())
.field("clock_node_id", &self.clock.node_id())
.field("has_moon_native_path", &self.moon_storage.is_some())
.field("reranker_applies", &self.reranker.applies())
.field("graph_pipeline_enabled", &self.graph_pipeline.is_enabled())
.field("verify_pipeline_enabled", &self.verify_pipeline.is_enabled())
.field("consolidator_pipeline_enabled", &self.consolidator_pipeline.is_enabled())
.field("reflect_supervisor_applies", &self.reflect_supervisor.applies())
.field("boost_cache_len", &self.boost_cache.read().len())
.field("warm_up_semaphore_permits", &self.warm_up_semaphore.available_permits())
.finish()
}
}
impl Lunaris {
#[must_use]
pub fn embedder_backend(&self) -> String {
resolved_embedder_backend().as_str().to_string()
}
pub async fn open(url: &str) -> Result<Self, LunarisError> {
let embedder = resolve_embedder(embed_max_batch_tokens()).await?;
announce_degradation_once();
Self::open_with_embedder(url, embedder).await
}
pub async fn open_with_embedder(
url: &str,
embedder: Arc<dyn Embedder>,
) -> Result<Self, LunarisError> {
let embedder = maybe_cached_embedder(embedder);
let scheme = url.split("://").next().unwrap_or("");
let clock = HlcClock::new(0);
let token_counter = make_token_counter(Some(&embedder_dir().join("tokenizer.json")));
let reranker = resolve_reranker().await?;
let extractor = default_extractor().await;
let initial_graph_state = GraphPipelineHandle::initial_state_from_env();
let graph_pipeline = Arc::new(GraphPipelineHandle::new(initial_graph_state, extractor));
let verifier = default_verifier().await;
let consolidator = default_consolidator()?;
let initial_verify_state = VerifierPipelineHandle::initial_state_from_env();
let initial_consolidate_state = ConsolidatorPipelineHandle::initial_state_from_env();
let verify_pipeline = Arc::new(VerifierPipelineHandle::new(initial_verify_state, verifier));
let consolidator_pipeline =
Arc::new(ConsolidatorPipelineHandle::new(initial_consolidate_state, consolidator));
match scheme {
"moon" => {
let m = Arc::new(MoonStorage::connect_with_dim(url, embedder.dim()).await?);
let storage_arc: Arc<dyn StoragePort> = m.clone();
verify_pipeline.bind_storage(storage_arc.clone());
verify_pipeline.bind_clock(clock.clone());
consolidator_pipeline.bind_storage(storage_arc.clone());
if initial_verify_state {
verify_pipeline.spawn_worker_if_idle();
}
if initial_consolidate_state {
consolidator_pipeline.spawn_worker_if_idle();
}
Ok(Self {
storage: storage_arc,
keyword: m.clone() as Arc<dyn KeywordPort>,
embedder,
clock,
moon_storage: Some(m),
reranker,
recall_rerank: crate::recall_rerank::RecallRerankConfig::from_env(),
graph_pipeline,
verify_pipeline,
consolidator_pipeline,
reflect_supervisor: Arc::new(NoopReflectSupervisor),
boost_cache: Arc::new(parking_lot::RwLock::new(lru::LruCache::new(
boost_cache_capacity(),
))),
warm_up_semaphore: Arc::new(tokio::sync::Semaphore::new(
resolve_prewarm_concurrency(),
)),
token_counter: token_counter.clone(),
bakeoff_config: None,
})
}
other => {
Err(LunarisError::Storage(crate::open::retired_scheme_error(other).unwrap_or_else(
|| lunaris_core::StorageError::UnsupportedScheme(other.to_string()),
)))
}
}
}
#[doc(hidden)]
pub fn with_parts(
storage: Arc<dyn StoragePort>,
embedder: Arc<dyn Embedder>,
clock: Arc<HlcClock>,
) -> Self {
let verify_pipeline = Arc::new(VerifierPipelineHandle::new(
false,
Arc::new(NoopVerifier) as Arc<dyn Verifier>,
));
let consolidator = ConsolidatorPipelineHandle::backend_from_env()
.expect("LUNARIS_CONSOLIDATOR_BACKEND resolution failed in with_parts test seam");
let consolidator_pipeline = Arc::new(ConsolidatorPipelineHandle::new(false, consolidator));
verify_pipeline.bind_storage(storage.clone());
verify_pipeline.bind_clock(clock.clone());
consolidator_pipeline.bind_storage(storage.clone());
Self {
storage,
keyword: Arc::new(NoKeywordSupport) as Arc<dyn KeywordPort>,
embedder,
clock,
moon_storage: None,
reranker: Arc::new(NoopReranker) as Arc<dyn Reranker>,
recall_rerank: crate::recall_rerank::RecallRerankConfig::default(),
graph_pipeline: Arc::new(GraphPipelineHandle::new(
false,
Arc::new(NoopExtractor) as Arc<dyn Extractor>,
)),
verify_pipeline,
consolidator_pipeline,
reflect_supervisor: Arc::new(NoopReflectSupervisor),
boost_cache: Arc::new(parking_lot::RwLock::new(lru::LruCache::new(
boost_cache_capacity(),
))),
warm_up_semaphore: Arc::new(tokio::sync::Semaphore::new(resolve_prewarm_concurrency())),
token_counter: make_token_counter(None),
bakeoff_config: None,
}
}
#[doc(hidden)]
pub fn with_parts_keyword(
storage: Arc<dyn StoragePort>,
keyword: Arc<dyn KeywordPort>,
embedder: Arc<dyn Embedder>,
clock: Arc<HlcClock>,
) -> Self {
let verify_pipeline = Arc::new(VerifierPipelineHandle::new(
false,
Arc::new(NoopVerifier) as Arc<dyn Verifier>,
));
let consolidator = ConsolidatorPipelineHandle::backend_from_env().expect(
"LUNARIS_CONSOLIDATOR_BACKEND resolution failed in with_parts_keyword test seam",
);
let consolidator_pipeline = Arc::new(ConsolidatorPipelineHandle::new(false, consolidator));
verify_pipeline.bind_storage(storage.clone());
verify_pipeline.bind_clock(clock.clone());
consolidator_pipeline.bind_storage(storage.clone());
Self {
storage,
keyword,
embedder,
clock,
moon_storage: None,
reranker: Arc::new(NoopReranker) as Arc<dyn Reranker>,
recall_rerank: crate::recall_rerank::RecallRerankConfig::default(),
graph_pipeline: Arc::new(GraphPipelineHandle::new(
false,
Arc::new(NoopExtractor) as Arc<dyn Extractor>,
)),
verify_pipeline,
consolidator_pipeline,
reflect_supervisor: Arc::new(NoopReflectSupervisor),
boost_cache: Arc::new(parking_lot::RwLock::new(lru::LruCache::new(
boost_cache_capacity(),
))),
warm_up_semaphore: Arc::new(tokio::sync::Semaphore::new(resolve_prewarm_concurrency())),
token_counter: make_token_counter(None),
bakeoff_config: None,
}
}
pub fn with_embedder(mut self, embedder: Arc<dyn Embedder>) -> Self {
if self.embedder.dim() != embedder.dim() {
tracing::warn!(
target: "lunaris::handle",
store_dim = self.embedder.dim(),
new_dim = embedder.dim(),
"with_embedder: dim mismatch — silently swapping; vector index is sized for store_dim. \
Use try_with_embedder() to refuse the swap, or open_with_embedder() for a fresh handle."
);
}
self.embedder = maybe_cached_embedder(embedder);
self
}
pub fn with_bakeoff(mut self, config: Arc<BakoffConfig>) -> Self {
self.bakeoff_config = Some(config);
self
}
pub fn try_with_embedder(mut self, embedder: Arc<dyn Embedder>) -> Result<Self, LunarisError> {
let store_dim = self.embedder.dim();
let new_dim = embedder.dim();
if store_dim != new_dim {
return Err(LunarisError::Storage(lunaris_core::StorageError::Backend(format!(
"embedder dim {new_dim} != store dim {store_dim}; drop and re-open with \
matching config or migrate (no auto-resize — vectors at the storage \
layer are sized for a specific dim, swapping would produce garbage \
similarity scores)"
))));
}
self.embedder = maybe_cached_embedder(embedder);
Ok(self)
}
pub fn with_reranker(mut self, reranker: Arc<dyn Reranker>) -> Self {
self.reranker = reranker;
self
}
pub fn with_extractor(self, extractor: Arc<dyn Extractor>) -> Self {
self.graph_pipeline.set_extractor(extractor);
self
}
pub fn with_verifier(self, verifier: Arc<dyn Verifier>) -> Self {
self.verify_pipeline.set_verifier(verifier);
self
}
pub fn with_consolidator(self, consolidator: Arc<dyn Consolidator>) -> Self {
self.consolidator_pipeline.set_consolidator(consolidator);
self
}
pub fn with_reflect_supervisor(mut self, supervisor: Arc<dyn ReflectSupervisor>) -> Self {
self.reflect_supervisor = supervisor;
self
}
pub async fn end_turn(&self, input: ReflectInput) -> Result<ReflectOutput, LunarisError> {
let turn_id = input.turn_id;
let output = self.reflect_supervisor.reflect(input).await?;
tracing::info!(
target: "lunaris::handle",
turn_id = ?turn_id,
invalidate_count = output.invalidate.len(),
boost_count = output.boost.len(),
pre_warm_query = output.pre_warm_query.is_some(),
"end_turn_reflect_complete"
);
Ok(output)
}
pub fn storage(&self) -> Arc<dyn StoragePort> {
self.storage.clone()
}
pub fn keyword(&self) -> Arc<dyn KeywordPort> {
self.keyword.clone()
}
pub fn embedder(&self) -> Arc<dyn Embedder> {
self.embedder.clone()
}
pub async fn health_check(&self) -> Result<(), LunarisError> {
self.storage.health_check().await.map_err(LunarisError::Storage)
}
pub fn clock(&self) -> Arc<HlcClock> {
self.clock.clone()
}
pub fn moon_storage(&self) -> Option<Arc<MoonStorage>> {
self.moon_storage.clone()
}
pub fn reranker(&self) -> Arc<dyn Reranker> {
self.reranker.clone()
}
pub fn graph_pipeline(&self) -> Arc<GraphPipelineHandle> {
self.graph_pipeline.clone()
}
pub fn extractor(&self) -> Option<Arc<dyn Extractor>> {
self.graph_pipeline.snapshot_extractor()
}
pub fn verify_pipeline(&self) -> Arc<VerifierPipelineHandle> {
self.verify_pipeline.clone()
}
pub fn consolidator_pipeline(&self) -> Arc<ConsolidatorPipelineHandle> {
self.consolidator_pipeline.clone()
}
pub fn verifier(&self) -> Option<Arc<dyn Verifier>> {
self.verify_pipeline.snapshot_verifier()
}
pub fn consolidator(&self) -> Option<Arc<dyn Consolidator>> {
self.consolidator_pipeline.snapshot_consolidator()
}
pub fn reflect_supervisor(&self) -> Arc<dyn ReflectSupervisor> {
self.reflect_supervisor.clone()
}
pub fn warm_up_semaphore(&self) -> Arc<tokio::sync::Semaphore> {
self.warm_up_semaphore.clone()
}
#[doc(hidden)]
pub fn with_prewarm_concurrency(mut self, capacity: usize) -> Self {
self.warm_up_semaphore = Arc::new(tokio::sync::Semaphore::new(capacity));
self
}
pub fn scoped(&self, scope: Scope) -> ScopedLunaris<'_> {
ScopedLunaris { engine: self, scope }
}
pub async fn list_scopes(
&self,
prefix: Option<&str>,
limit: usize,
cursor: Option<&str>,
) -> Result<lunaris_core::ScopePage, LunarisError> {
self.storage.list_scopes(prefix, limit, cursor).await.map_err(LunarisError::from)
}
pub async fn invalidate_range(
&self,
scope: &Scope,
node_id: &str,
hlc_wall_lo_inclusive: i64,
hlc_wall_hi_inclusive: i64,
) -> Result<u64, LunarisError> {
crate::invalidate::invalidate_range(
&self.storage,
scope,
node_id,
hlc_wall_lo_inclusive,
hlc_wall_hi_inclusive,
)
.await
}
}
#[derive(Debug, Clone, Copy)]
struct NoKeywordSupport;
#[async_trait::async_trait]
impl KeywordPort for NoKeywordSupport {
async fn keyword_search(
&self,
_scope: &lunaris_core::Scope,
_index: &str,
_query: &str,
_k: usize,
_filter: Option<&lunaris_core::Filter>,
_as_of: Option<lunaris_core::Hlc>,
) -> Result<Vec<lunaris_core::KeywordHit>, lunaris_core::StorageError> {
Err(lunaris_core::StorageError::NotSupported(
"Lunaris::with_parts was called without a KeywordPort — use with_parts_keyword or open(url)",
))
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IngestKind {
Fresh,
Duplicate(lunaris_core::Lsn),
}
pub struct ScopedLunaris<'a> {
pub(crate) engine: &'a Lunaris,
pub(crate) scope: Scope,
}
impl<'a> ScopedLunaris<'a> {
pub fn scope(&self) -> &Scope {
&self.scope
}
pub async fn ingest(&self, builder: EpisodeBuilder) -> Result<Lsn, LunarisError> {
let episode = builder.into_episode(self.scope.clone(), &self.engine.clock);
self.engine.ingest(episode).await
}
pub async fn ingest_idempotent(
&self,
builder: EpisodeBuilder,
dedupe_key: &str,
) -> Result<(Lsn, IngestKind), LunarisError> {
match self.engine.storage.lookup_by_dedupe_key(&self.scope, dedupe_key).await {
Ok(Some(prior_lsn)) => {
tracing::debug!(
dedupe_key,
prior_lsn = %prior_lsn,
scope = self.scope.as_str(),
"duplicate dedupe key — returning prior LSN without ingest",
);
return Ok((prior_lsn, IngestKind::Duplicate(prior_lsn)));
}
Ok(None) => {}
Err(e) => {
tracing::warn!(
err = %e,
dedupe_key,
"dedupe key lookup failed — proceeding as fresh ingest",
);
}
}
let lsn = self.ingest(builder).await?;
if let Err(e) = self.engine.storage.insert_dedupe_key(&self.scope, dedupe_key, lsn).await {
tracing::warn!(
err = %e,
dedupe_key,
lsn = %lsn,
"dedupe key insert failed — continuing (non-fatal, T-24-03-06 race window)",
);
}
Ok((lsn, IngestKind::Fresh))
}
pub async fn ingest_structured(
&self,
payload: crate::structured_ingest::StructuredIngest,
) -> Result<Lsn, LunarisError> {
self.engine.ingest_structured(payload, self.scope.clone()).await
}
pub async fn recall(
&self,
query: lunaris_retrieve::Query,
) -> Result<Vec<lunaris_retrieve::Hit>, LunarisError> {
self.engine.recall().with_scope(self.scope.clone()).execute(query).await
}
pub async fn set_retention_policy(
&self,
policy: lunaris_core::retention::RetentionPolicy,
) -> Result<(), LunarisError> {
crate::retention::write_policy(&self.engine.storage, &self.scope, policy).await
}
pub async fn retention_policy(
&self,
) -> Result<Option<lunaris_core::retention::RetentionPolicy>, LunarisError> {
crate::retention::read_policy(&self.engine.storage, &self.scope, &self.engine.clock).await
}
pub async fn enforce_retention(
&self,
) -> Result<crate::retention::RetentionReceipt, LunarisError> {
let now_ms = self.engine.clock.tick().wall_ms;
crate::retention::enforce_at(self.engine, &self.scope, now_ms).await
}
pub async fn enforce_retention_at(
&self,
now_ms: u64,
) -> Result<crate::retention::RetentionReceipt, LunarisError> {
crate::retention::enforce_at(self.engine, &self.scope, now_ms).await
}
pub async fn preview_retention(
&self,
) -> Result<crate::retention::RetentionReceipt, LunarisError> {
let now_ms = self.engine.clock.tick().wall_ms;
crate::retention::preview_at(self.engine, &self.scope, now_ms).await
}
pub async fn preview_retention_at(
&self,
now_ms: u64,
) -> Result<crate::retention::RetentionReceipt, LunarisError> {
crate::retention::preview_at(self.engine, &self.scope, now_ms).await
}
pub async fn audit_events(
&self,
from_ms: Option<u64>,
to_ms: Option<u64>,
limit: usize,
) -> Result<lunaris_core::audit::AuditPage, LunarisError> {
lunaris_core::audit::read_audit_events(
&self.engine.storage,
&self.scope,
from_ms,
to_ms,
limit,
)
.await
.map_err(LunarisError::Storage)
}
pub async fn forget(
&self,
request: impl Into<crate::forget::ForgetRequest>,
) -> Result<crate::forget::ForgetReceipt, LunarisError> {
crate::forget::forget_scoped(
&self.engine.storage,
&self.engine.clock,
&self.scope,
request.into(),
)
.await
}
pub fn dsl(&self) -> lunaris_retrieve::RetrievalBuilder {
self.engine.recall().with_scope(self.scope.clone())
}
pub async fn end_turn(&self, input: ReflectInput) -> Result<ReflectOutput, LunarisError> {
let turn_id = input.turn_id;
let output = match self.engine.reflect_supervisor.reflect(input).await {
Ok(o) => o,
Err(e) => {
tracing::warn!(
target: "lunaris::scoped",
err = %e,
turn_id = ?turn_id,
"reflect_supervisor_error; emitting empty output"
);
ReflectOutput::default()
}
};
if !output.invalidate.is_empty() {
match apply_reflect_invalidate(
&self.engine.storage,
&self.scope,
&self.engine.clock,
turn_id,
&output.invalidate,
)
.await
{
Ok(stamped) => {
tracing::debug!(
target: "lunaris::scoped",
turn_id = ?turn_id,
invalidated_count = stamped.len(),
"reflect_invalidate_applied"
);
}
Err(e) => {
tracing::warn!(
target: "lunaris::scoped",
err = %e,
turn_id = ?turn_id,
"reflect_invalidate_storage_error; continuing"
);
}
}
}
if !output.boost.is_empty() {
apply_reflect_boost(&self.engine.boost_cache, &self.scope, &output.boost, BOOST_DELTA);
tracing::debug!(
target: "lunaris::scoped",
turn_id = ?turn_id,
boost_count = output.boost.len(),
boost_delta = BOOST_DELTA,
"reflect_boost_cache_populated"
);
}
if let Some(query_str) = output.pre_warm_query.clone() {
match self.engine.warm_up_semaphore.clone().try_acquire_owned() {
Ok(permit) => {
let storage = self.engine.storage.clone();
let keyword = self.engine.keyword.clone();
let embedder = self.engine.embedder.clone();
let moon_storage = self.engine.moon_storage.clone();
let scope = self.scope.clone();
let q = query_str.clone();
tokio::spawn(async move {
let _permit = permit;
let mut builder = lunaris_retrieve::RetrievalBuilder::from_handle(
storage, keyword, embedder,
)
.with_scope(scope);
if let Some(moon) = moon_storage {
builder = builder.with_moon_storage(moon);
}
match builder.execute(lunaris_retrieve::Query::text(q.as_str())).await {
Ok(hits) => tracing::debug!(
target: "lunaris::scoped",
hits = hits.len(),
query = %q,
"pre_warm_complete"
),
Err(e) => tracing::warn!(
target: "lunaris::scoped",
err = %e,
query = %q,
"pre_warm_failed"
),
}
});
tracing::debug!(
target: "lunaris::scoped",
query = %query_str,
"pre_warm_spawned"
);
}
Err(_) => {
tracing::debug!(
target: "lunaris::scoped",
query = %query_str,
"pre_warm_skipped_semaphore_full"
);
}
}
}
tracing::info!(
target: "lunaris::scoped",
turn_id = ?turn_id,
invalidated_count = output.invalidate.len(),
boost_count = output.boost.len(),
pre_warm_query = output.pre_warm_query.is_some(),
"scoped_end_turn_complete"
);
Ok(output)
}
pub async fn record_activation_refs(&self, signals: &[RefSignal]) -> Result<(), LunarisError> {
if signals.is_empty() {
return Ok(());
}
let now = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
let read_at = self.engine.clock.tick();
let mut order: Vec<Ulid> = Vec::new();
let mut by_id: HashMap<Ulid, Vec<RefSignal>> = HashMap::new();
for s in signals {
by_id
.entry(s.id)
.or_insert_with(|| {
order.push(s.id);
Vec::new()
})
.push(*s);
}
let mut ops: Vec<lunaris_core::WriteOp> = Vec::with_capacity(order.len());
for id in order {
let key = activation_key(&self.scope, id);
let mut record = match self
.engine
.storage
.read_as_of(&self.scope, &key, read_at)
.await
.map_err(LunarisError::Storage)?
{
Some(row) => {
serde_json::from_slice::<lunaris_core::activation::ActivationRecord>(&row.value)
.unwrap_or_else(|e| {
tracing::warn!(
err = %e,
%id,
scope = self.scope.as_str(),
"activation_ledger_corrupt_record_reseeded"
);
lunaris_core::activation::ActivationRecord::default()
})
}
None => lunaris_core::activation::ActivationRecord::default(),
};
for s in &by_id[&id] {
record.apply(s, now);
}
let value = serde_json::to_vec(&record).map_err(|e| {
LunarisError::Storage(StorageError::Backend(format!(
"activation_ledger_serialize_failed: {e}"
)))
})?;
ops.push(lunaris_core::WriteOp::KvPut { key, value });
}
self.engine.storage.atomic_write(&self.scope, &ops).await.map_err(LunarisError::Storage)?;
Ok(())
}
pub async fn archive_activation(&self, ids: &[Ulid], now: u64) -> Result<usize, LunarisError> {
if ids.is_empty() {
return Ok(0);
}
let read_at = self.engine.clock.tick();
let mut seen: HashSet<Ulid> = HashSet::new();
let mut ops: Vec<lunaris_core::WriteOp> = Vec::with_capacity(ids.len());
let mut marked = 0usize;
for &id in ids {
if !seen.insert(id) {
continue; }
let key = activation_key(&self.scope, id);
let existing = self
.engine
.storage
.read_as_of(&self.scope, &key, read_at)
.await
.map_err(LunarisError::Storage)?;
let Some(row) = existing else {
continue; };
let mut record = match serde_json::from_slice::<
lunaris_core::activation::ActivationRecord,
>(&row.value)
{
Ok(r) => r,
Err(e) => {
tracing::warn!(
err = %e,
%id,
scope = self.scope.as_str(),
"activation_ledger_corrupt_record_skipped_on_archive"
);
continue;
}
};
record.archived_at = Some(now);
let value = serde_json::to_vec(&record).map_err(|e| {
LunarisError::Storage(StorageError::Backend(format!(
"activation_ledger_serialize_failed: {e}"
)))
})?;
ops.push(lunaris_core::WriteOp::KvPut { key, value });
marked += 1;
}
if !ops.is_empty() {
self.engine
.storage
.atomic_write(&self.scope, &ops)
.await
.map_err(LunarisError::Storage)?;
}
Ok(marked)
}
pub async fn dream_agenda(
&self,
cfg: lunaris_consolidate::DreamConfig,
) -> Result<lunaris_consolidate::DreamAgenda, LunarisError> {
let now = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
lunaris_consolidate::build_dream_agenda(self.engine.storage.clone(), &self.scope, &cfg, now)
.await
}
pub async fn upsert_verify_agenda(
&self,
entries: &[VerifyAgendaEntry],
) -> Result<(), LunarisError> {
if entries.is_empty() {
return Ok(());
}
let read_at = self.engine.clock.tick();
let mut ops: Vec<lunaris_core::WriteOp> = Vec::with_capacity(entries.len());
for entry in entries {
let key = verify_agenda_key(&self.scope, entry.episode_id);
let first_seen_ms = match self
.engine
.storage
.read_as_of(&self.scope, &key, read_at)
.await
.map_err(LunarisError::Storage)?
{
Some(row) => serde_json::from_slice::<VerifyAgendaEntry>(&row.value)
.map(|existing| existing.first_seen_ms)
.unwrap_or_else(|e| {
tracing::warn!(
err = %e,
episode_id = %entry.episode_id,
scope = self.scope.as_str(),
"verify_agenda_corrupt_record_reseeded"
);
entry.first_seen_ms
}),
None => entry.first_seen_ms,
};
let mut merged = entry.clone();
merged.first_seen_ms = first_seen_ms;
let value = serde_json::to_vec(&merged).map_err(|e| {
LunarisError::Storage(StorageError::Backend(format!(
"verify_agenda_serialize_failed: {e}"
)))
})?;
ops.push(lunaris_core::WriteOp::KvPut { key, value });
}
self.engine.storage.atomic_write(&self.scope, &ops).await.map_err(LunarisError::Storage)?;
Ok(())
}
pub async fn list_verify_agenda(&self) -> Result<Vec<VerifyAgendaEntry>, LunarisError> {
use futures::stream::StreamExt;
let prefix = lunaris_core::keyspace::verify_agenda_prefix(&self.scope);
let mut stream = self
.engine
.storage
.scan_range(&self.scope, &prefix, None)
.await
.map_err(LunarisError::Storage)?;
let mut entries: Vec<VerifyAgendaEntry> = Vec::new();
let mut scanned = 0usize;
while let Some(item) = stream.next().await {
let (_key, value) = item.map_err(LunarisError::Storage)?;
scanned += 1;
match serde_json::from_slice::<VerifyAgendaEntry>(&value) {
Ok(entry) => entries.push(entry),
Err(e) => {
tracing::warn!(
err = %e,
scope = self.scope.as_str(),
"verify_agenda_list_corrupt_row_skipped"
);
}
}
if scanned >= VERIFY_AGENDA_LIST_SCAN_CAP {
tracing::warn!(
scanned,
scope = self.scope.as_str(),
"verify_agenda_list: scan cap reached — partial list"
);
break;
}
}
entries.sort_by(|a, b| b.last_seen_ms.cmp(&a.last_seen_ms));
Ok(entries)
}
pub async fn remove_verify_agenda(&self, episode_id: Ulid) -> Result<bool, LunarisError> {
let key = verify_agenda_key(&self.scope, episode_id);
let read_at = self.engine.clock.tick();
let existed = self
.engine
.storage
.read_as_of(&self.scope, &key, read_at)
.await
.map_err(LunarisError::Storage)?
.is_some();
if existed {
let ops = vec![lunaris_core::WriteOp::KvDelete { key }];
self.engine
.storage
.atomic_write(&self.scope, &ops)
.await
.map_err(LunarisError::Storage)?;
}
Ok(existed)
}
}
const VERIFY_AGENDA_LIST_SCAN_CAP: usize = 5_000;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VerifyAgendaEntry {
pub episode_id: Ulid,
pub anchor_head: String,
pub current_head: String,
pub files: Vec<String>,
pub first_seen_ms: u64,
pub last_seen_ms: u64,
pub v: u32,
}
pub const EMBEDDER_DIR_ENV_VAR: &str = "LUNARIS_EMBEDDER_DIR";
pub const RERANKER_DIR_ENV_VAR: &str = "LUNARIS_RERANKER_DIR";
pub const EMBEDDER_GGUF_ENV_VAR: &str = "LUNARIS_EMBEDDER_GGUF";
pub const RERANKER_GGUF_ENV_VAR: &str = "LUNARIS_RERANKER_GGUF";
pub const EMBED_DIM_ENV_VAR: &str = "LUNARIS_EMBED_DIM";
pub const PREWARM_CONCURRENCY_ENV_VAR: &str = "LUNARIS_PREWARM_CONCURRENCY";
const PREWARM_CONCURRENCY_DEFAULT: usize = 4;
pub const EMBED_CACHE_CAPACITY_ENV_VAR: &str = "LUNARIS_EMBED_CACHE_CAPACITY";
const EMBED_CACHE_CAPACITY_DEFAULT: usize = 2048;
fn resolve_prewarm_concurrency() -> usize {
static LOG_ONCE: OnceLock<()> = OnceLock::new();
let capacity = match std::env::var(PREWARM_CONCURRENCY_ENV_VAR).ok().as_deref() {
None | Some("") => PREWARM_CONCURRENCY_DEFAULT,
Some(s) => match s.trim().parse::<usize>() {
Ok(0) => {
tracing::warn!(
env = PREWARM_CONCURRENCY_ENV_VAR,
value = s,
default = PREWARM_CONCURRENCY_DEFAULT,
"LUNARIS_PREWARM_CONCURRENCY=0 is invalid (would skip all warm-ups); \
using default"
);
PREWARM_CONCURRENCY_DEFAULT
}
Ok(n) => n,
Err(_) => {
tracing::warn!(
env = PREWARM_CONCURRENCY_ENV_VAR,
value = s,
default = PREWARM_CONCURRENCY_DEFAULT,
"LUNARIS_PREWARM_CONCURRENCY is not a valid positive integer; using default"
);
PREWARM_CONCURRENCY_DEFAULT
}
},
};
LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
prewarm_concurrency = capacity,
"prewarm_concurrency_resolved"
);
});
capacity
}
fn embed_cache_capacity() -> Option<NonZeroUsize> {
let capacity = match std::env::var(EMBED_CACHE_CAPACITY_ENV_VAR).ok().as_deref() {
None | Some("") => EMBED_CACHE_CAPACITY_DEFAULT,
Some("0") => return None,
Some(raw) => match raw.trim().parse::<usize>() {
Ok(0) => return None,
Ok(n) => n,
Err(_) => {
tracing::warn!(
env = EMBED_CACHE_CAPACITY_ENV_VAR,
value = raw,
default = EMBED_CACHE_CAPACITY_DEFAULT,
"LUNARIS_EMBED_CACHE_CAPACITY is not a valid non-negative integer; using default"
);
EMBED_CACHE_CAPACITY_DEFAULT
}
},
};
NonZeroUsize::new(capacity)
}
fn maybe_cached_embedder(embedder: Arc<dyn Embedder>) -> Arc<dyn Embedder> {
match embed_cache_capacity() {
Some(capacity) => Arc::new(CachedEmbedder::new(embedder, capacity)) as Arc<dyn Embedder>,
None => embedder,
}
}
static EMBEDDER_BACKEND_LOG_ONCE: OnceLock<()> = OnceLock::new();
static RERANKER_BACKEND_LOG_ONCE: OnceLock<()> = OnceLock::new();
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum EmbedderBackend {
LlamaCpp,
OpenAiRemote,
OllamaRemote,
Noop,
Unresolved,
}
impl EmbedderBackend {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
EmbedderBackend::LlamaCpp => "llamacpp",
EmbedderBackend::OpenAiRemote => "openai-remote",
EmbedderBackend::OllamaRemote => "ollama-remote",
EmbedderBackend::Noop => "noop",
EmbedderBackend::Unresolved => "unresolved",
}
}
#[must_use]
pub const fn produces_real_vectors(self) -> bool {
!matches!(self, EmbedderBackend::Noop)
}
}
impl std::fmt::Display for EmbedderBackend {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
static RESOLVED_EMBEDDER_BACKEND: OnceLock<EmbedderBackend> = OnceLock::new();
static DEGRADATION_ANNOUNCED: OnceLock<()> = OnceLock::new();
pub const SUPPRESS_DEGRADED_WARNING_ENV: &str = "LUNARIS_SUPPRESS_DEGRADED_WARNING";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DegradationNotice {
Silent,
Emit(String),
}
#[must_use]
pub fn degradation_notice(
backend: EmbedderBackend,
subscriber_installed: bool,
suppress_env: Option<&str>,
) -> DegradationNotice {
if backend.produces_real_vectors() || subscriber_installed {
return DegradationNotice::Silent;
}
if let Some(raw) = suppress_env {
let v = raw.trim().to_ascii_lowercase();
if matches!(v.as_str(), "1" | "true" | "yes" | "on") {
return DegradationNotice::Silent;
}
}
DegradationNotice::Emit(format!(
"lunaris: WARNING — embedder backend is '{backend}': every vector is zeros, \
so semantic recall silently degrades to keyword-only while every call keeps \
succeeding. Stage a GGUF (LUNARIS_EMBEDDER_GGUF, or ~/.lunaris/models/) or \
configure a remote embedder (LUNARIS_EMBEDDER_OPENAI_URL / \
LUNARIS_EMBEDDER_OLLAMA_URL). Query it with embedder_backend(); silence this \
with {SUPPRESS_DEGRADED_WARNING_ENV}=1."
))
}
fn announce_degradation_once() {
if DEGRADATION_ANNOUNCED.get().is_some() {
return;
}
let notice = degradation_notice(
resolved_embedder_backend(),
tracing::dispatcher::has_been_set(),
std::env::var(SUPPRESS_DEGRADED_WARNING_ENV).ok().as_deref(),
);
if let DegradationNotice::Emit(msg) = notice {
if DEGRADATION_ANNOUNCED.set(()).is_ok() {
eprintln!("{msg}");
}
}
}
#[must_use]
pub fn resolved_embedder_backend() -> EmbedderBackend {
RESOLVED_EMBEDDER_BACKEND.get().copied().unwrap_or(EmbedderBackend::Unresolved)
}
const GRANITE_R2_DIR: &str = "granite-embedding-311m-multilingual-r2";
#[cfg(test)]
const BGE_RERANKER_DIR: &str = "bge-reranker-v2-m3";
fn default_model_dir(name: &str) -> std::path::PathBuf {
dirs::cache_dir()
.unwrap_or_else(|| std::path::PathBuf::from("."))
.join("lunaris")
.join("models")
.join(name)
}
fn embedder_dir() -> std::path::PathBuf {
std::env::var(EMBEDDER_DIR_ENV_VAR)
.ok()
.filter(|s| !s.trim().is_empty())
.map(std::path::PathBuf::from)
.unwrap_or_else(|| default_model_dir(GRANITE_R2_DIR))
}
#[cfg_attr(not(feature = "llamacpp"), allow(unused_variables))]
async fn resolve_embedder(max_batch_tokens: u32) -> Result<Arc<dyn Embedder>, LunarisError> {
#[cfg(feature = "llamacpp")]
{
if let Some(gguf_path) = llamacpp_gguf_path(EMBEDDER_GGUF_ENV_VAR, LLAMACPP_EMBEDDER_MODEL)
{
let opts = lunaris_llamacpp::LlamaCppEmbedderOpts {
gguf_path: gguf_path.clone(),
n_gpu_layers: llamacpp_gpu_layers(),
max_batch_tokens,
..Default::default()
};
let opened =
tokio::task::spawn_blocking(move || lunaris_llamacpp::LlamaCppEmbedder::open(opts))
.await
.map_err(|e| {
LunarisError::Storage(lunaris_core::StorageError::Backend(format!(
"llamacpp embedder init join: {e}"
)))
})?;
match opened {
Ok(e) => {
let _ = RESOLVED_EMBEDDER_BACKEND.set(EmbedderBackend::LlamaCpp);
EMBEDDER_BACKEND_LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
embedder_backend = "llamacpp",
gguf = %gguf_path.display(),
"embedder_backend_resolved"
);
});
return Ok(Arc::new(e) as Arc<dyn Embedder>);
}
Err(err) => {
tracing::warn!(
error = %err,
gguf = %gguf_path.display(),
"llamacpp embedder failed to open; falling through to the \
remote/Noop chain"
);
}
}
}
}
#[cfg(feature = "embed-remote")]
{
if std::env::var(lunaris_embed_remote::openai::OPENAI_URL_ENV_VAR)
.ok()
.filter(|s| !s.trim().is_empty())
.is_some()
{
let opts = lunaris_embed_remote::openai::OpenAiEmbedderOpts::default();
let e = lunaris_embed_remote::openai::OpenAiEmbedder::new(opts)?;
let _ = RESOLVED_EMBEDDER_BACKEND.set(EmbedderBackend::OpenAiRemote);
EMBEDDER_BACKEND_LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
embedder_backend = "openai-remote",
"embedder_backend_resolved (remote OpenAI-compatible /embeddings)"
);
});
return Ok(Arc::new(e) as Arc<dyn Embedder>);
}
}
#[cfg(feature = "embed-remote")]
{
if let Some(url) =
std::env::var(lunaris_embed_remote::OLLAMA_URL_ENV_VAR).ok().filter(|s| !s.is_empty())
{
let opts =
lunaris_embed_remote::OllamaEmbedderOpts { endpoint: url, ..Default::default() };
let e = lunaris_embed_remote::OllamaEmbedder::new(opts)?;
let _ = RESOLVED_EMBEDDER_BACKEND.set(EmbedderBackend::OllamaRemote);
EMBEDDER_BACKEND_LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
embedder_backend = "ollama-remote",
"embedder_backend_resolved (operator escape hatch)"
);
});
return Ok(Arc::new(e) as Arc<dyn Embedder>);
}
}
let dim = resolve_embed_dim();
let _ = RESOLVED_EMBEDDER_BACKEND.set(EmbedderBackend::Noop);
EMBEDDER_BACKEND_LOG_ONCE.get_or_init(|| {
tracing::warn!(
target: "lunaris::handle",
fallback_dim = dim,
"no embedder backend available — using NoopEmbedder (zero vectors). \
Stage the llama.cpp GGUF (LUNARIS_EMBEDDER_GGUF or ~/.lunaris/models/) \
or configure a remote embedder (--features embed-remote + \
LUNARIS_EMBEDDER_OPENAI_URL / LUNARIS_EMBEDDER_OLLAMA_URL) for real vectors."
);
});
Ok(Arc::new(lunaris_core::NoopEmbedder::new(dim)) as Arc<dyn Embedder>)
}
fn parse_batch_tokens(raw: Option<String>, default: u32) -> u32 {
raw.and_then(|s| s.trim().parse::<u32>().ok()).filter(|&n| n >= 16).unwrap_or(default)
}
fn embed_max_batch_tokens() -> u32 {
parse_batch_tokens(std::env::var("LUNARIS_EMBED_MAX_BATCH_TOKENS").ok(), 4096)
}
fn context_embed_max_batch_tokens() -> u32 {
parse_batch_tokens(std::env::var("LUNARIS_CONTEXT_EMBED_MAX_BATCH_TOKENS").ok(), 1024)
}
pub async fn resolve_default_embedder() -> Result<Arc<dyn Embedder>, LunarisError> {
resolve_embedder(context_embed_max_batch_tokens()).await
}
pub fn lazy_default_embedder() -> Arc<dyn Embedder> {
Arc::new(LazyDefaultEmbedder { cell: tokio::sync::OnceCell::new() })
}
struct LazyDefaultEmbedder {
cell: tokio::sync::OnceCell<Arc<dyn Embedder>>,
}
impl LazyDefaultEmbedder {
async fn get_or_load(&self) -> Result<&Arc<dyn Embedder>, LunarisError> {
self.cell.get_or_try_init(resolve_default_embedder).await
}
}
impl std::fmt::Debug for LazyDefaultEmbedder {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LazyDefaultEmbedder").field("loaded", &self.cell.initialized()).finish()
}
}
#[async_trait::async_trait]
impl Embedder for LazyDefaultEmbedder {
fn dim(&self) -> usize {
self.cell.get().map(|e| e.dim()).unwrap_or_else(resolve_embed_dim)
}
async fn embed_batch(&self, inputs: &[&str]) -> Result<Vec<Vec<f32>>, LunarisError> {
self.get_or_load().await?.embed_batch(inputs).await
}
async fn embed_batch_lowpri(&self, inputs: &[&str]) -> Result<Vec<Vec<f32>>, LunarisError> {
self.get_or_load().await?.embed_batch_lowpri(inputs).await
}
}
fn resolve_embed_dim() -> usize {
static LOG_ONCE: OnceLock<()> = OnceLock::new();
let dim = match std::env::var(EMBED_DIM_ENV_VAR).ok().as_deref() {
None | Some("") => lunaris_core::NOOP_DEFAULT_DIM,
Some(s) => match s.trim().parse::<usize>() {
Ok(0) => {
tracing::warn!(
env = EMBED_DIM_ENV_VAR,
value = s,
default = lunaris_core::NOOP_DEFAULT_DIM,
"LUNARIS_EMBED_DIM=0 is invalid (storage rejects dim=0); using default"
);
lunaris_core::NOOP_DEFAULT_DIM
}
Ok(n) => n,
Err(_) => {
tracing::warn!(
env = EMBED_DIM_ENV_VAR,
value = s,
default = lunaris_core::NOOP_DEFAULT_DIM,
"LUNARIS_EMBED_DIM is not a valid positive integer; using default"
);
lunaris_core::NOOP_DEFAULT_DIM
}
},
};
LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
embed_dim = dim,
"embed_dim_resolved"
);
});
dim
}
async fn resolve_reranker() -> Result<Arc<dyn Reranker>, LunarisError> {
#[cfg(feature = "llamacpp")]
{
if let Some(gguf_path) = llamacpp_gguf_path(RERANKER_GGUF_ENV_VAR, LLAMACPP_RERANKER_MODEL)
{
let lazy = LazyLlamaCppReranker::new(lunaris_llamacpp::LlamaCppRerankerOpts {
gguf_path: gguf_path.clone(),
n_gpu_layers: llamacpp_gpu_layers(),
..Default::default()
});
RERANKER_BACKEND_LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
reranker_backend = "llamacpp (lazy)",
gguf = %gguf_path.display(),
"reranker_backend_resolved (load deferred to first rerank())"
);
});
return Ok(Arc::new(lazy) as Arc<dyn Reranker>);
}
}
RERANKER_BACKEND_LOG_ONCE.get_or_init(|| {
tracing::info!(
target: "lunaris::handle",
reranker_backend = "noop",
"no reranker backend available — using NoopReranker (rerank pass skipped \
per RETRIEVE-06 contract). Stage the llama.cpp GGUF \
(LUNARIS_RERANKER_GGUF or ~/.lunaris/models/) for a real reranker."
);
});
Ok(Arc::new(NoopReranker) as Arc<dyn Reranker>)
}
pub async fn resolve_default_reranker() -> Result<Arc<dyn Reranker>, LunarisError> {
resolve_reranker().await
}
#[cfg(feature = "cloud-api")]
fn remote_extractor_from_env() -> Option<Arc<dyn Extractor>> {
let raw = std::env::var("LUNARIS_EXTRACT_PROVIDER").ok()?;
if raw.trim().is_empty() {
return None;
}
let opts = lunaris_extract::CloudApiExtractorOpts::default();
let label = opts.model.clone();
match lunaris_extract::CloudApiExtractor::new(opts) {
Ok(e) => {
tracing::info!(
target: "lunaris::handle",
provider = %raw.trim(),
model = %label,
"extractor_backend_resolved (remote cloud-api)"
);
Some(lunaris_extract::fallback::fallback_wrap(e, &label))
}
Err(e) => {
tracing::warn!(
error = %e,
provider = %raw.trim(),
"LUNARIS_EXTRACT_PROVIDER set but the remote extractor failed to construct; \
graph extraction disabled (NoopExtractor) until the config is fixed"
);
Some(Arc::new(NoopExtractor) as Arc<dyn Extractor>)
}
}
}
#[cfg(feature = "cloud-api")]
fn remote_verifier_from_env() -> Option<Arc<dyn Verifier>> {
let raw = std::env::var(lunaris_verify::cloud_api::ENV_PROVIDER).ok()?;
if raw.trim().is_empty() {
return None;
}
let opts = lunaris_verify::CloudApiVerifierOpts::default();
match lunaris_verify::CloudApiVerifier::new(opts) {
Ok(v) => {
tracing::info!(
target: "lunaris::handle",
provider = %raw.trim(),
"verifier_backend_resolved (remote cloud-api)"
);
Some(Arc::new(v) as Arc<dyn Verifier>)
}
Err(e) => {
tracing::warn!(
error = %e,
provider = %raw.trim(),
"LUNARIS_VERIFY_PROVIDER set but the remote verifier failed to construct; \
verification disabled (NoopVerifier) until the config is fixed"
);
Some(Arc::new(NoopVerifier) as Arc<dyn Verifier>)
}
}
}
async fn default_extractor() -> Arc<dyn Extractor> {
#[cfg(feature = "cloud-api")]
if let Some(e) = remote_extractor_from_env() {
return e;
}
Arc::new(NoopExtractor) as Arc<dyn Extractor>
}
async fn default_verifier() -> Arc<dyn Verifier> {
#[cfg(feature = "cloud-api")]
if let Some(v) = remote_verifier_from_env() {
return v;
}
Arc::new(NoopVerifier) as Arc<dyn Verifier>
}
fn default_consolidator() -> Result<Arc<dyn Consolidator>, LunarisError> {
ConsolidatorPipelineHandle::backend_from_env()
}
#[cfg(feature = "llamacpp")]
const LLAMACPP_EMBEDDER_MODEL: lunaris_core::models::ModelKind =
lunaris_core::models::ModelKind::EmbedderGraniteQ4KM;
#[cfg(feature = "llamacpp")]
const LLAMACPP_RERANKER_MODEL: lunaris_core::models::ModelKind =
lunaris_core::models::ModelKind::RerankerBgeV2M3Q5KM;
#[cfg(feature = "llamacpp")]
fn llamacpp_gguf_path(
env_var: &str,
kind: lunaris_core::models::ModelKind,
) -> Option<std::path::PathBuf> {
if let Some(p) = std::env::var_os(env_var)
.map(std::path::PathBuf::from)
.filter(|p| !p.as_os_str().is_empty())
{
if p.exists() {
return Some(p);
}
tracing::warn!(
env = env_var,
path = %p.display(),
"GGUF path from env does not exist; falling through"
);
return None;
}
lunaris_core::models::staged_path(kind).filter(|p| p.exists())
}
#[cfg(feature = "llamacpp")]
fn llamacpp_gpu_layers() -> u32 {
let forced_cpu = std::env::var("LUNARIS_DEVICE")
.map(|v| v.trim().eq_ignore_ascii_case("cpu"))
.unwrap_or(false);
if !forced_cpu && cfg!(feature = "metal") { u32::MAX } else { 0 }
}
#[cfg(feature = "llamacpp")]
struct LazyLlamaCppReranker {
opts: lunaris_llamacpp::LlamaCppRerankerOpts,
cell: tokio::sync::OnceCell<Arc<lunaris_llamacpp::LlamaCppReranker>>,
}
#[cfg(feature = "llamacpp")]
impl LazyLlamaCppReranker {
fn new(opts: lunaris_llamacpp::LlamaCppRerankerOpts) -> Self {
Self { opts, cell: tokio::sync::OnceCell::new() }
}
async fn get_or_load(&self) -> Result<Arc<lunaris_llamacpp::LlamaCppReranker>, LunarisError> {
let opts = self.opts.clone();
self.cell
.get_or_try_init(|| async move {
tokio::task::spawn_blocking(move || lunaris_llamacpp::LlamaCppReranker::open(opts))
.await
.map_err(|e| {
LunarisError::Storage(lunaris_core::StorageError::Backend(format!(
"lazy llamacpp reranker init join: {e}"
)))
})?
.map(Arc::new)
.map_err(LunarisError::from)
})
.await
.cloned()
}
}
#[cfg(feature = "llamacpp")]
impl std::fmt::Debug for LazyLlamaCppReranker {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LazyLlamaCppReranker")
.field("gguf", &self.opts.gguf_path)
.field("loaded", &self.cell.initialized())
.finish()
}
}
#[cfg(feature = "llamacpp")]
#[async_trait::async_trait]
impl Reranker for LazyLlamaCppReranker {
fn applies(&self) -> bool {
true
}
async fn rerank(
&self,
query: &str,
docs: Vec<lunaris_rerank::RerankCandidate>,
) -> Result<Vec<lunaris_rerank::RerankCandidate>, LunarisError> {
let inner = self.get_or_load().await?;
inner.rerank(query, docs).await
}
}
#[cfg(test)]
mod backend_resolution_tests {
use super::*;
use lunaris_core::StubEmbedder;
struct CountingEmbedder {
inner: StubEmbedder,
calls: Arc<AtomicUsize>,
}
#[async_trait::async_trait]
impl Embedder for CountingEmbedder {
fn dim(&self) -> usize {
self.inner.dim()
}
async fn embed_batch(&self, inputs: &[&str]) -> Result<Vec<Vec<f32>>, LunarisError> {
self.calls.fetch_add(1, Ordering::Relaxed);
self.inner.embed_batch(inputs).await
}
}
#[test]
fn default_model_dir_layout_is_canonical() {
let p = default_model_dir(GRANITE_R2_DIR);
assert!(
p.ends_with("lunaris/models/granite-embedding-311m-multilingual-r2"),
"default granite-r2 dir was: {}",
p.display()
);
let p = default_model_dir(BGE_RERANKER_DIR);
assert!(
p.ends_with("lunaris/models/bge-reranker-v2-m3"),
"default bge dir was: {}",
p.display()
);
}
#[test]
fn env_var_constants_are_grep_pinned() {
assert_eq!(EMBEDDER_DIR_ENV_VAR, "LUNARIS_EMBEDDER_DIR");
assert_eq!(RERANKER_DIR_ENV_VAR, "LUNARIS_RERANKER_DIR");
assert_eq!(EMBEDDER_GGUF_ENV_VAR, "LUNARIS_EMBEDDER_GGUF");
assert_eq!(RERANKER_GGUF_ENV_VAR, "LUNARIS_RERANKER_GGUF");
assert_eq!(EMBED_DIM_ENV_VAR, "LUNARIS_EMBED_DIM");
}
#[tokio::test]
async fn cached_embedder_dedupes_batch_and_reuses_later_hits() {
let calls = Arc::new(AtomicUsize::new(0));
let inner = Arc::new(CountingEmbedder { inner: StubEmbedder::new(8), calls: calls.clone() })
as Arc<dyn Embedder>;
let cached = CachedEmbedder::new(inner, NonZeroUsize::new(8).unwrap());
let first = cached.embed_batch(&["alpha", "alpha", "beta"]).await.unwrap();
assert_eq!(first.len(), 3);
assert_eq!(calls.load(Ordering::Relaxed), 1, "first batch should dedupe misses");
assert_eq!(first[0], first[1]);
let second = cached.embed_batch(&["beta", "alpha"]).await.unwrap();
assert_eq!(second.len(), 2);
assert_eq!(
calls.load(Ordering::Relaxed),
1,
"second batch should be served entirely from cache"
);
}
#[test]
fn parse_batch_tokens_defaults_when_absent() {
assert_eq!(parse_batch_tokens(None, 1024), 1024);
assert_eq!(parse_batch_tokens(None, 4096), 4096);
}
#[test]
fn parse_batch_tokens_honors_numeric_override() {
assert_eq!(parse_batch_tokens(Some("2048".to_owned()), 1024), 2048);
assert_eq!(parse_batch_tokens(Some(" 512 ".to_owned()), 1024), 512);
}
#[test]
fn parse_batch_tokens_falls_back_on_bogus() {
assert_eq!(parse_batch_tokens(Some(String::new()), 1024), 1024, "empty -> default");
assert_eq!(
parse_batch_tokens(Some("abc".to_owned()), 1024),
1024,
"non-numeric -> default"
);
assert_eq!(parse_batch_tokens(Some("8".to_owned()), 1024), 1024, "below the >=16 floor");
}
#[test]
fn context_budget_is_1024_general_is_4096_by_default() {
assert_eq!(context_embed_max_batch_tokens(), 1024);
assert_eq!(embed_max_batch_tokens(), 4096);
}
}
#[cfg(test)]
mod end_turn_tests {
use super::*;
use async_trait::async_trait;
use bytes::Bytes;
use futures::stream::{self, BoxStream};
use lunaris_core::storage::keyword::{KeywordHit, KeywordPort};
use lunaris_core::storage::types::{
CypherQuery, Filter, GraphResult, Lsn, QueueMsg, Row, VectorHit, WriteOp,
};
use lunaris_core::{
CypherDialect, HlcClock, LunarisError, Scope, StorageCapabilities, StorageError,
StoragePort, StubEmbedder,
};
use lunaris_verify::{ReflectInput, ReflectOutput, ReflectSupervisor};
use std::sync::Arc;
use ulid::Ulid;
struct NullStorage;
#[async_trait]
impl StoragePort for NullStorage {
async fn atomic_write(
&self,
_scope: &Scope,
_ops: &[WriteOp],
) -> Result<Lsn, StorageError> {
Ok(Lsn { wall_ms: 1, counter: 0 })
}
async fn read_as_of(
&self,
_scope: &Scope,
_key: &[u8],
_as_of: lunaris_core::Hlc,
) -> Result<Option<Row<Bytes>>, StorageError> {
Ok(None)
}
async fn vector_search(
&self,
_scope: &Scope,
_index: &str,
_query: &[f32],
_k: usize,
_filter: Option<&Filter>,
_as_of: Option<lunaris_core::Hlc>,
_rerank: bool,
) -> Result<Vec<VectorHit>, StorageError> {
Ok(vec![])
}
async fn graph_traverse(
&self,
_scope: &Scope,
_q: &CypherQuery,
_as_of: Option<lunaris_core::Hlc>,
) -> Result<GraphResult, StorageError> {
Ok(GraphResult::default())
}
async fn scan_range(
&self,
_scope: &Scope,
_prefix: &[u8],
_as_of: Option<lunaris_core::Hlc>,
) -> Result<BoxStream<'_, Result<(Bytes, Bytes), StorageError>>, StorageError> {
Ok(Box::pin(stream::iter(Vec::<Result<(Bytes, Bytes), StorageError>>::new())))
}
async fn publish(
&self,
_scope: &Scope,
_topic: &str,
_partition: u16,
_payload: Bytes,
) -> Result<u64, StorageError> {
Ok(0)
}
async fn subscribe(
&self,
_scope: &Scope,
_group: &str,
_topic: &str,
_partition: u16,
) -> Result<BoxStream<'static, Result<QueueMsg, StorageError>>, StorageError> {
Ok(Box::pin(stream::empty()))
}
fn capabilities(&self) -> StorageCapabilities {
StorageCapabilities {
bi_temporal_native: false,
graph_native: false,
rerank_native: false,
queue_native: false,
max_vector_dim: 768,
native_rrf: false,
max_scopes_recommended: 0,
cypher_dialect: CypherDialect::Legacy,
graph_decay_native: false,
graph_navigate_native: false,
}
}
}
#[async_trait]
impl KeywordPort for NullStorage {
async fn keyword_search(
&self,
_scope: &Scope,
_index: &str,
_query: &str,
_k: usize,
_filter: Option<&Filter>,
_as_of: Option<lunaris_core::Hlc>,
) -> Result<Vec<KeywordHit>, StorageError> {
Ok(vec![])
}
}
fn make_handle() -> Lunaris {
let storage: Arc<dyn StoragePort> = Arc::new(NullStorage);
let keyword: Arc<dyn KeywordPort> = Arc::new(NullStorage);
let embedder = Arc::new(StubEmbedder::new(4));
let clock = HlcClock::new(0);
Lunaris::with_parts_keyword(storage, keyword, embedder, clock)
}
#[tokio::test]
async fn end_turn_noop_returns_empty_output() {
let handle = make_handle();
assert!(!handle.reflect_supervisor().applies());
let input = ReflectInput {
turn_id: Some(Ulid::new()),
turn_summary: "agent answered a question".into(),
recent_fact_ids: vec![Ulid::new()],
recent_chunk_ids: vec![Ulid::new()],
};
let out = handle.end_turn(input).await.unwrap();
assert_eq!(out, ReflectOutput::default());
assert!(out.invalidate.is_empty());
assert!(out.boost.is_empty());
assert!(out.pre_warm_query.is_none());
}
struct CapturingReflectSupervisor {
output: ReflectOutput,
captured: parking_lot::Mutex<Option<ReflectInput>>,
}
#[async_trait]
impl ReflectSupervisor for CapturingReflectSupervisor {
async fn reflect(&self, input: ReflectInput) -> Result<ReflectOutput, LunarisError> {
*self.captured.lock() = Some(input);
Ok(self.output.clone())
}
fn applies(&self) -> bool {
true
}
}
#[tokio::test]
async fn end_turn_stub_supervisor_propagates_output_and_input() {
let fact_id = Ulid::new();
let chunk_id = Ulid::new();
let turn_id = Ulid::new();
let expected_output = ReflectOutput {
invalidate: vec![fact_id],
boost: vec![chunk_id],
pre_warm_query: Some("what is Alice's role?".into()),
};
let supervisor = Arc::new(CapturingReflectSupervisor {
output: expected_output.clone(),
captured: parking_lot::Mutex::new(None),
});
let handle = make_handle().with_reflect_supervisor(supervisor.clone());
assert!(handle.reflect_supervisor().applies());
let input = ReflectInput {
turn_id: Some(turn_id),
turn_summary: "turn summary text".into(),
recent_fact_ids: vec![fact_id],
recent_chunk_ids: vec![chunk_id],
};
let out = handle.end_turn(input).await.unwrap();
assert_eq!(out, expected_output);
let captured = supervisor.captured.lock().take().unwrap();
assert_eq!(captured.turn_id, Some(turn_id));
assert_eq!(captured.recent_fact_ids, vec![fact_id]);
assert_eq!(captured.recent_chunk_ids, vec![chunk_id]);
assert_eq!(captured.turn_summary, "turn summary text");
}
struct ErrReflectSupervisor;
#[async_trait]
impl ReflectSupervisor for ErrReflectSupervisor {
async fn reflect(&self, _input: ReflectInput) -> Result<ReflectOutput, LunarisError> {
Err(LunarisError::Storage(StorageError::NotSupported("reflect budget exhausted")))
}
}
#[tokio::test]
async fn end_turn_propagates_supervisor_error() {
let handle = make_handle().with_reflect_supervisor(Arc::new(ErrReflectSupervisor));
let result = handle.end_turn(ReflectInput::default()).await;
assert!(result.is_err(), "end_turn must propagate supervisor error");
let msg = format!("{}", result.unwrap_err());
assert!(msg.contains("reflect budget exhausted"), "error message: {msg}");
}
}
#[cfg(test)]
mod embedder_backend_visibility_tests {
use super::{EmbedderBackend, resolved_embedder_backend};
#[test]
fn every_backend_has_a_stable_lowercase_tag() {
let all = [
(EmbedderBackend::LlamaCpp, "llamacpp"),
(EmbedderBackend::OpenAiRemote, "openai-remote"),
(EmbedderBackend::OllamaRemote, "ollama-remote"),
(EmbedderBackend::Noop, "noop"),
(EmbedderBackend::Unresolved, "unresolved"),
];
for (backend, tag) in all {
assert_eq!(
backend.as_str(),
tag,
"{backend:?} tag changed — that is a breaking change"
);
assert_eq!(backend.to_string(), tag, "Display must agree with as_str for {backend:?}");
}
}
#[test]
fn tags_are_distinct() {
let tags = [
EmbedderBackend::LlamaCpp,
EmbedderBackend::OpenAiRemote,
EmbedderBackend::OllamaRemote,
EmbedderBackend::Noop,
EmbedderBackend::Unresolved,
]
.map(EmbedderBackend::as_str);
let unique: std::collections::BTreeSet<_> = tags.iter().collect();
assert_eq!(unique.len(), tags.len(), "two backends share a tag: {tags:?}");
}
#[test]
fn only_noop_is_reported_as_not_producing_real_vectors() {
assert!(!EmbedderBackend::Noop.produces_real_vectors());
for ok in [
EmbedderBackend::LlamaCpp,
EmbedderBackend::OpenAiRemote,
EmbedderBackend::OllamaRemote,
EmbedderBackend::Unresolved,
] {
assert!(ok.produces_real_vectors(), "{ok:?} must not read as degraded");
}
}
#[test]
fn the_handle_accessor_agrees_with_the_process_resolver() {
assert_eq!(resolved_embedder_backend().as_str(), resolved_embedder_backend().to_string());
}
}