lunaris/

handle.rs

//! `Lunaris` — the high-level memory-engine handle (Phase 2 surface).
//!
//! Wraps `Arc<dyn StoragePort> + Arc<dyn Embedder> + Arc<HlcClock>` so callers
//! can construct multiple instances against different URLs (originally for the
//! Moon-vs-Postgres benches; since 0.7.0 that means several independent Moons).
//! All three fields are `Arc`-shared so `Lunaris::clone()` is cheap and
//! `Lunaris` is `Send + Sync` for free.
//!
//! ## Construction paths
//!
//! - [`Lunaris::open`] — production constructor. Routes the `url` through the
//!   Phase 1 [`crate::open::open`] dispatcher to pick a [`StoragePort`] backend,
//!   constructs the default embedder (llama.cpp Q4_K_M granite-r2 GGUF —
//!   llama.cpp-only cutover) and a fresh `HlcClock(node_id=0)`.
//! - [`Lunaris::with_parts`] — escape hatch for tests + the Plan 02-01
//!   latency-budget swap. Lets callers wire any `Arc<dyn StoragePort>` and
//!   `Arc<dyn Embedder>` directly. Used by the Phase 2 ingest smoke test
//!   (in-memory recording storage + `StubEmbedder`).
//! - [`Lunaris::with_embedder`] — public escape hatch to replace the
//!   embedder on an already-constructed handle (e.g., swap to the
//!   feature-gated `lunaris_embed_remote::OllamaEmbedder`
//!   or to a BYO `Arc<dyn Embedder>`).
//!
//! ## Invariant
//!
//! `Lunaris` does NOT cache mutable per-call retrieval state. Every call
//! constructs a fresh borrow of the shared Arcs, so the same handle is safe to
//! use from multiple tokio tasks concurrently. The production constructor wraps
//! the embedder in a small exact-text LRU cache so repeated agent prompts and
//! repeated chunk text do not re-run model inference.

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,
};
// ADD task activation-ledger — persistent per-memory activation ledger types.
use lunaris_core::activation::RefSignal;
use lunaris_core::keyspace::activation_key;
// engram-soul-loop task 6 (staleness-pass) — verify-agenda keyspace helper.
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>,
    /// Concrete `MoonStorage` Arc when the handle was opened against a `moon://` URL.
    /// Plan 02-02's `fuse_rrf` Moon-native dispatch reads this to opt into the
    /// one-round-trip `text().hybrid_search()` path. `None` for a handle built
    /// via `with_parts*` from a custom or decorated `StoragePort` — including a
    /// test double wrapping a live Moon, which is exactly how the client-side
    /// fusion path is still exercised now that no second backend exists.
    pub(crate) moon_storage: Option<Arc<MoonStorage>>,
    /// Plan 02-03: cross-encoder reranker for the recall hot path.
    /// Defaults to `BgeRerankerV2M3` when `~/.cache/lunaris/models/bge-reranker-v2-m3/`
    /// is present; falls back to `NoopReranker` per RETRIEVE-06 contract when
    /// the cache is missing. Callers swap via `with_reranker(reranker)`.
    pub(crate) reranker: Arc<dyn Reranker>,
    /// GA-1 — opt-in rerank stage on the production recall root. Read ONCE
    /// from `LUNARIS_RECALL_RERANK` / `LUNARIS_RECALL_RERANK_TOP_IN` at
    /// `open*` construction (default OFF; the `with_parts*` test seams stay
    /// OFF like the graph pipeline's hardcoded `false`). Accessor + escape
    /// hatch live in `crate::recall_rerank`.
    pub(crate) recall_rerank: crate::recall_rerank::RecallRerankConfig,
    /// Plan 03-03: graph extraction pipeline toggle (D-10/D-11). Default OFF.
    /// The `Extractor` itself lives INSIDE the handle's
    /// `RwLock<Option<Arc<dyn Extractor>>>` — callers `swap` via
    /// [`Self::with_extractor`] which delegates to
    /// [`GraphPipelineHandle::set_extractor`]; toggle ON/OFF via
    /// `handle.graph_pipeline().enable() / .disable()` (D-10 single-switch
    /// surface, EXTRACT-06).
    pub(crate) graph_pipeline: Arc<GraphPipelineHandle>,
    /// Plan 04-04: slow-path Verifier worker toggle (D-08, default OFF per
    /// blueprint §5.1). Owns the `Arc<dyn Verifier>`, the late-bound
    /// `Arc<dyn StoragePort>`, the worker JoinHandle, and the shutdown
    /// `tokio::sync::Notify`. Toggle ON/OFF via
    /// `handle.verify_pipeline().enable() / .disable()` (D-08 single-switch
    /// surface, VERIFY-01..06).
    pub(crate) verify_pipeline: Arc<VerifierPipelineHandle>,
    /// Plan 04-04: ACT-R Consolidator worker toggle (D-08, default OFF per
    /// blueprint §5.1). Same shape as `verify_pipeline`. Toggle ON/OFF via
    /// `handle.consolidator_pipeline().enable() / .disable()` (D-08
    /// single-switch surface, CONSOL-01..05).
    pub(crate) consolidator_pipeline: Arc<ConsolidatorPipelineHandle>,
    /// Phase 13 — per-turn reflection supervisor. Default OFF
    /// (`NoopReflectSupervisor`), matching blueprint §5.1 default-OFF pattern
    /// for all optional LLM pipeline stages. Callers install a real supervisor
    /// via [`Self::with_reflect_supervisor`] and call [`Self::end_turn`] at the
    /// end of each agent turn to trigger the reflection pass.
    pub(crate) reflect_supervisor: Arc<dyn ReflectSupervisor>,
    /// Phase 14.2 — ephemeral per-handle LRU boost cache.
    ///
    /// Populated by [`ScopedLunaris::end_turn`] from
    /// [`lunaris_verify::ReflectOutput::boost`]; consumed as a post-hydrate
    /// rescorer by every [`lunaris_retrieve::RetrievalBuilder`] returned from
    /// [`Self::recall`]. Cache key is `(Scope, Ulid)` so boost signals from
    /// one tenant scope never leak into another scope's recall results.
    ///
    /// Lock discipline: the guard is acquired, all entries are written /
    /// read, then the guard is dropped before the next `.await` point. This
    /// upholds the CLAUDE.md "never hold a lock across `.await`" invariant.
    ///
    /// Capacity: controlled by `LUNARIS_BOOST_CACHE_CAPACITY` (default 10 000)
    /// via [`lunaris_verify::boost_cache_capacity`].
    pub(crate) boost_cache: Arc<parking_lot::RwLock<lru::LruCache<(Scope, Ulid), f32>>>,
    /// Phase 14.3 — concurrency bound for speculative warm-up recalls spawned
    /// by [`ScopedLunaris::end_turn`] when [`ReflectOutput::pre_warm_query`] is
    /// `Some`. Capacity defaults to 4; override via
    /// `LUNARIS_PREWARM_CONCURRENCY` env var (positive integer; 0 or
    /// non-numeric values fall back to the default). If the semaphore is
    /// exhausted when `end_turn` fires, the warm-up is silently skipped (logged
    /// at `DEBUG`) — `end_turn` never blocks on the semaphore.
    pub(crate) warm_up_semaphore: Arc<tokio::sync::Semaphore>,
    /// BPE token counter for the ingest chunker (CHUNK-01 / Finding 1 fix).
    ///
    /// Loaded from the embedder model directory (`embedder_dir()/tokenizer.json`)
    /// at `open` time via `make_token_counter`. Falls back to
    /// `SurrogateTokenCounter` (words×1.3) when the file is absent or
    /// malformed — `tracing::warn!` is emitted in that case. The `with_parts`
    /// and `with_parts_keyword` test seams always use the surrogate so tests
    /// have no model-artifact dependency.
    ///
    /// Passed to `ingest_episode_with_counter` so production chunking uses
    /// real BPE token counts rather than the v0 heuristic.
    pub(crate) token_counter: Arc<dyn TokenCounter + Send + Sync>,
    /// Phase 28 — adaptive meta-framework bake-off config.
    ///
    /// When `Some`, [`Lunaris::ingest`] routes through
    /// [`lunaris_ingest::ingest_episode_with_bakeoff`] which runs the multi-generator
    /// bake-off and persists the winning candidate. The winner's scoring embeddings
    /// are reused directly (SINGLE-PASS — no re-embed). When `None` (default),
    /// the standard [`lunaris_ingest::ingest_episode_with_counter`] path is used.
    ///
    /// Install via [`Self::with_bakeoff`]. `Arc` allows cheap clone of the handle
    /// without copying the config on every ingest call.
    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),
        }
    }

    /// Shared cache-then-embed path. `lowpri` selects the inner embedder's
    /// background lane (`embed_batch_lowpri`) for cache misses so the wrapper
    /// preserves the priority the caller asked for — without this forwarding,
    /// ingest promotion would be silently upgraded to the interactive lane,
    /// defeating the whole non-blocking design (every real embedder is wrapped
    /// in a `CachedEmbedder`).
    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 {
    /// Which embedder backend this process resolved, as a stable lowercase
    /// string (`"llamacpp"`, `"openai-remote"`, `"ollama-remote"`, `"noop"`,
    /// `"unresolved"`).
    ///
    /// **This is the only way an SDK caller can see a degraded embedder.**
    /// The `Noop` fallback is silent by construction: every vector is zeros,
    /// so hybrid recall collapses to BM25 plus insertion-order tie-breaks
    /// while `recall` keeps returning successfully with a plausible-looking
    /// hit list. `NoopEmbedder::dim()` deliberately reports a non-zero
    /// dimension so the operator's existing index geometry stays valid, which
    /// means no amount of inspecting the results reveals it either. Rust
    /// callers had `resolved_embedder_backend()`; Python and TypeScript
    /// callers had nothing at all, which is the gap this closes.
    ///
    /// Process-global, not per-handle: `resolve_embedder` reads env and the
    /// model cache once, on the first `open`. Taking `&self` is for SDK
    /// discoverability — a free function does not appear on the handle a
    /// caller already has, and one that cannot be found does not report.
    ///
    /// Keyword-only operation is a SUPPORTED mode (`npx`/`uvx` standalone
    /// with no staged GGUF), so this reports rather than refuses. See the
    /// W0.7 ledger entry for why the hard-error variant was reverted.
    ///
    /// ## What it does NOT cover
    ///
    /// Only [`Lunaris::open`] records a backend. [`Lunaris::open_with_embedder`]
    /// and the post-open [`Lunaris::with_embedder`] /
    /// [`Lunaris::try_with_embedder`] swaps do not, because the caller already
    /// holds the `Arc<dyn Embedder>` and knows what it is. So a process that
    /// only ever built handles through those paths reports `"unresolved"`
    /// (correct: nothing was resolved), and a process that called `open` and
    /// then swapped in a different embedder keeps reporting what `open`
    /// resolved. Both SDK `open` entry points route through `Lunaris::open`,
    /// so this caveat does not reach a Python or TypeScript caller today —
    /// it matters only to Rust embedders using the BYO seam.
    #[must_use]
    pub fn embedder_backend(&self) -> String {
        resolved_embedder_backend().as_str().to_string()
    }

    /// Production constructor. Opens a storage backend by URL and constructs
    /// the default embedder.
    ///
    /// - `moon://...` → [`lunaris_storage_moon::MoonStorage`] backend.
    ///   Plan 02-02 wires the typed `Arc<MoonStorage>` alongside the dyn
    ///   trait Arcs so `recall().fuse_rrf()` can take the Moon-native one-
    ///   round-trip path. **This is the only scheme 0.7.0 serves.**
    /// - `postgres://`, `sqlite:///path` and `memory://` were retired in
    ///   0.7.0 with `lunaris-storage-postgres` / `lunaris-storage-embedded`.
    ///   They now fail with an `UnsupportedScheme` error that names the
    ///   migration path (`lunaris-migrate` from the v0.6.2 release binary —
    ///   see `docs/migration/0.6-to-0.7.md`) rather than opening a store.
    ///
    /// ## Default backend resolution (llama.cpp-only cutover)
    ///
    /// - **Embedder** — [`lunaris_llamacpp::LlamaCppEmbedder`] backed by the
    ///   `granite-embedding-311m-multilingual-r2` Q4_K_M GGUF (768-d).
    ///   Resolved from `LUNARIS_EMBEDDER_GGUF`, else the
    ///   `~/.lunaris/models/` staged default. Missing GGUF →
    ///   `tracing::warn!` + [`lunaris_core::NoopEmbedder`] (zero vectors).
    /// - **Reranker** — [`lunaris_llamacpp::LlamaCppReranker`] backed by the
    ///   `bge-reranker-v2-m3` Q5_K_M GGUF (sigmoid scores ∈ [0, 1]).
    ///   Resolved from `LUNARIS_RERANKER_GGUF`, else the staged default;
    ///   weight load deferred to the first `rerank()` (N-04 D1). Missing
    ///   GGUF → [`NoopReranker`] (RETRIEVE-06 contract: recall path runs
    ///   even without the rerank pass).
    /// - **Extractor / Verifier** — REMOTE-ONLY. Resolved from
    ///   `LUNARIS_EXTRACT_PROVIDER` / `LUNARIS_VERIFY_PROVIDER`
    ///   (see `default_extractor`, `default_verifier`); unset → degraded
    ///   Noop mode.
    /// - **Consolidator** — resolved from `LUNARIS_CONSOLIDATOR_BACKEND`
    ///   (see `default_consolidator`).
    /// - **Remote embedder (Tier-0 / air-gap)** — build with
    ///   `--features embed-remote` and set
    ///   `LUNARIS_EMBEDDER_OPENAI_URL` (OpenAI-compatible `/v1/embeddings`)
    ///   or `LUNARIS_EMBEDDER_OLLAMA_URL` to skip local inference entirely.
    pub async fn open(url: &str) -> Result<Self, LunarisError> {
        let embedder = resolve_embedder(embed_max_batch_tokens()).await?;
        // W0.7 successor: `resolve_embedder` has now recorded the backend, so a
        // degraded one announces itself here rather than waiting to be asked.
        // This is deliberately on `open` and NOT on `open_with_embedder`: the
        // latter is the injection seam, where the caller supplied the embedder
        // and already knows what it is. Both SDK entry points route through
        // `open`, so neither needs its own copy.
        announce_degradation_once();
        Self::open_with_embedder(url, embedder).await
    }

    /// Like [`Lunaris::open`] but uses the caller-provided `embedder`
    /// directly instead of constructing the default llama.cpp embedder /
    /// `NoopEmbedder` fallback.
    ///
    /// Use this when:
    ///
    /// - The compile-time feature set has no real embedder backend and the
    ///   silent-fallback `NoopEmbedder` is unacceptable — pass a BYO
    ///   embedder you constructed elsewhere.
    /// - You need to pin a specific vector dim BEFORE Moon creates its FT
    ///   indices. Moon's `FT.CREATE` is idempotent and DOES NOT auto-resize
    ///   an existing index, so post-`open()` `with_embedder` calls cannot
    ///   change the on-disk dim of an existing collection. This method runs
    ///   the embedder's `dim()` through `MoonStorage::connect_with_dim` on
    ///   first open, which is the right time to size the index.
    /// - You want a `NoopEmbedder` at a specific dim:
    ///   ```no_run
    ///   use std::sync::Arc;
    ///   use lunaris::{Lunaris, LunarisError};
    ///   use lunaris_core::NoopEmbedder;
    ///
    ///   # async fn demo() -> Result<(), LunarisError> {
    ///   let handle = Lunaris::open_with_embedder(
    ///       "moon://localhost:6380",
    ///       Arc::new(NoopEmbedder::new(1536)),
    ///   ).await?;
    ///   # Ok(()) }
    ///   ```
    ///
    /// The reranker / extractor / verifier / consolidator are still
    /// resolved from their env vars exactly as in [`Lunaris::open`].
    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);
        // Build the BPE token counter from the embedder's tokenizer.json.
        // Falls back to SurrogateTokenCounter (tracing::warn!) when absent.
        let token_counter = make_token_counter(Some(&embedder_dir().join("tokenizer.json")));
        let reranker = resolve_reranker().await?;
        // Plan 03-03: Construct the graph pipeline handle. Initial state
        // comes from `LUNARIS_GRAPH_ENABLED=1|0` env var (D-10); default OFF
        // per blueprint §5.2. The default extractor is candle Gemma-3 4B (or
        // NoopExtractor on cache miss — see `default_extractor`).
        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));
        // Plan 04-04: Construct the verifier + consolidator pipeline handles.
        // Initial state from `LUNARIS_VERIFY_ENABLED` / `LUNARIS_CONSOLIDATE_ENABLED`
        // env vars (D-08); default OFF per blueprint §5.1. Default backends
        // are NoopVerifier / NoopConsolidator — production callers wire real
        // backends via `with_verifier` / `with_consolidator`.
        let verifier = default_verifier().await;
        // Phase 16-01 (CONSOL-V1-01): resolve backend from LUNARIS_CONSOLIDATOR_BACKEND;
        // fail-fast on unknown env values (no silent fallback).
        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" => {
                // Size the Moon FT vector indices to the resolved embedder's
                // dimension (default 768-d for granite-r2; pass a wider embedder
                // via `Lunaris::open_with_embedder` and the indices grow to
                // match). Moon's FT.CREATE has no dimension cap. Footgun: if
                // the Moon instance already holds indices at a different dim,
                // they are NOT auto-resized — drop them first.
                let m = Arc::new(MoonStorage::connect_with_dim(url, embedder.dim()).await?);
                let storage_arc: Arc<dyn StoragePort> = m.clone();
                // B-10: bind the StoragePort Arc to BOTH pipelines AFTER
                // we've constructed it. Also bind the HlcClock so the
                // Plan 04-04 Task 4 apply_supersede has a tick source. If
                // env var initial-state was ON, also kick the worker via
                // spawn_worker_if_idle so callers don't have to call
                // enable() a second time post-bind.
                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,
                    // GA-1: rerank toggle frozen at construction — the ONLY
                    // env read (mirrors the graph pipeline's D-10 pattern).
                    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()),
                )))
            }
        }
    }

    /// Legacy test / latency-budget-swap escape hatch. Wires a custom
    /// storage handle, embedder, and clock — bypasses [`Self::open`]'s
    /// default constructors. The keyword Arc is taken from the same
    /// `storage` Arc by attempting an Arc-to-trait downcast — when the
    /// caller's storage type also impls `KeywordPort`, this works
    /// transparently. Otherwise the keyword path returns
    /// `StorageError::NotSupported` at call time.
    ///
    /// Production callers should use [`Self::open`] OR
    /// [`Self::with_parts_keyword`] with explicit `keyword` Arc.
    #[doc(hidden)]
    pub fn with_parts(
        storage: Arc<dyn StoragePort>,
        embedder: Arc<dyn Embedder>,
        clock: Arc<HlcClock>,
    ) -> Self {
        // Plan 04-04 B-10: construct the verify + consolidator pipelines
        // BEFORE the Self struct so we can call bind_storage on each handle
        // with the storage Arc.
        let verify_pipeline = Arc::new(VerifierPipelineHandle::new(
            false,
            Arc::new(NoopVerifier) as Arc<dyn Verifier>,
        ));
        // Phase 16-01 (CONSOL-V1-01): resolve backend from env. Test seam is
        // infallible — `expect` surfaces env misconfiguration loudly rather
        // than silently falling back (matches fail-fast contract of the
        // `Lunaris::open` path).
        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));
        // B-10: bind storage to BOTH pipelines (2 of the 4 total bind_storage
        // call sites in handle.rs). Also bind the HlcClock to verify_pipeline
        // so the Plan 04-04 Task 4 apply_supersede has a tick source.
        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,
            // Default to NoopReranker so existing callers (Plan 02-01 smoke
            // tests) keep working without picking up the candle dep
            // transitively. Production callers swap via with_reranker.
            reranker: Arc::new(NoopReranker) as Arc<dyn Reranker>,
            // GA-1: test seam stays OFF (no env read) — same shape as the
            // graph pipeline's hardcoded `false` below. Tests opt in via
            // `with_recall_rerank`.
            recall_rerank: crate::recall_rerank::RecallRerankConfig::default(),
            // Plan 03-03: graph pipeline OFF by default with a NoopExtractor
            // installed. Tests that exercise the graph-ON path call
            // `handle.graph_pipeline().enable()` + `handle.with_extractor(...)`
            // explicitly; default-OFF preserves the Phase 2 fast path.
            graph_pipeline: Arc::new(GraphPipelineHandle::new(
                false,
                Arc::new(NoopExtractor) as Arc<dyn Extractor>,
            )),
            verify_pipeline,
            consolidator_pipeline,
            // Phase 13 — default OFF per blueprint §5.1 default-OFF pattern.
            reflect_supervisor: Arc::new(NoopReflectSupervisor),
            // Phase 14.2 — ephemeral boost cache, capacity from env (default 10_000).
            boost_cache: Arc::new(parking_lot::RwLock::new(lru::LruCache::new(
                boost_cache_capacity(),
            ))),
            // Phase 14.3 — semaphore for bounded fire-and-forget warm-up spawns.
            warm_up_semaphore: Arc::new(tokio::sync::Semaphore::new(resolve_prewarm_concurrency())),
            // Test seam: no model artifact available; use the surrogate counter.
            token_counter: make_token_counter(None),
            // Phase 28: bakeoff OFF by default in test seam; install via with_bakeoff.
            bakeoff_config: None,
        }
    }

    /// Test seam used by Plan 02-02 Task 3's `recall_smoke` — wire a
    /// `KeywordPort` Arc explicitly. Production callers go through
    /// [`Self::open`] which constructs both Arcs from the URL.
    #[doc(hidden)]
    pub fn with_parts_keyword(
        storage: Arc<dyn StoragePort>,
        keyword: Arc<dyn KeywordPort>,
        embedder: Arc<dyn Embedder>,
        clock: Arc<HlcClock>,
    ) -> Self {
        // Plan 04-04 B-10: same shape as with_parts — construct the pipeline
        // handles BEFORE the Self struct, then bind_storage on both.
        let verify_pipeline = Arc::new(VerifierPipelineHandle::new(
            false,
            Arc::new(NoopVerifier) as Arc<dyn Verifier>,
        ));
        // Phase 16-01 (CONSOL-V1-01): resolve backend from env (same fail-fast
        // contract as `with_parts`).
        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));
        // B-10: bind storage to BOTH pipelines (the OTHER 2 of the 4 total
        // bind_storage call sites in handle.rs). Also bind the HlcClock to
        // verify_pipeline.
        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>,
            // GA-1: test seam stays OFF (no env read) — see `with_parts`.
            recall_rerank: crate::recall_rerank::RecallRerankConfig::default(),
            // Plan 03-03 — see `with_parts` for the rationale.
            graph_pipeline: Arc::new(GraphPipelineHandle::new(
                false,
                Arc::new(NoopExtractor) as Arc<dyn Extractor>,
            )),
            verify_pipeline,
            consolidator_pipeline,
            // Phase 13 — default OFF per blueprint §5.1 default-OFF pattern.
            reflect_supervisor: Arc::new(NoopReflectSupervisor),
            // Phase 14.2 — ephemeral boost cache, capacity from env (default 10_000).
            boost_cache: Arc::new(parking_lot::RwLock::new(lru::LruCache::new(
                boost_cache_capacity(),
            ))),
            // Phase 14.3 — semaphore for bounded fire-and-forget warm-up spawns.
            warm_up_semaphore: Arc::new(tokio::sync::Semaphore::new(resolve_prewarm_concurrency())),
            // Test seam: no model artifact available; use the surrogate counter.
            token_counter: make_token_counter(None),
            // Phase 28: bakeoff OFF by default in test seam; install via with_bakeoff.
            bakeoff_config: None,
        }
    }

    /// Public escape hatch — replace the embedder on an existing handle.
    ///
    /// [`Lunaris::open`] constructs the default llama.cpp embedder backed by
    /// the granite-r2 Q4_K_M GGUF; call this method post-construction
    /// to swap in any `Arc<dyn Embedder>` (e.g., a `StubEmbedder` in tests,
    /// the feature-gated `lunaris_embed_remote::OllamaEmbedder`, or a remote
    /// embedder service).
    ///
    /// **Footgun**: this method does NOT re-size the underlying storage
    /// vector index. If you swap embedders post-`open()`, ensure the new
    /// embedder's `dim()` matches the original; otherwise `FT.SEARCH` /
    /// `pgvector` queries will reject the dimension mismatch at call time.
    /// Use [`Lunaris::open_with_embedder`] for the pre-index-creation path.
    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
    }

    /// Phase 28 — install an adaptive meta-framework bake-off config.
    ///
    /// When installed, every subsequent [`Lunaris::ingest`] call routes through
    /// [`lunaris_ingest::ingest_episode_with_bakeoff`], which runs the
    /// multi-generator bake-off and persists the winning candidate. The winner's
    /// scoring embeddings are reused directly (SINGLE-PASS — no re-embed).
    ///
    /// Pass `None` (or call this with `Arc::new(BakoffConfig::default())`) to
    /// restore the standard counter-based ingest path. The `Arc` wrapper lets
    /// the config be shared cheaply across `Lunaris::clone()` calls.
    ///
    /// ## INGEST-04 invariant
    ///
    /// Installing a bakeoff config does NOT add a second `atomic_write` call.
    /// Both the standard path and the bakeoff path funnel through
    /// `assemble_and_write` in `lunaris_ingest::pipeline`, which holds the
    /// single executable `storage.atomic_write` call site.
    pub fn with_bakeoff(mut self, config: Arc<BakoffConfig>) -> Self {
        self.bakeoff_config = Some(config);
        self
    }

    /// N-04 D2 — fallible counterpart to [`Self::with_embedder`].
    ///
    /// Refuses the swap when `embedder.dim() != self.embedder.dim()`. The
    /// handle's existing `embedder.dim()` is the dim Moon's `FT.CREATE` index
    /// was sized for at `Lunaris::open*` time
    /// (see [`Lunaris::open_with_embedder`] — the dim flows into
    /// `MoonStorage::connect_with_dim`). Replacing it with a different-width
    /// embedder produces garbage similarity scores until the index is
    /// rebuilt, which is silent corruption masquerading as a working
    /// recall path. This method exposes the check at the API boundary so
    /// callers can either match the dim or migrate explicitly.
    ///
    /// Returns `Ok(Self)` on match, otherwise
    /// `Err(LunarisError::Storage(StorageError::Backend(_)))` carrying
    /// both dims in the message.
    ///
    /// The infallible [`Self::with_embedder`] is intentionally retained
    /// (and emits a `tracing::warn!` on mismatch) for backwards-compat with
    /// callers that have proven their store tolerates the swap (e.g., tests
    /// that never run a vector query).
    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)
    }

    /// Escape hatch — replace the reranker on an existing handle.
    ///
    /// [`Lunaris::open`] constructs the default llama.cpp reranker backed by
    /// the bge-reranker-v2-m3 Q5_K_M GGUF and falls back to
    /// [`NoopReranker`] on cache miss per the RETRIEVE-06 contract. Tests
    /// pass `Arc::new(NoopReranker)` for determinism; production callers can
    /// wire a custom cross-encoder (e.g., a remote rerank service) without
    /// touching the rest of the construction path. Per RETRIEVE-06 this is
    /// also how callers turn the rerank pass off entirely if the per-batch
    /// budget busts on their hardware:
    /// `handle.with_reranker(Arc::new(NoopReranker))`.
    pub fn with_reranker(mut self, reranker: Arc<dyn Reranker>) -> Self {
        self.reranker = reranker;
        self
    }

    /// Plan 03-03 escape hatch — replace the extractor on an existing handle.
    /// Production callers wiring a `CloudApiExtractor` (cfg-gated behind the
    /// `cloud-api` feature) or a custom [`lunaris_extract::Extractor`] impl
    /// use this; tests pass `Arc::new(lunaris_extract::NoopExtractor)` for
    /// determinism.
    ///
    /// Note: the extractor lives inside the [`GraphPipelineHandle`]'s
    /// `RwLock<Option<Arc<dyn Extractor>>>` — this method swaps it via
    /// [`GraphPipelineHandle::set_extractor`], NOT by replacing the entire
    /// `graph_pipeline` field. Toggle state and the state-change counter are
    /// preserved across the swap (D-12 idempotent observability).
    pub fn with_extractor(self, extractor: Arc<dyn Extractor>) -> Self {
        self.graph_pipeline.set_extractor(extractor);
        self
    }

    /// Plan 04-04 escape hatch — replace the verifier on an existing handle.
    /// Production callers wiring `CandleGemma3_27B` (cfg-gated `candle`) /
    /// `OllamaVerifier` / `CloudApiVerifier` use this; tests pass
    /// `Arc::new(NoopVerifier)` for determinism.
    ///
    /// The verifier lives inside the [`VerifierPipelineHandle`]'s
    /// `RwLock<Option<Arc<dyn Verifier>>>` — this method swaps it via
    /// [`VerifierPipelineHandle::set_verifier`], NOT by replacing the entire
    /// `verify_pipeline` field. Toggle state and the state-change counter are
    /// preserved across the swap (D-12 idempotent observability).
    pub fn with_verifier(self, verifier: Arc<dyn Verifier>) -> Self {
        self.verify_pipeline.set_verifier(verifier);
        self
    }

    /// Plan 04-04 escape hatch — replace the consolidator on an existing handle.
    /// Production callers install a real ACT-R consolidator via this; tests pass
    /// `Arc::new(NoopConsolidator)` for determinism.
    ///
    /// Same swap semantics as [`Self::with_verifier`] — toggle + counter
    /// preserved.
    pub fn with_consolidator(self, consolidator: Arc<dyn Consolidator>) -> Self {
        self.consolidator_pipeline.set_consolidator(consolidator);
        self
    }

    /// Phase 13 escape hatch — replace the reflection supervisor on an existing
    /// handle. Production callers install an [`LlmReflectSupervisor`] (or a
    /// custom [`ReflectSupervisor`] impl) via this; tests pass
    /// `Arc::new(NoopReflectSupervisor)` for determinism.
    ///
    /// Unlike `verify_pipeline` and `consolidator_pipeline`, the reflect
    /// supervisor is a plain `Arc` (no background worker, no toggle) — it is
    /// invoked synchronously per [`Self::end_turn`] call on the caller's task.
    ///
    /// [`LlmReflectSupervisor`]: lunaris_verify::LlmReflectSupervisor
    pub fn with_reflect_supervisor(mut self, supervisor: Arc<dyn ReflectSupervisor>) -> Self {
        self.reflect_supervisor = supervisor;
        self
    }

    /// Phase 13 — signal the end of an agent turn and run the reflection pass.
    ///
    /// Calls [`ReflectSupervisor::reflect`] with `input` and returns the
    /// advisory [`ReflectOutput`] (`invalidate`, `boost`, `pre_warm_query`).
    ///
    /// ## Budget + failure discipline
    ///
    /// The supervisor enforces its own timeout (default 500 ms for
    /// [`LlmReflectSupervisor`]). If the supervisor returns `Err`, this method
    /// propagates it — callers that treat reflect as best-effort should wrap
    /// with `.unwrap_or_default()`. If the installed supervisor is
    /// [`NoopReflectSupervisor`] (the default), this call is a cheap no-op
    /// returning `ReflectOutput::default()`.
    ///
    /// ## Non-requirements in this commit
    ///
    /// The returned [`ReflectOutput`] is **advisory only** — storage-side
    /// application (`invalidate` → `BiTemporal::invalidate_sys`, `boost` →
    /// retrieval-rank adjustment, `pre_warm_query` → speculative recall) is a
    /// Phase 13 follow-up. For now, the output is logged and returned to the
    /// caller.
    ///
    /// [`LlmReflectSupervisor`]: lunaris_verify::LlmReflectSupervisor
    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)
    }

    /// Borrow accessors — needed by Plan 02-02's retrieve DSL builder.
    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()
    }

    /// Liveness probe for `lunaris-server`'s `/healthz` rollout-cutback surface
    /// (`observability-rollout-maturity`): delegates to the storage backend's
    /// [`StoragePort::health_check`] (Moon issues a real PING; in-process
    /// backends report healthy via the additive default). `Err` → the server
    /// answers 503 so the 5%→100% rollout controller cuts traffic back.
    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()
    }
    /// Borrow the typed `Arc<MoonStorage>` when the handle was opened against
    /// a Moon backend; `None` otherwise. Plan 02-02's `recall()` plumbs this
    /// into the `RetrievalBuilder` so `fuse_rrf` can opt into Moon-native
    /// hybrid search.
    pub fn moon_storage(&self) -> Option<Arc<MoonStorage>> {
        self.moon_storage.clone()
    }
    /// Borrow the configured reranker. Lets callers chain
    /// `handle.recall().rerank(handle.reranker())` when they want the rerank
    /// pass without re-declaring it.
    pub fn reranker(&self) -> Arc<dyn Reranker> {
        self.reranker.clone()
    }

    /// Plan 03-03 — borrow the [`GraphPipelineHandle`] for runtime toggle
    /// control. EXTRACT-06 single-switch surface (D-10):
    ///
    /// - `handle.graph_pipeline().enable()` / `.disable()` — flip the
    ///   pipeline ON / OFF (idempotent, observable per D-12).
    /// - `handle.graph_pipeline().is_enabled()` — current state.
    /// - `handle.graph_pipeline().force_reload().await` — reload the
    ///   extractor from default cache (e.g., after `huggingface-cli` finished
    ///   downloading weights).
    pub fn graph_pipeline(&self) -> Arc<GraphPipelineHandle> {
        self.graph_pipeline.clone()
    }

    /// Plan 03-03 — snapshot the currently-installed [`Extractor`] `Arc`.
    /// Useful for the canonical compose example in tests + bench harnesses.
    /// Returns `None` only when the [`GraphPipelineHandle`] has no extractor
    /// installed (rare — only via explicit `set_extractor` with a None which
    /// is not exposed in the public surface; the public surface always
    /// installs at least [`NoopExtractor`]).
    pub fn extractor(&self) -> Option<Arc<dyn Extractor>> {
        self.graph_pipeline.snapshot_extractor()
    }

    /// Plan 04-04 — borrow the [`VerifierPipelineHandle`] for runtime toggle
    /// control. D-08 single-switch surface:
    ///
    /// - `handle.verify_pipeline().enable()` / `.disable()` — flip the
    ///   pipeline ON / OFF (idempotent, observable per D-12). Spawns / signals
    ///   shutdown on the in-process tokio worker.
    /// - `handle.verify_pipeline().is_enabled()` — current state.
    /// - `handle.verify_pipeline().join_worker().await` — await full worker
    ///   exit after a `disable()`.
    pub fn verify_pipeline(&self) -> Arc<VerifierPipelineHandle> {
        self.verify_pipeline.clone()
    }

    /// Plan 04-04 — borrow the [`ConsolidatorPipelineHandle`] for runtime
    /// toggle control. Same surface shape as [`Self::verify_pipeline`].
    pub fn consolidator_pipeline(&self) -> Arc<ConsolidatorPipelineHandle> {
        self.consolidator_pipeline.clone()
    }

    /// Plan 04-04 — snapshot the currently-installed [`Verifier`] `Arc`.
    pub fn verifier(&self) -> Option<Arc<dyn Verifier>> {
        self.verify_pipeline.snapshot_verifier()
    }

    /// Plan 04-04 — snapshot the currently-installed [`Consolidator`] `Arc`.
    pub fn consolidator(&self) -> Option<Arc<dyn Consolidator>> {
        self.consolidator_pipeline.snapshot_consolidator()
    }

    /// Phase 13 — borrow the configured [`ReflectSupervisor`] `Arc`.
    /// Returns the currently-installed supervisor — `NoopReflectSupervisor` by
    /// default, or whatever was last passed to [`Self::with_reflect_supervisor`].
    pub fn reflect_supervisor(&self) -> Arc<dyn ReflectSupervisor> {
        self.reflect_supervisor.clone()
    }

    /// Phase 14.3 — borrow the warm-up semaphore `Arc`.
    ///
    /// Primarily for testing: callers can assert `available_permits()` to
    /// verify the semaphore was / was not acquired.
    pub fn warm_up_semaphore(&self) -> Arc<tokio::sync::Semaphore> {
        self.warm_up_semaphore.clone()
    }

    /// Phase 14.3 test seam — replace the warm-up semaphore with a custom
    /// capacity. Use in integration tests that need to control the concurrency
    /// bound (e.g., capacity=1 for the semaphore-bound test).
    ///
    /// This is intentionally `#[doc(hidden)]` — production code uses the
    /// env-var knob (`LUNARIS_PREWARM_CONCURRENCY`) at construction time.
    #[doc(hidden)]
    pub fn with_prewarm_concurrency(mut self, capacity: usize) -> Self {
        self.warm_up_semaphore = Arc::new(tokio::sync::Semaphore::new(capacity));
        self
    }

    /// RFC 0001 Wave 0 — construct a scope-bound view over this handle.
    ///
    /// All operations issued through the returned [`ScopedLunaris`] carry
    /// `scope` as their partitioning key. The underlying `Lunaris` handle is
    /// borrowed for the lifetime `'a` — no cloning occurs.
    ///
    /// Wave 1 will route each method through the real scope-aware backends.
    /// Wave 0 stubs return `todo!()` so the API surface is frozen before the
    /// routing logic lands.
    pub fn scoped(&self, scope: Scope) -> ScopedLunaris<'_> {
        ScopedLunaris { engine: self, scope }
    }

    /// Cross-scope enumeration — pass-through to
    /// [`StoragePort::list_scopes`].
    ///
    /// Returns a paginated [`ScopePage`](lunaris_core::ScopePage) of scopes
    /// known to the underlying backend, optionally filtered by `prefix`. The
    /// cursor is opaque (Q-U1 lock) and MUST be passed back unchanged on
    /// subsequent calls; `next_cursor == None` means enumeration is exhausted.
    ///
    /// ## Backend support
    ///
    /// Supported on Moon — lazy SCAN-parse derivation from the
    /// `lunaris:{scope}:…` keyspace. The method still returns `Result` and a
    /// custom `StoragePort` may answer `Err(StorageError::NotSupported(_))`:
    /// that is what the Postgres backend did (its primitive tables were
    /// RLS-protected with `FORCE ROW LEVEL SECURITY` and the application role
    /// could not bypass it), and the contract is kept so a future backend can
    /// decline without a signature change. Callers handling `NotSupported`
    /// supply a known scope list from caller context instead.
    ///
    /// This is the v0.3 surface introduced by the cross-scope enumeration
    /// patch. The higher-level `list_atoms` / `get_atom_by_scope_lsn` from the
    /// upstream brief are intentionally deferred — Lunaris exposes six
    /// primitive kinds (episode/chunk/entity/relation/fact/community) rather
    /// than a unified `Atom`, and introducing that abstraction is a separate
    /// design pass.
    ///
    /// ## Example
    ///
    /// ```no_run
    /// use lunaris::{Lunaris, LunarisError};
    ///
    /// # async fn demo() -> Result<(), LunarisError> {
    /// let engine = Lunaris::open("moon://127.0.0.1:6380").await?;
    /// let page = engine.list_scopes(None, 100, None).await?;
    /// for scope in page.scopes {
    ///     println!("known scope: {scope:?}");
    /// }
    /// # Ok(()) }
    /// ```
    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)
    }

    /// Bulk-invalidate FT index records authored by `node_id` within the HLC wall-clock
    /// window `[hlc_wall_lo_inclusive, hlc_wall_hi_inclusive]` (both ends inclusive).
    ///
    /// Called by Helios when `helios-git` detects a force-push or rebase that abandons
    /// commits. This evicts stale recall from the agent's memory so subsequent queries
    /// do not surface facts from the abandoned branch.
    ///
    /// ## Fan-out
    ///
    /// The method issues `FT.INVALIDATE_RANGE` against each known Lunaris collection
    /// (`chunks`, `entities`, `facts`, `communities`) in parallel via `join_all`.
    /// Collections whose index is missing on Moon (`WRONGTYPE` response) or whose
    /// backend does not support the primitive (`NotSupported`) are skipped with a
    /// `WARN` log (degraded mode — the caller receives a partial count, not an error).
    ///
    /// ## HLC wall-clock semantics
    ///
    /// `hlc_wall_lo_inclusive` and `hlc_wall_hi_inclusive` are milliseconds since
    /// the Unix epoch, matching Moon's `hlc_wall` NUMERIC field convention. Both
    /// bounds are **inclusive** (Moon `[lo, hi]` closed interval). Callers with a
    /// half-open Rust range `lo..hi` must pass `hi - 1` as the upper bound.
    ///
    /// ## Timeout
    ///
    /// Each per-index call is bounded to 250 ms (CONTEXT.md §5 IO failure surface).
    /// There is no retry — this is a bulk admin operation; the caller decides retry
    /// policy.
    ///
    /// ## Schema preconditions
    ///
    /// For the invalidation to match documents, the target FT indices must declare:
    /// - `hlc_node_id` as a `TAG` field
    /// - `hlc_wall` as a `NUMERIC` field
    ///
    /// Indices lacking these fields return 0 silently (Moon bitmap intersect returns
    /// empty). This is expected for indices predating the `helios-git` schema additions;
    /// see `.planning/W2-L2-INVALIDATE-RANGE-SUMMARY.md` for the full schema roadmap.
    ///
    /// ## Empty range
    ///
    /// If `hlc_wall_lo_inclusive > hlc_wall_hi_inclusive`, the method returns `Ok(0)`
    /// immediately without issuing any wire calls.
    ///
    /// # Example
    ///
    /// ```no_run
    /// // Helios force-push detector hands us the abandoned HLC window:
    /// use lunaris::{Lunaris, LunarisError};
    /// use lunaris_core::Scope;
    ///
    /// # async fn demo(engine: Lunaris) -> Result<(), LunarisError> {
    /// let scope = Scope::new("helios.my-worktree").unwrap();
    /// let count = engine.invalidate_range(
    ///     &scope,
    ///     "helios-git@aabbcc",
    ///     1_700_000_000_000,
    ///     1_700_000_100_000,
    /// ).await?;
    /// tracing::info!(count, "invalidated stale recall");
    /// # Ok(()) }
    /// ```
    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
    }
}

/// Sentinel `KeywordPort` impl returned by [`Lunaris::with_parts`] when the
/// caller did NOT supply a real keyword backend. Calling `keyword_search`
/// returns `StorageError::NotSupported` so callers see a clear failure
/// rather than a silent empty result.
#[derive(Debug, Clone, Copy)]
struct NoKeywordSupport;

#[async_trait::async_trait]
impl KeywordPort for NoKeywordSupport {
    /// Wave 2.5A: gains `scope: &Scope` per RFC 0001 §3.4 amendment.
    /// Scope is ignored — this sentinel returns NotSupported regardless.
    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)",
        ))
    }
}

// ── HOOK-05: idempotency ──────────────────────────────────────────────────────

/// Outcome of [`ScopedLunaris::ingest_idempotent`] (HOOK-05).
///
/// `Fresh` means a new episode was written; `Duplicate` means the dedupe key
/// was already present and the prior LSN is returned without a second
/// `atomic_write`. INGEST-04 is preserved: `Duplicate` does NOT call
/// `atomic_write` at all; `Fresh` calls it exactly once via [`ScopedLunaris::ingest`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IngestKind {
    /// New episode was written; the enclosed `Lsn` is its committed LSN.
    Fresh,
    /// Episode already present; the enclosed `Lsn` is the prior committed LSN.
    Duplicate(lunaris_core::Lsn),
}

/// RFC 0001 Wave 1D — scope-bound view over a [`Lunaris`] handle.
///
/// Constructed via [`Lunaris::scoped`]. All operations issued through this
/// wrapper carry the bound [`Scope`] as their partitioning key. The `'a`
/// lifetime ties the view to the underlying handle so no `Arc` clone is
/// required for the wrapper itself.
///
/// ## Scope enforcement
///
/// Callers build an [`EpisodeBuilder`] (scope-less payload) and pass it to
/// [`Self::ingest`]. The wrapper is the ONLY code path that can call
/// `EpisodeBuilder::into_episode` (it's `pub` but the scope value comes
/// exclusively from this wrapper's `self.scope` field). Callers cannot
/// construct an `Episode` with an arbitrary scope by bypassing this type.
pub struct ScopedLunaris<'a> {
    pub(crate) engine: &'a Lunaris,
    pub(crate) scope: Scope,
}

impl<'a> ScopedLunaris<'a> {
    /// Returns the [`Scope`] this view is bound to.
    pub fn scope(&self) -> &Scope {
        &self.scope
    }

    /// Ingest an episode payload under the bound scope.
    ///
    /// Takes an [`EpisodeBuilder`] (scope-less payload) rather than a fully
    /// constructed `Episode` so the caller cannot inject an arbitrary scope.
    /// The wrapper stamps `self.scope` onto the episode via
    /// `builder.into_episode(self.scope.clone(), &self.engine.clock)` before
    /// delegating to [`Lunaris::ingest`].
    ///
    /// INGEST-04 invariant: exactly one `atomic_write` call per ingest path.
    /// The write lives in `lunaris_ingest::ingest_episode` (graph OFF) or
    /// `ingest_episode_graph_on` (graph ON), unchanged from the non-scoped path.
    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
    }

    /// Idempotent ingest (HOOK-05): if `dedupe_key` has been seen before within
    /// this scope, return the prior `Lsn` without a second `atomic_write`.
    ///
    /// ## INGEST-04 invariant preserved
    ///
    /// The dedupe key lookup is READ-ONLY (`StoragePort::lookup_by_dedupe_key`).
    /// Only on [`IngestKind::Fresh`] does the existing single `atomic_write`
    /// (inside [`Self::ingest`]) run. No new `atomic_write` call site is introduced.
    ///
    /// ## Trait-method approach (W6 fix)
    ///
    /// Uses `StoragePort::lookup_by_dedupe_key` / `insert_dedupe_key` trait methods
    /// directly — no `as_any()` downcast. Moon implements them via the
    /// `lunaris:{scope}:dedupe:{blake3}` KV sidecar with SET-NX
    /// first-writer-wins (ADD task moon-parity-honesty — closed the former
    /// "SQLite-only idempotency" v0.5 boundary). Both methods keep a trait
    /// default returning `Ok(None)` / `Ok(())`, so a custom `StoragePort` that
    /// implements neither falls through to unconditional Fresh ingest rather
    /// than failing — which is why the HOOK-05 guard
    /// (`lunaris-hook/tests/idempotency.rs`) asserts against a real Moon and
    /// not a double.
    ///
    /// ## Post-commit race window (T-24-03-06)
    ///
    /// `insert_dedupe_key` runs AFTER the `atomic_write` commit. If the process is
    /// killed in the window between those two operations, replay produces a duplicate
    /// Episode. Mitigation deferred to v0.6. The `insert_dedupe_key` failure is
    /// non-fatal (logged at WARN level).
    pub async fn ingest_idempotent(
        &self,
        builder: EpisodeBuilder,
        dedupe_key: &str,
    ) -> Result<(Lsn, IngestKind), LunarisError> {
        // Attempt read-only lookup via StoragePort trait method.
        // Moon returns the real hit from its dedupe sidecar; a port that does
        // not implement the sidecar answers Ok(None) via the trait default.
        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",
                );
            }
        }

        // Fresh path: ingest (single atomic_write inside self.ingest), then
        // record the dedupe key in the sidecar table (best-effort, non-fatal).
        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))
    }

    /// Phase 23 — agent-supplied structured ingest under the bound scope.
    ///
    /// Delegates to [`Lunaris::ingest_structured`] with `self.scope` so
    /// the caller cannot inject an arbitrary scope. See the
    /// [`crate::structured_ingest`] module rustdoc for the full design
    /// (deterministic EntityId, always-on graph writes, single
    /// `atomic_write` per call).
    ///
    /// INGEST-04 invariant: exactly one `atomic_write` call per ingest
    /// path. The write lives in
    /// [`crate::structured_ingest::ingest_structured_inner`] for this path
    /// (vs. `lunaris_ingest::ingest_episode` / `ingest_episode_graph_on`
    /// for [`Self::ingest`]).
    pub async fn ingest_structured(
        &self,
        payload: crate::structured_ingest::StructuredIngest,
    ) -> Result<Lsn, LunarisError> {
        self.engine.ingest_structured(payload, self.scope.clone()).await
    }

    /// Recall hits under the bound scope.
    ///
    /// Runs the **default plan** — the GA-1 unified production root
    /// (`lunaris_retrieve::production_root`): `Vector ∧ BM25("chunks") →
    /// fuse_rrf(60) → top(30)`, fact legs when the graph pipeline is ON, and
    /// the opt-in `LUNARIS_RECALL_RERANK` cross-encoder stage — executes it,
    /// and returns the hydrated `Vec<Hit>`. This is the one-shot convenience
    /// form; for a custom plan (graph / tree, `as_of`, thresholds) use
    /// [`Self::dsl`].
    /// Wave 2.5C: the scope is applied to the `Vector` search and to hydrate,
    /// so only hits from this scope's partition are returned. (The same scope
    /// threading covers `Graph` / `Keyword` and any other operators you attach
    /// via [`Self::dsl`].)
    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
    }

    /// Set this scope's retention policy (W4.6 / D6.4).
    ///
    /// Retention is **opt-in per scope**: a scope with no policy is never
    /// swept. The failure mode of an accidental policy is unrecoverable data
    /// loss and the failure mode of an accidentally-absent one is disk, so the
    /// default is the recoverable one.
    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
    }

    /// Read this scope's retention policy, or `None` when it has none.
    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
    }

    /// Run one retention pass over this scope, against the current wall clock.
    ///
    /// A no-op returning `policy: None` when the scope has no policy. See
    /// [`crate::retention`] for why a sweep goes through `forget` rather than
    /// deleting directly, and why Lunaris does not schedule this for you.
    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
    }

    /// [`Self::enforce_retention`] against a caller-chosen wall clock, so a
    /// backfill or a replay can pin the cutoff instead of racing it.
    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
    }

    /// Report what [`Self::enforce_retention`] would sweep, sweeping nothing.
    ///
    /// Wave 6 / R1 — the preview half of retention, so a caller (notably the
    /// LLM-driven `memory.retention_enforce` tool, which previews by default)
    /// can answer "what would this take?" without recomputing the cutoff.
    /// See [`crate::retention::preview_at`].
    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
    }

    /// [`Self::preview_retention`] against a caller-chosen wall clock.
    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
    }

    /// Read this scope's audit trail over a closed time range.
    ///
    /// W4.6 / D6.3. Until now the audit log was write-only: every producer
    /// published to `__lunaris_audit__` and nothing in the repo read it back,
    /// so "who deleted this?" — the question the trail exists to answer — had
    /// no answer. This is the consumer.
    ///
    /// **Non-destructive.** It does not pop, ack, or advance any consumer
    /// group, so it can be run repeatedly and a background subscriber on the
    /// same topic is unaffected.
    ///
    /// **Reads only this scope's own topic**, which since W4.6 is also the
    /// only place this scope's events are written. That ordering was not
    /// optional: a reader built on the previous `Scope::dev()`-for-everyone
    /// publish would have served one tenant another tenant's history.
    ///
    /// `from_ms` / `to_ms` are inclusive wall-clock milliseconds, `None`
    /// unbounded. Records come back oldest-first, capped at `limit`. Entries
    /// that fail to decode are counted in [`lunaris_core::audit::AuditPage::undecodable`] rather
    /// than silently skipped.
    ///
    /// Returns `StorageError::NotSupported` on a backend with no range read.
    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)
    }

    /// Forget primitive bound to the wrapper's scope — the canonical entry
    /// point superseding the deprecated [`Lunaris::forget`].
    ///
    /// Wave 1D (ADD task forget-scope-routing, 2026-07-14): the per-scope
    /// storage routing is REAL — scan, read, and the single `atomic_write`
    /// all run under `self.scope`. The former shim delegated to the
    /// `Scope::dev()`-hard-coded pipeline and silently returned
    /// `rows_written = 0` for every real scope (proved live on Moon in the
    /// 2026-07-14 deep test). Soft-deleted rows are hidden from recall by
    /// the hydrate sys-gate (`lunaris_retrieve::hydrate`).
    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
    }

    /// Return a [`lunaris_retrieve::RetrievalBuilder`] bound to the engine's
    /// storage / embedder / keyword Arcs AND this wrapper's scope for
    /// DSL-style query composition.
    ///
    /// ```no_run
    /// use lunaris::{Keyword, Lunaris, LunarisError, Query, Scope, Vector};
    ///
    /// # async fn demo(engine: Lunaris, scope: Scope) -> Result<(), LunarisError> {
    /// let hits = engine.scoped(scope)
    ///     .dsl()
    ///     .with_root(Vector::new("chunks", 30).and(Keyword::bm25("chunks", 30)).fuse_rrf(60).top(5))
    ///     .execute(Query::text("brown fox"))
    ///     .await?;
    /// # Ok(()) }
    /// ```
    pub fn dsl(&self) -> lunaris_retrieve::RetrievalBuilder {
        // Wave 2.5C: pre-seed scope so all operators in the tree use the
        // bound scope rather than Scope::dev() placeholders.
        self.engine.recall().with_scope(self.scope.clone())
    }

    /// Phase 14.1 — signal the end of an agent turn, run the reflection pass,
    /// and apply MVCC invalidations for every ulid in [`ReflectOutput::invalidate`].
    ///
    /// ## What this does (Phase 14.1)
    ///
    /// 1. Delegates to the handle's [`ReflectSupervisor`] (same as
    ///    [`Lunaris::end_turn`]). If the supervisor is a
    ///    [`NoopReflectSupervisor`] (the default), the reflect call is a
    ///    cheap no-op returning `ReflectOutput::default()`.
    ///
    /// 2. For every ulid in `output.invalidate`, calls
    ///    [`apply_reflect_invalidate`] which:
    ///    - reads the fact row,
    ///    - stamps `bt.sys.1 = Some(now)` (JSON-patched into the payload),
    ///    - commits **one `atomic_write`** for the entire batch (D-11), and
    ///    - publishes one `AuditEvent::ReflectInvalidation` per stamped ulid
    ///      (D-22, fire-and-forget).
    ///
    /// ## What is NOT done in this commit (Phase 14.2 / 14.3)
    ///
    /// - `boost` — deferred to Phase 14.2 (ephemeral LRU per-handle cache).
    /// - `pre_warm_query` — deferred to Phase 14.3 (fire-and-forget recall).
    ///
    /// ## Failure discipline
    ///
    /// Reflect is advisory.  Supervisor errors and storage errors during
    /// invalidation are logged via `tracing::warn!` and swallowed — this
    /// method **never** fails the agent's next turn due to a reflect error.
    /// The full `ReflectOutput` (including `boost` and `pre_warm_query`) is
    /// returned to the caller regardless.
    ///
    /// ## Scope enforcement
    ///
    /// `self.scope` (the JWT-bound partition key) is the sole source of
    /// truth for the storage partition.  The caller cannot inject a different
    /// scope — that is the whole point of `ScopedLunaris`.
    pub async fn end_turn(&self, input: ReflectInput) -> Result<ReflectOutput, LunarisError> {
        let turn_id = input.turn_id;

        // Step 1: run the reflection pass (best-effort — never fail the turn).
        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()
            }
        };

        // Step 2 (Phase 14.1): apply invalidations — one atomic_write for the batch.
        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"
                    );
                }
            }
        }

        // Step 3 (Phase 14.2): populate the per-handle boost cache for every
        // chunk ulid nominated by the reflect supervisor.
        //
        // `apply_reflect_boost` is synchronous — it acquires the write lock,
        // writes all entries, and drops the guard before returning.  No `.await`
        // appears between the guard acquisition and its release, satisfying the
        // CLAUDE.md lock-across-await invariant.
        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"
            );
        }

        // Summary log at turn boundary (Phase 14.1 + 14.2 combined).
        // Step 3 (Phase 14.3): fire-and-forget speculative warm-up recall.
        //
        // If the reflector predicted a next-turn query, spawn a background task
        // to issue a real recall against the storage backend. This populates
        // Moon's FT page cache before the
        // agent issues the actual query, reducing first-hit latency on the next
        // turn.
        //
        // Design constraints (§4 of docs/design/phase-14-reflect-output-application.md):
        // - MUST NOT block `end_turn` — use `try_acquire_owned`, never
        //   `acquire_owned().await`.
        // - Concurrency is bounded by `engine.warm_up_semaphore` (default 4,
        //   configurable via `LUNARIS_PREWARM_CONCURRENCY`). Exhausted semaphore
        //   → skip + DEBUG log, never block.
        // - `OwnedSemaphorePermit` moves into the spawned task via
        //   `let _permit = permit;` INSIDE the `async move {}` block so it is
        //   released when the task ends, not when `end_turn` returns.
        // - Errors inside the task become `tracing::warn!` — never propagate,
        //   never panic.
        // - Warm-up uses the same `Scope` as this `ScopedLunaris` handle so no
        //   cross-tenant data can be accessed.
        if let Some(query_str) = output.pre_warm_query.clone() {
            match self.engine.warm_up_semaphore.clone().try_acquire_owned() {
                Ok(permit) => {
                    // Clone all Arcs needed by the spawned task before the move.
                    // `moon_storage` is included so the warm-up uses the Moon-native
                    // one-round-trip FT path when available — without it the task
                    // would take the generic retrieval path and miss the FT cache.
                    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 {
                        // PERMIT MOVE: `_permit` is dropped when this task ends,
                        // releasing the semaphore slot. It MUST live inside this
                        // `async move {}` block — placing it outside would release
                        // the permit when `end_turn` returns, defeating the bound.
                        let _permit = permit;
                        // Build a default Vector top-30 recall — the goal is to
                        // warm the backend's FT/page cache, not to return results
                        // to the caller. The default root is the same shape used
                        // by `Lunaris::recall()` and `ScopedLunaris::dsl()`.
                        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"
                    );
                }
            }
        }

        // Summary log at turn boundary (Phase 14.1 requirement).
        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)
    }

    /// ADD task activation-ledger — record usage signals into the
    /// persistent per-memory activation ledger.
    ///
    /// Read-modify-write of every DISTINCT id touched by `signals`: for each
    /// id, reads the existing `ActivationRecord` at
    /// `lunaris_core::keyspace::activation_key(scope, id)` (or starts from
    /// `ActivationRecord::default()` when none exists, or when the existing
    /// row is corrupt — a malformed stored record must not block new
    /// reinforcement), applies every signal for that id in input order via
    /// `ActivationRecord::apply`, and commits ALL touched records in exactly
    /// ONE `atomic_write` (mirrors D-11 — one atomic write per logical batch).
    ///
    /// ## Best-effort contract
    ///
    /// This method itself SURFACES storage errors (`Result::Err`) rather
    /// than swallowing them — it is a library primitive, not a turn-path
    /// caller. Callers on the agent-turn / injection path (e.g.
    /// `lunaris-hook::trace_injection`) MUST log-and-continue on `Err`: a
    /// reinforcement-signal failure must never fail the agent's turn (same
    /// contract as `apply_reflect_invalidate` / `apply_reflect_boost`).
    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();

        // Group signals by id, preserving first-seen order for a
        // deterministic WriteOp ordering in the batch.
        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 });
        }

        // Mirrors D-11: exactly ONE atomic_write for the whole batch.
        self.engine.storage.atomic_write(&self.scope, &ops).await.map_err(LunarisError::Storage)?;
        Ok(())
    }

    /// engram-soul-loop task 8b (`memory.distill`, `.add/tasks/distill/
    /// TASK.md` §3 CONTRACT, frozen) — archive every `id` in `ids`: RMW its
    /// [`lunaris_core::keyspace::activation_key`] row, set
    /// `archived_at = Some(now)`, and commit ALL touched records in exactly
    /// ONE batch write — same D-11 shape as [`Self::record_activation_refs`].
    ///
    /// Archive is activation drop, NOT a tombstone: this method never
    /// touches the episode itself (no `forget`/soft-delete). It only flips
    /// the ledger marker that [`lunaris_retrieve::LedgerBoostProvider`]
    /// (0 boost) and `lunaris_consolidate::dream::build_dream_agenda`
    /// (dropped from candidates) both read via
    /// [`lunaris_core::activation::ActivationRecord::is_archived`].
    ///
    /// - An `id` with NO existing ledger record is skipped (already
    ///   unboosted — nothing to mark) — this method never CREATES a record.
    /// - A corrupt existing record is skipped with a `tracing::warn!`
    ///   (mirrors [`Self::record_activation_refs`]'s corrupt-row handling)
    ///   rather than failing the whole batch.
    /// - Duplicate ids in `ids` are archived once (deduped defensively).
    /// - Returns the count of records ACTUALLY marked — ids skipped for
    ///   either reason above are not counted.
    /// - An empty `ids` slice is a no-op: `Ok(0)`, no storage call at all.
    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; // duplicate id in the input slice — archive once
            }
            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; // no ledger record — already unboosted, nothing to mark
            };
            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;
        }

        // Mirrors D-11 / record_activation_refs: exactly ONE batch write for
        // the whole call — and skip it entirely when nothing was touched.
        if !ops.is_empty() {
            self.engine
                .storage
                .atomic_write(&self.scope, &ops)
                .await
                .map_err(LunarisError::Storage)?;
        }
        Ok(marked)
    }

    /// engram-soul-loop task 8a (dream-agenda) — build a READ-ONLY
    /// distillation agenda: Leiden-clustered (or source-class-bucketed)
    /// candidate clusters of ripe raw episodes, with activation stats, for
    /// the coding-harness distiller to reason over. Never calls
    /// `atomic_write` — see `.add/tasks/dream-agenda/TASK.md` §3 CONTRACT.
    ///
    /// `now` is resolved here (`SystemTime::now()`, unix seconds) rather
    /// than threaded from the caller — the frozen §3 engine signature takes
    /// a plain `now: u64` (no live `HlcClock`), so this wrapper is the one
    /// place that turns "now" into a concrete wall-clock reading.
    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
    }

    /// engram-soul-loop task 6 (staleness-pass) — RMW upsert of verify-
    /// agenda entries.
    ///
    /// For each entry, reads the existing row at
    /// `lunaris_core::keyspace::verify_agenda_key(scope, entry.episode_id)`
    /// (if any) and preserves its `first_seen_ms` (a missing or corrupt
    /// existing row falls back to the caller-supplied `first_seen_ms` —
    /// a malformed stored agenda row must not block a fresh upsert), then
    /// commits every touched entry in exactly ONE `atomic_write` (mirrors
    /// [`Self::record_activation_refs`] / D-11: one atomic write per
    /// logical batch). An empty `entries` slice is a no-op — no
    /// `atomic_write` call at all.
    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 });
        }

        // Mirrors D-11: exactly ONE atomic_write for the whole batch.
        self.engine.storage.atomic_write(&self.scope, &ops).await.map_err(LunarisError::Storage)?;
        Ok(())
    }

    /// engram-soul-loop task 7 (verify-agenda-tools) — list every verify-
    /// agenda entry under this scope, freshest staleness first.
    ///
    /// Scans [`lunaris_core::keyspace::verify_agenda_prefix`] (mirrors
    /// [`crate::digest::recent_by_source`]'s `StreamExt::next` loop): a
    /// mid-stream storage error propagates, but a single corrupt/foreign row
    /// is skipped and never aborts the whole list (`.add/tasks/
    /// verify-agenda-tools/TASK.md` §1 Reject). Bounded by a 5_000-row scan
    /// cap (mirrors `lunaris-hook::staleness::SCAN_CAP`) — a warn-and-partial
    /// DoS guard for huge scopes. Results are sorted by `last_seen_ms` DESC
    /// (freshest staleness first).
    ///
    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 {
            // A mid-stream storage error propagates (the storage call
            // itself failed) — distinct from a corrupt VALUE, which is
            // skipped below without aborting the list.
            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)
    }

    /// engram-soul-loop task 7 (verify-agenda-tools) — remove one
    /// verify-agenda entry, returning whether it existed.
    ///
    /// Presence is checked via `read_as_of` on
    /// [`lunaris_core::keyspace::verify_agenda_key`]; when present, issues
    /// exactly ONE `WriteOp::KvDelete` `atomic_write` (mirrors D-19 — no
    /// write at all when the row was already absent, an idempotent no-op).
    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)
    }
}

/// Hard cap on rows scanned per [`ScopedLunaris::list_verify_agenda`] call
/// (mirrors `lunaris-hook::staleness::SCAN_CAP` = 5_000) — a DoS guard for
/// huge scopes; excess is a warn-and-partial list, never a hard failure.
const VERIFY_AGENDA_LIST_SCAN_CAP: usize = 5_000;

/// engram-soul-loop task 6 (staleness-pass) — one verify-agenda entry
/// (`.add/tasks/staleness-pass/TASK.md` §3 CONTRACT, task-7 wire shape —
/// KEEP STABLE, the MCP `verify_agenda` / `resolve` tools consume this
/// exact JSON shape).
///
/// `episode_id` doubles as the KV key's ULID
/// ([`lunaris_core::keyspace::verify_agenda_key`]) — one agenda row per
/// stale-anchored episode, RMW-upserted by
/// [`ScopedLunaris::upsert_verify_agenda`].
#[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,
}

// ── llama.cpp-only cutover: embedder + reranker resolution ────────────────────
//
// The supported runtime is in-process llama.cpp + the frozen GGUF pair
// `granite-embedding-311m-multilingual-r2.Q4_K_M` (embedder) and
// `bge-reranker-v2-m3.Q5_K_M` (reranker). The knobs are:
//
// - `LUNARIS_EMBEDDER_GGUF` — path to the embedder GGUF; default is the
//   `~/.lunaris/models/` staged artifact.
// - `LUNARIS_RERANKER_GGUF` — same for the reranker GGUF.
// - `LUNARIS_DEVICE=cpu` — force CPU even on Metal-enabled builds.
// - `LUNARIS_EMBEDDER_OPENAI_URL` / `LUNARIS_EMBEDDER_OLLAMA_URL` — remote
//   embedder endpoints; only consulted when the `embed-remote` feature is
//   enabled (Tier-0 / air-gap path).
// - `LUNARIS_EMBEDDER_DIR` — legacy dir override; still consulted for
//   `tokenizer.json` by the BPE token counter (`make_token_counter`).
// - `LUNARIS_EMBED_DIM` — only applies when the resolver falls back to
//   `NoopEmbedder` (no GGUF staged); default 768.
//
// One-shot tracing::info! per process logs the resolved backend + path; if
// the embedder falls back to noop the operator gets a tracing::warn! banner.

/// Optional override for the directory holding the granite-r2 model
/// artifacts. Default: `<cache-dir>/lunaris/models/granite-embedding-311m-multilingual-r2/`.
/// Expected layout: `model.safetensors`, `tokenizer.json`, `config.json`.
pub const EMBEDDER_DIR_ENV_VAR: &str = "LUNARIS_EMBEDDER_DIR";

/// Optional override for the directory holding the bge-reranker-v2-m3 model
/// artifacts. Default: `<cache-dir>/lunaris/models/bge-reranker-v2-m3/`.
pub const RERANKER_DIR_ENV_VAR: &str = "LUNARIS_RERANKER_DIR";

/// Optional path override for the embedder Q4_K_M GGUF (llama.cpp runtime).
/// Default: the `~/.lunaris/models/` staged artifact.
pub const EMBEDDER_GGUF_ENV_VAR: &str = "LUNARIS_EMBEDDER_GGUF";

/// Optional path override for the reranker Q5_K_M GGUF (llama.cpp runtime).
pub const RERANKER_GGUF_ENV_VAR: &str = "LUNARIS_RERANKER_GGUF";

/// Env var that controls the dim of the `NoopEmbedder` fallback used when
/// the granite-r2 weights are missing AND the operator has not supplied a
/// custom embedder via [`Lunaris::with_embedder`]. Positive integer; default
/// [`lunaris_core::NOOP_DEFAULT_DIM`] (768).
pub const EMBED_DIM_ENV_VAR: &str = "LUNARIS_EMBED_DIM";

/// Env var that controls the maximum number of concurrent speculative warm-up
/// recall tasks spawned by [`ScopedLunaris::end_turn`] (Phase 14.3).
/// Must be a positive integer. `0`, non-numeric, or unset values fall back to
/// the default of `4`. One `tracing::info!` is emitted per process when the
/// capacity is resolved.
pub const PREWARM_CONCURRENCY_ENV_VAR: &str = "LUNARIS_PREWARM_CONCURRENCY";

/// Default semaphore capacity for speculative warm-up recalls.
const PREWARM_CONCURRENCY_DEFAULT: usize = 4;

/// Env var that controls the exact-text embedding cache capacity.
///
/// Set to `0` to disable the cache. The default is intentionally modest:
/// enough for repeated agent prompts, context-injection recalls, and common
/// chunk text, but bounded so long-running agents do not grow without limit.
pub const EMBED_CACHE_CAPACITY_ENV_VAR: &str = "LUNARIS_EMBED_CACHE_CAPACITY";

const EMBED_CACHE_CAPACITY_DEFAULT: usize = 2048;

/// Resolve the warm-up semaphore capacity from [`PREWARM_CONCURRENCY_ENV_VAR`].
///
/// Non-numeric, `0`, and unset values all return `PREWARM_CONCURRENCY_DEFAULT`
/// with a `tracing::warn!` for non-numeric/zero inputs. Negative values are
/// impossible since we parse as `usize`. A one-shot `tracing::info!` is emitted
/// per process on the resolved capacity.
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();

/// Which embedder backend [`Lunaris::open`] actually resolved in this process.
///
/// W1.2 (2026-08-21). The `Noop` arm is a **silent** degradation: every vector
/// is zeros, so hybrid recall collapses to BM25 + insertion-order tie-breaks
/// while every surface keeps answering `200`. Until this enum existed the only
/// evidence was one `tracing::warn!` fired once per process — which
/// `lunaris-server`'s `/readyz` could not see, so a server built without
/// `llamacpp` (the workspace entry sets `default-features = false`) reported
/// itself READY with a zero-vector embedder. `dim()` cannot distinguish the
/// cases: `NoopEmbedder` reports a non-zero dim on purpose so the operator's
/// existing `FT.CREATE` index geometry stays valid.
///
/// This is deliberately a *structural* signal. A probe must never call
/// `embed_batch` to find out — a wedged ggml pool cannot be cancelled from
/// async Rust, so an inference-based probe wedges on a timer forever (see
/// `lunaris-server/src/readiness.rs` module docs).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum EmbedderBackend {
    /// In-process llama.cpp over a staged GGUF — the shipped local path.
    LlamaCpp,
    /// Remote OpenAI-compatible `/embeddings` endpoint.
    OpenAiRemote,
    /// Remote Ollama endpoint (operator escape hatch).
    OllamaRemote,
    /// Zero-vector fallback. Real vectors are NOT being produced.
    Noop,
    /// `Lunaris::open` has not run in this process — the handle was built
    /// through a `with_parts*` test seam, so no backend was resolved. Callers
    /// must treat this as "unknown", never as "degraded".
    Unresolved,
}

impl EmbedderBackend {
    /// Stable lowercase identifier, safe to compare against across releases.
    ///
    /// This is the SDK-facing spelling: the Python and TypeScript bindings
    /// cannot carry a Rust enum, so `Lunaris::embedder_backend` hands them
    /// this string. Treat these values as API — changing one is a breaking
    /// change for every caller doing `if backend == "noop"`.
    #[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",
        }
    }

    /// Whether this backend produces real vectors.
    ///
    /// `Noop` does not — every vector is zeros, so hybrid recall silently
    /// collapses to BM25 plus insertion-order tie-breaks while every surface
    /// keeps answering successfully. `Unresolved` means `open` has not run in
    /// this process, which is "unknown", NOT "degraded", so it answers `true`
    /// here rather than raising a false alarm in a test-seam handle.
    #[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();

/// Environment override that silences `announce_degradation_once`.
pub const SUPPRESS_DEGRADED_WARNING_ENV: &str = "LUNARIS_SUPPRESS_DEGRADED_WARNING";

/// What [`Lunaris::open`] should do about the backend it just resolved.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DegradationNotice {
    /// Say nothing — healthy, unknown, suppressed, or already covered by a
    /// subscriber that will receive the `tracing::warn!`.
    Silent,
    /// Write this to stderr, once per process.
    Emit(String),
}

/// Decide whether a resolved backend should announce itself on stderr.
///
/// W0.7 successor. `embedder_backend()` made degradation *queryable*, which
/// still requires the caller to know to ask. The `tracing::warn!` on the Noop
/// path is real but reaches nobody in an SDK process: neither `lunaris-py` nor
/// `lunaris-ts` installs a subscriber, so for a `pip install lunaris` user it is
/// emitted into a void — and that is precisely the population whose symptom is
/// "recall returns nothing and every call succeeded".
///
/// Split out as a pure function on purpose. The alternative — reading the env
/// and the dispatcher inside `open` — is untestable without `env::set_var`,
/// which edition 2024 makes `unsafe` and which races every sibling test in the
/// same binary through code that never names the variable.
///
/// `subscriber_installed` should come from `tracing::dispatcher::has_been_set()`:
/// a host that set one up already gets the `warn!` through its own routing, and
/// printing to stderr as well would both double-report and bypass that routing.
#[must_use]
pub fn degradation_notice(
    backend: EmbedderBackend,
    subscriber_installed: bool,
    suppress_env: Option<&str>,
) -> DegradationNotice {
    // `Unresolved` is "open never ran here", not "degraded" — see
    // `produces_real_vectors`, which returns true for it for the same reason.
    if backend.produces_real_vectors() || subscriber_installed {
        return DegradationNotice::Silent;
    }
    // Key on the ACCEPTED SET, never on presence: an `is_some()` check would let
    // `LUNARIS_SUPPRESS_DEGRADED_WARNING=0` — and the empty string a shell
    // produces for an exported-but-unset var — silence the one warning the user
    // most needs.
    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."
    ))
}

/// Apply [`degradation_notice`] for the process-resolved backend, at most once.
///
/// Called at the end of [`Lunaris::open`] — the single seam both SDKs route
/// through, so neither needs its own copy and `generated.rs` stays untouched.
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 {
        // Once per process: the resolution itself is process-global, so warning
        // twice would imply two independent decisions were made.
        if DEGRADATION_ANNOUNCED.set(()).is_ok() {
            eprintln!("{msg}");
        }
    }
}

/// The embedder backend [`Lunaris::open`] resolved, or
/// [`EmbedderBackend::Unresolved`] if `open` has not run in this process.
///
/// Process-global because `resolve_embedder` is process-global: it reads env
/// and the model cache, and a process runs one server. Set once, on the first
/// `open`.
#[must_use]
pub fn resolved_embedder_backend() -> EmbedderBackend {
    RESOLVED_EMBEDDER_BACKEND.get().copied().unwrap_or(EmbedderBackend::Unresolved)
}

/// Granite-r2 model directory name under `<cache>/lunaris/models/`.
const GRANITE_R2_DIR: &str = "granite-embedding-311m-multilingual-r2";
/// bge-reranker-v2-m3 model directory name under `<cache>/lunaris/models/`.
/// Referenced by the dir-layout unit test (the FP32 dir also provides the
/// canonical `tokenizer.json` location some operator tooling still stages).
#[cfg(test)]
const BGE_RERANKER_DIR: &str = "bge-reranker-v2-m3";

/// Resolve the canonical cache directory for a named model artifact. Returns
/// `<cache_dir>/lunaris/models/<name>/`, or `./lunaris/models/<name>/` when
/// `dirs::cache_dir()` is unavailable (rare on Unix/macOS — surfaced as a
/// warning to operators of stripped-down environments).
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)
}

/// Resolve the embedder model directory from [`EMBEDDER_DIR_ENV_VAR`],
/// falling back to the default cache layout.
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))
}

/// Resolve the default embedder for [`Lunaris::open`] (llama.cpp-only
/// cutover). Tries:
///
/// 1. (feature `llamacpp`, default) `LUNARIS_EMBEDDER_GGUF` or the
///    `~/.lunaris/models/` staged Q4_K_M GGUF via
///    [`lunaris_llamacpp::LlamaCppEmbedder`].
/// 2. (feature `embed-remote`) `LUNARIS_EMBEDDER_OPENAI_URL`
///    (OpenAI-compatible `/v1/embeddings`) then `LUNARIS_EMBEDDER_OLLAMA_URL`
///    — the Tier-0 no-C++-toolchain remote path.
/// 3. Otherwise, emit a `tracing::warn!` and fall back to [`NoopEmbedder`]
///    at [`lunaris_core::NOOP_DEFAULT_DIM`] so the rest of the open path
///    completes (vector recall returns empty rows; operator sees the banner
///    and can stage the GGUF).
// `max_batch_tokens` feeds only the llamacpp opts; a Tier-0 (no-inference)
// build compiles that branch out, so the param is legitimately unused there.
#[cfg_attr(not(feature = "llamacpp"), allow(unused_variables))]
async fn resolve_embedder(max_batch_tokens: u32) -> Result<Arc<dyn Embedder>, LunarisError> {
    // 0. llama.cpp GGUF embedder (cutover Phase B) — wins whenever the
    //    feature is compiled in AND the GGUF artifact is reachable
    //    (LUNARIS_EMBEDDER_GGUF, else the ~/.lunaris/models/ staged
    //    default). Missing artifact or open failure falls through to the
    //    remote/Noop chain.
    #[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()
            };
            // Weight load + context creation are synchronous — keep them off
            // the runtime worker.
            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"
                    );
                }
            }
        }
    }

    // 1. Remote OpenAI-compatible embedder (`POST /v1/embeddings`) — the
    //    supported remote path when no local GGUF is reachable. Selected
    //    when LUNARIS_EMBEDDER_OPENAI_URL is set; wins over the Ollama hatch.
    #[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>);
        }
    }

    // 1b. Ollama HTTP escape hatch — legacy remote path.
    #[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>);
        }
    }

    // 2. No local runtime reachable — NoopEmbedder (zero vectors). Rows
    //    ingested in this state are written WITHOUT a `vec` field (see
    //    `lunaris_storage_moon::atomic::unindexable_reason`), so they are
    //    absent from vector recall but still reachable by BM25 and still
    //    hydratable; a later real embedding for the same id promotes them
    //    into the KNN index.
    //
    //    That skip is load-bearing, not tidiness. A zero vector is NOT a
    //    neutral placeholder: under the `1/(1+d)` score it sits at distance
    //    `||q||` from any unit query and OUTRANKS genuine matches, forever
    //    (F22). Before the skip, this comment claimed vector recall returned
    //    empty rows here — it did not, and the false comment is part of why
    //    the defect went unnoticed.
    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>)
}

/// Parse a batch-token budget from an env value, falling back to `default` for
/// any absent / empty / non-numeric / below-floor input. The `>= 16` floor
/// mirrors the embedder's own `budget.max(16)` guard so a bogus env can never
/// produce a degenerate llama context. Pure (env read stays in the wrappers) so
/// tests need no `env::set_var` — the crate is edition-2024 where that is unsafe.
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)
}

/// General-purpose embedder batch-token budget (`Lunaris::open`): default 4096,
/// overridable via `LUNARIS_EMBED_MAX_BATCH_TOKENS`. Bench/ingest want a large
/// window so long documents embed without truncation.
fn embed_max_batch_tokens() -> u32 {
    parse_batch_tokens(std::env::var("LUNARIS_EMBED_MAX_BATCH_TOKENS").ok(), 4096)
}

/// contextd (interactive) embedder batch-token budget: default 1024, overridable
/// via `LUNARIS_CONTEXT_EMBED_MAX_BATCH_TOKENS`. A long-lived daemon that only
/// embeds short hook captures does not need the 4096 throughput window, whose
/// llama.cpp compute-buffer reservation cost ~2.5 GB (the 2026-07-14 contextd
/// footprint); 1024 reserves ~1.1 GB with no truncation of real captures.
fn context_embed_max_batch_tokens() -> u32 {
    parse_batch_tokens(std::env::var("LUNARIS_CONTEXT_EMBED_MAX_BATCH_TOKENS").ok(), 1024)
}

/// Resolve the process-default embedder (the same llama.cpp GGUF → remote →
/// Noop chain [`Lunaris::open`] uses internally), exposed so a long-lived host
/// (e.g. `lunaris-contextd`) can load it ONCE and share the resulting
/// `Arc<dyn Embedder>` across many per-scope [`Lunaris`] handles via
/// [`Lunaris::open_with_embedder`] — instead of loading a full resident GGUF
/// model per scope (the 7.32 GB contextd RSS leak, 2026-07-14).
pub async fn resolve_default_embedder() -> Result<Arc<dyn Embedder>, LunarisError> {
    resolve_embedder(context_embed_max_batch_tokens()).await
}

/// Deferred-load twin of [`resolve_default_embedder`] (unified-inference,
/// 2026-07-19). The returned handle resolves NOTHING at construction — the
/// full GGUF resolve chain only runs on the first `embed_batch` call, and the
/// result is cached for the process lifetime.
///
/// This exists for hosts that normally NEVER embed in-process: `lunaris-mcp`
/// proxies every embed-needing op to the warm `lunaris-contextd` daemon, so an
/// eager boot-time load parks a second resident copy of the weights (and a
/// second llama.cpp threadpool) next to contextd's — the double-residency
/// found in the 2026-07-19 CPU investigation. With this handle the local
/// weights only materialize if an embed op must genuinely be served in-process
/// (standalone npx/uvx installs, or contextd unreachable).
///
/// The deferred resolve runs the same GGUF → remote → Noop chain as
/// [`resolve_default_embedder`] and caches whatever it lands on. It does NOT
/// hard-error on a `NoopEmbedder` fallback: **ingest** legitimately runs
/// without a dense embedder (the KV + BM25 write still succeeds; only vector
/// recall degrades), exactly as the pre-lazy `NoopEmbedder` path did. The
/// "no embedder → loud error instead of silent empty hits"
/// (`mcp-recall-empty-hits`) guard lives on the **recall** path
/// (`lunaris_memory_service::recall::handle`), which is the only caller for
/// which a zero query vector is a silent-failure — ingest storing a
/// zero-vector is degraded-but-useful, not a lie.
pub fn lazy_default_embedder() -> Arc<dyn Embedder> {
    Arc::new(LazyDefaultEmbedder { cell: tokio::sync::OnceCell::new() })
}

/// See [`lazy_default_embedder`].
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 {
        // Loaded → the real backend's dim. Unloaded → the configured default
        // (LUNARIS_EMBED_DIM, else 768 — granite-r2's width), WITHOUT forcing
        // a load: dim() is called on cold paths (index bootstrap) that must
        // not pull weights in.
        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> {
        // Forward to the inner lowpri lane — the trait default would route
        // through embed_batch and head-of-line-block interactive recall.
        self.get_or_load().await?.embed_batch_lowpri(inputs).await
    }
}

/// Resolve the NoopEmbedder fallback dim from [`EMBED_DIM_ENV_VAR`].
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
}

/// Resolve the default reranker for [`Lunaris::open`] (llama.cpp-only
/// cutover). Tries:
///
/// 1. (feature `llamacpp`, default) `LUNARIS_RERANKER_GGUF` or the
///    `~/.lunaris/models/` staged Q5_K_M GGUF, deferred-loaded via
///    `LazyLlamaCppReranker` (N-04 D1).
/// 2. Otherwise fall back to [`NoopReranker`] per the RETRIEVE-06 contract —
///    the recall path runs end-to-end even without the cross-encoder pass.
async fn resolve_reranker() -> Result<Arc<dyn Reranker>, LunarisError> {
    // 0. llama.cpp GGUF reranker (cutover Phase B) — same precedence rule as
    //    the embedder. Load is DEFERRED to the first `rerank()` call via
    //    `LazyLlamaCppReranker` (N-04 D1: the recall hot path may never
    //    reach the rerank stage; don't pay the weight load + context RSS at
    //    open()). Pre-flight only checks the artifact exists so a typo'd
    //    path still falls through to Noop immediately.
    #[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>);
        }
    }

    // 1. No local runtime reachable — NoopReranker (rerank pass skipped per
    //    RETRIEVE-06). Stage the llama.cpp GGUF for a real 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>)
}

/// Resolve the process-default reranker (the same lazy llama.cpp GGUF → Noop
/// chain [`Lunaris::open`] uses), exposed alongside [`resolve_default_embedder`]
/// so a long-lived host can load it ONCE and share the `Arc<dyn Reranker>`
/// across per-scope handles via [`Lunaris::with_reranker`]. Like the embedder,
/// the reranker model is scope-independent; a per-scope reranker was the other
/// half of the 7.32 GB contextd RSS growth (2026-07-14). The returned reranker
/// is lazy — its GGUF still loads on the first `rerank()`, now exactly once.
pub async fn resolve_default_reranker() -> Result<Arc<dyn Reranker>, LunarisError> {
    resolve_reranker().await
}

/// Plan 03-03: Construct the default extractor for [`Lunaris::open`].
///
/// Callers wire their own extractor via [`Lunaris::with_extractor`] or
/// `handle.graph_pipeline().set_extractor(extractor)` for late binding.
/// Cutover decision 1 (llama.cpp-only, 2026-07-10): extraction is going
/// remote-only. When `LUNARIS_EXTRACT_PROVIDER` names a provider
/// (anthropic|openai|gemini|minimax|openai-compat), construct the
/// `CloudApiExtractor` from env — model via `<PROVIDER>_EXTRACT_MODEL`,
/// key via `<PROVIDER>_API_KEY` (optional for openai-compat), base URL via
/// `LUNARIS_OPENAI_COMPAT_BASE_URL` — wrapped in the production fallback
/// floor. A set-but-broken provider degrades to `NoopExtractor` with a
/// warn (config error must NOT silently fall back to a different backend).
/// Returns `None` when the env is unset → caller continues its chain.
#[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>)
        }
    }
}

/// Verifier twin of [`remote_extractor_from_env`], keyed on
/// `LUNARIS_VERIFY_PROVIDER` (`lunaris_verify::cloud_api::ENV_PROVIDER`).
#[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>)
        }
    }
}

/// Remote-only (llama.cpp cutover, Phase C): a remote provider env resolves
/// a real extractor; otherwise degraded mode. Production callers wanting
/// graph extraction either set `LUNARIS_EXTRACT_PROVIDER` or pass a custom
/// [`Extractor`] impl (e.g., [`lunaris_extract::OllamaExtractor`] under the
/// `ollama` feature) via [`Lunaris::with_extractor`].
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>
}

/// Plan 04-04: Construct the default verifier for [`Lunaris::open`].
///
/// Remote-only (llama.cpp cutover, Phase C): `LUNARIS_VERIFY_PROVIDER`
/// resolves a real remote verifier; otherwise [`NoopVerifier`] per the D-02
/// default-OFF contract. Callers wire their own verifier via
/// [`Lunaris::with_verifier`].
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>
}

/// Plan 04-04 + Phase 16-01 (CONSOL-V1-01): Construct the default consolidator
/// for [`Lunaris::open`], resolving from
/// [`ConsolidatorPipelineHandle::BACKEND_ENV_VAR`] (`LUNARIS_CONSOLIDATOR_BACKEND`).
///
/// Default (env unset) → [`lunaris_consolidate::ActRConsolidator`] (production
/// default per CONSOL-V1-01). Operators opt out to [`NoopConsolidator`] by
/// setting `LUNARIS_CONSOLIDATOR_BACKEND=noop` (preserved third toggle surface:
/// code override via [`Lunaris::with_consolidator`] also still works).
///
/// Unknown env values fail-fast via [`LunarisError::Storage`]
/// (`StorageError::Backend`) — NO silent fallback.
fn default_consolidator() -> Result<Arc<dyn Consolidator>, LunarisError> {
    ConsolidatorPipelineHandle::backend_from_env()
}

// ── llama.cpp cutover Phase B — path resolution + lazy reranker ─────────────

/// The staged artifacts under `~/.lunaris/models/`, named by the ONE catalogue
/// every stager reads.
///
/// W0.7: the filenames used to be literals here and again in each of the two
/// stagers. Staging 253 MB under a name this lookup does not consult is a
/// silent no-op that presents as success, so the agreement is now structural
/// — there is a single `filename()`, and both sides call it.
#[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;

/// Resolve a llama.cpp GGUF artifact: env override first, then the staged
/// `~/.lunaris/models/` default. Returns `None` (→ caller falls through to
/// the candle chain) unless the file actually exists.
#[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;
    }
    // Same resolution the stager writes to — one function, so a staged file
    // and the lookup for it cannot land in different directories.
    lunaris_core::models::staged_path(kind).filter(|p| p.exists())
}

/// GPU offload for the llama.cpp backends: everything when the `metal`
/// feature is compiled in (Apple Silicon default), nothing otherwise.
/// `LUNARIS_DEVICE=cpu` is the operator kill-switch, mirroring the candle
/// `device_select` contract.
#[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 }
}

/// Deferred-load wrapper for [`lunaris_llamacpp::LlamaCppReranker`]
/// (N-04 D1 rationale: the Q5_K_M weights + warm context only materialize
/// on the first `rerank()` call; `applies()` answers `true` eagerly because
/// config promises a real reranker). On init failure the OnceCell stays
/// empty so a later call can retry.
#[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 {
        // Config promises a real reranker — answer eagerly so
        // `Hit { rerank_applied }` doesn't lie on cold paths.
        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
    }
}

// ── v0.4 N-03 — unit tests for env-var resolution (cache-dir layout) ─────────
//
// `resolve_embedder()` / `resolve_reranker()` are async + perform I/O; the
// unit tests below cover only the pure path-resolution helpers and the
// `resolve_embed_dim()` parser, which are deterministic and side-effect-free.
// Construction of the real llama.cpp embedder/reranker is exercised by
// lunaris-llamacpp's own integration tests + the `llamacpp_wired` test.
#[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() {
        // Pin the v0.4 env-var surface area so accidental renames surface in
        // review. Operators wire these strings into Helm charts / k8s
        // manifests; renaming silently breaks deployments.
        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"
        );
    }

    // contextd-embed-budget (2026-07-14): contextd's embedder llama.cpp context
    // reserved ~2.5 GB because max_batch_tokens=4096 (the n_ubatch compute
    // buffer). contextd only embeds short hook captures, so it uses a smaller
    // budget (default 1024 → ~1.1 GB) while Lunaris::open keeps 4096.

    #[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() {
        // Read-only: the test process does not set these env vars, so the
        // wrappers return their defaults (no env::set_var — edition-2024 unsafe).
        assert_eq!(context_embed_max_batch_tokens(), 1024);
        assert_eq!(embed_max_batch_tokens(), 4096);
    }
}

// ── Phase 13 unit tests — end_turn / ReflectSupervisor wire-up ──────────────
//
// All tests use stub storage + `StubEmbedder` from lunaris_core (proven by
// the existing verify_pipeline_smoke integration tests) so no I/O is
// performed. The three tests cover:
//   1. Default Noop supervisor → empty ReflectOutput.
//   2. Custom stub supervisor → output propagates; input fields thread through.
//   3. Supervisor returning Err → end_turn propagates the error.
#[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;

    // ── minimal stub storage (matches actual StoragePort signatures) ──────────

    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 {
        // HlcClock::new already returns Arc<HlcClock> — no extra Arc::new wrap.
        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)
    }

    // ── test 1: default noop supervisor → empty output ────────────────────────

    #[tokio::test]
    async fn end_turn_noop_returns_empty_output() {
        let handle = make_handle();
        // Default is NoopReflectSupervisor — applies() = false.
        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());
    }

    // ── test 2: custom stub supervisor → output + input fields thread through ─

    /// Captures the input it received so the test can assert field propagation.
    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);

        // Confirm the supervisor received the exact input we passed.
        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");
    }

    // ── test 3: supervisor returns Err → end_turn propagates error ────────────

    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 {
    //! W0.7 successor — an SDK caller must be able to SEE a degraded embedder.
    //!
    //! The `Noop` fallback is silent by construction: every vector is zeros,
    //! so hybrid recall collapses to BM25 plus insertion-order tie-breaks
    //! while `recall` keeps answering with a plausible hit list, and
    //! `NoopEmbedder::dim()` reports a real dimension on purpose so the index
    //! geometry stays valid. Nothing about the results reveals it. Until
    //! `embedder_backend()` existed, `resolved_embedder_backend()` was
    //! Rust-only and `grep -rn degraded crates/lunaris-py/src
    //! crates/lunaris-ts/src` returned nothing at all.

    use super::{EmbedderBackend, resolved_embedder_backend};

    /// The strings are API — both SDKs compare against them, so a rename is a
    /// breaking change for every caller doing `if backend == "noop"`.
    #[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:?}");
        }
    }

    /// Distinctness is what makes the tag usable as a discriminator at all.
    #[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:?}");
    }

    /// `Noop` is the only backend that does not produce real vectors.
    ///
    /// `Unresolved` deliberately answers `true`: it means `open` has not run
    /// in this process (a `with_parts*` test seam), which is "unknown", not
    /// "degraded". Reporting a test-seam handle as degraded would train
    /// callers to ignore the signal.
    #[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");
        }
    }

    /// The handle accessor must report the SAME thing the process-global
    /// resolver does. If it drifted, the SDKs would be reading a second
    /// opinion — and the one the engine actually uses would be the other one.
    #[test]
    fn the_handle_accessor_agrees_with_the_process_resolver() {
        // No `open` in this test binary, so this is `Unresolved` — the point
        // is the agreement, not the value.
        assert_eq!(resolved_embedder_backend().as_str(), resolved_embedder_backend().to_string());
    }
}