velesdb_memory/context/

memory_bridge.rs

//! The context compiler's memory bridge: memory-backed fragment selection,
//! recoverable sources, aggregatable compilation events, and persisted
//! working contexts — the `MemoryService` half of EPIC-P-070's US-002.
//!
//! Everything the bridge persists is a **system fact**: hub-marked
//! (`_veles_hub`) and carrying **only reserved `_veles_*` metadata keys**, so
//! it is invisible to unfiltered recall (hub exclusion), can never match a
//! caller's include filter (callers cannot name reserved keys), and can never
//! be forged by a caller fact (reserved keys are rejected at `remember`).
//! Stored ids are salted, and both the source writer and the handle resolver
//! verify the `_veles_ctx_source` marker, so a caller fact squatting a salt
//! preimage is neither overwritten nor ever served back as a source. Events
//! carry metadata and hashes only — never fragment content. Event recording
//! stamps wall-clock time; the compile pipeline itself stays clock-free and
//! deterministic.

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(not(target_arch = "wasm32"))]
use std::time::{SystemTime, UNIX_EPOCH};

/// Wall-clock nanos since the Unix epoch, stamped on savings events only —
/// never in the compile pipeline. On `wasm32-unknown-unknown`
/// `SystemTime::now()` aborts (`std` has no clock there), so events carry 0:
/// the per-process sequence alone uniquifies their ids, and wasm stats are
/// per-session by design (in-memory store).
fn now_nanos() -> u128 {
    #[cfg(target_arch = "wasm32")]
    {
        0
    }
    #[cfg(not(target_arch = "wasm32"))]
    {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|elapsed| elapsed.as_nanos())
            .unwrap_or(0)
    }
}

/// Current Unix time in seconds — used only by
/// [`MemoryService::should_upgrade_ttl`]'s extension-only comparison (the
/// storage/expiry layer; the `compile` pipeline itself stays clock-free). On
/// `wasm32-unknown-unknown` this is 0 (no clock, mirrors [`now_nanos`]); the
/// wasm `MemoryStore` is in-memory only, so a stored durable expiry (a real
/// epoch second count) never actually exists there for 0 to be compared
/// against.
fn now_unix_secs() -> u64 {
    #[cfg(target_arch = "wasm32")]
    {
        0
    }
    #[cfg(not(target_arch = "wasm32"))]
    {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|elapsed| elapsed.as_secs())
            .unwrap_or(0)
    }
}

use serde_json::{Map, Number, Value};

use super::{positive_ttl, MemoryService, Metadata, HUB_FIELD};
use crate::context::model::{
    CompilePolicy, CompileRequest, CompiledContext, ContextDecision, ContextFragment,
    ContextSavings, ContextSource, ImportanceWeights, LoadedWorkingContext, MediaRef, MemoryScope,
    WorkingContext, WorkingContextIndex, WorkingContextSession,
};
use crate::context::{media, provenance, ContextCompiler};
use crate::embedder::Embedder;
use crate::error::MemoryError;
use crate::id::stable_id;
use crate::model::FusionOptions;
use crate::storage::MemoryStore;

/// Salt for stored source ids — disjoint from natural fact ids, so a caller
/// later remembering the same text can never overwrite a stored source (or
/// inherit its system marker).
const SOURCE_ID_SALT: &str = "veles-ctx-source:";
/// Salt for compilation-event ids.
const EVENT_ID_SALT: &str = "veles-ctx-event:";
/// Salt for working-context ids (deterministic per project+session, so a
/// save is an idempotent upsert).
const WORKING_ID_SALT: &str = "veles-ctx-working:";
/// Salt for a project's working-context index id (deterministic per
/// project, so every `save_working_context` call updates the SAME system
/// fact rather than minting a new one).
const WORKING_INDEX_ID_SALT: &str = "veles-ctx-working-index:";

/// The constant lexical anchor every event's content starts with, so one
/// vector query can sweep the event family for aggregation.
const EVENT_ANCHOR: &str = "veles context compilation event";

/// Reserved metadata keys of the bridge's system facts. Reserved (`_veles_`)
/// on purpose: callers can neither set them (forgery) nor filter on them, and
/// [`MemoryService::context_savings`] aggregates only genuine events (it
/// filters at the storage layer, below the caller-facing validation).
///
/// Being unfilterable was once claimed here to make these facts "invisible to
/// every caller-facing recall path". It did not (#1737). A caller cannot
/// filter ON a reserved key, but `field != value` MATCHES a fact that has no
/// such field — and a system fact has none of the caller's columns, so every
/// `!=` predicate swept all of them in. Invisibility is now an exclusion
/// [`crate::storage::INTERNAL_MARKER_FIELDS`] states and each backend
/// applies, not a side effect of the naming rule.
///
/// The four markers below are therefore imported rather than redeclared: they
/// ARE entries of that list, and a local copy could drift from it silently.
use crate::storage::{
    CTX_EVENT_FIELD, CTX_SOURCE_FIELD, CTX_WORKING_FIELD, CTX_WORKING_INDEX_FIELD,
};

const CTX_PROJECT_FIELD: &str = "_veles_ctx_project";
const CTX_MODEL_FIELD: &str = "_veles_ctx_model";
/// A stored source's media payload (US-009, PR2): `{"mime", "bytes_b64"}`,
/// the exact [`MediaRef`] shape, set only when the source fragment carried
/// one. Reserved like every other `_veles_ctx_*` key — a caller can neither
/// set nor filter on it.
const CTX_SOURCE_MEDIA_FIELD: &str = "_veles_ctx_source_media";
/// The durable-TTL payload key set by [`super::positive_ttl`]-backed writes
/// (`store_with_ttl`, via `store_fact`). Mirrors `velesdb_core::EXPIRES_AT_KEY`
/// as a literal rather than an import: that re-export is `persistence`-gated,
/// and this module (unlike `NativeStore`) must keep compiling under `context`
/// alone (e.g. `velesdb-wasm`, which never enables `persistence`).
const EXPIRES_AT_FIELD: &str = "_veles_expires_at";
const CTX_SESSION_FIELD: &str = "_veles_ctx_session";
const CTX_TOKENS_IN_FIELD: &str = "_veles_ctx_tokens_in";
const CTX_TOKENS_OUT_FIELD: &str = "_veles_ctx_tokens_out";
const CTX_TOKENS_SAVED_FIELD: &str = "_veles_ctx_tokens_saved";
const CTX_COST_FIELD: &str = "_veles_ctx_cost_micros";
const CTX_CURRENCY_FIELD: &str = "_veles_ctx_currency";
const CTX_AT_FIELD: &str = "_veles_ctx_at";

/// Per-process sequence folded into event ids so two compilations landing on
/// the same clock tick (coarse timers, concurrent calls) never collide.
static EVENT_SEQ: AtomicU64 = AtomicU64::new(0);

/// Serializes the read-modify-write of the per-project working-context index.
///
/// The index is ONE fact per project, rewritten wholesale on every
/// `save_working_context`. Without this, two saves racing on the same project
/// both read the same pre-state and the second write erases the first
/// session's entry — a silent loss: the erased session's own fact is still on
/// disk and still loadable by exact id, but `list_working_contexts` (and
/// therefore `load_working_context`'s `other_sessions` recovery hint) no
/// longer knows it exists, and nothing anywhere returns an error.
///
/// **Scope, honestly: this is an INTRA-PROCESS lock only.** Two processes
/// opening the same store still race, because nothing below this layer offers
/// a compare-and-swap. The durable fix is a CAS or a transaction on the
/// [`MemoryStore`] trait itself; until then, the single-process case (the MCP
/// server, whose `spawn_blocking` handlers are exactly what made this
/// reachable) is covered and the multi-process case is not.
///
/// One global lock rather than one per project: index writes are rare (one
/// per `save_working_context`), so the contention is negligible, whereas a
/// `HashMap<String, _>` keyed by caller-supplied project names is an unbounded
/// slow leak for no measurable gain. Per-project striping is the obvious
/// upgrade if index writes ever become hot.
static WORKING_INDEX_WRITE: parking_lot::Mutex<()> = parking_lot::Mutex::new(());

impl<E: Embedder, S: MemoryStore> MemoryService<E, S> {
    /// [`ContextCompiler::compile`] with this service's memory folded in:
    /// when the request carries a [`MemoryScope`], relevant memories are
    /// pulled through the fused vector+graph recall and compiled alongside
    /// the caller's fragments, each with its `memory_id` and a normalised
    /// fused-ranking relevance recorded in provenance. Afterwards (policy
    /// permitting) the distinct originals are stored so every
    /// `ctx://source/<hash>` handle round-trips, and a metadata-only
    /// compilation event is recorded for [`Self::context_savings`].
    ///
    /// # Errors
    /// Returns [`MemoryError`] if compilation itself fails (budget, caps),
    /// or if recall, embedding, or storage fails.
    pub fn compile_context(
        &self,
        compiler: &ContextCompiler,
        request: &CompileRequest,
    ) -> Result<CompiledContext, MemoryError> {
        let importance = compiler.effective_policy(request).importance.clone();
        let memories = self.context_memories(request, &importance)?;
        self.compile_with_memories(compiler, request, memories)
    }

    /// [`Self::compile_context`] with a caller-supplied [`crate::Reranker`] driving
    /// memory selection: the reranker receives the FULL fused candidate pool
    /// (vector + graph, before the `k` cutoff) and its ordering decides
    /// which `k` memories are compiled in — the seam for a semantic
    /// cross-encoder or LLM judge a Rust embedder brings along. Not exposed
    /// on the wire (a reranker is code, not JSON), and never a default: the
    /// shipped [`crate::context::DeterministicReranker`] is *lexical*, and a
    /// lexical second stage demotes exactly the zero-vocabulary-overlap
    /// evidence the graph walk rescues (measured in the BDD suite) — bring
    /// a semantic one.
    ///
    /// # Errors
    /// Returns [`MemoryError`] if compilation, recall, the reranker itself,
    /// or storage fails.
    pub fn compile_context_reranked<R: crate::Reranker>(
        &self,
        compiler: &ContextCompiler,
        request: &CompileRequest,
        reranker: &R,
    ) -> Result<CompiledContext, MemoryError> {
        let importance = compiler.effective_policy(request).importance.clone();
        let memories = self.context_memories_reranked(request, reranker, &importance)?;
        self.compile_with_memories(compiler, request, memories)
    }

    /// The shared back half of every compile flavour: augment the request
    /// with the pulled memories, compile, annotate provenance, persist
    /// sources/events per policy.
    fn compile_with_memories(
        &self,
        compiler: &ContextCompiler,
        request: &CompileRequest,
        memories: Vec<PulledMemory>,
    ) -> Result<CompiledContext, MemoryError> {
        let mut augmented = request.clone();
        let mut pulled: BTreeMap<u64, PulledMemory> = BTreeMap::new();
        for memory in memories {
            augmented.fragments.push(memory.fragment.clone());
            pulled.insert(stable_id(&memory.fragment.content), memory);
        }
        // `compile_raw`, not `compile`: annotating memory provenance below
        // can rewrite a pulled fragment's `relevance`/`reason` (and thus
        // whether it crosses the `warnings` threshold), so `decisions` must
        // stay full until that has happened and `warnings` is recomputed —
        // `slim_response` (if requested) is applied as the LAST step.
        let mut out = compiler.compile_raw(&augmented)?;
        annotate_memory_provenance(&mut out, &pulled);
        out.warnings = crate::context::warnings_for(&out.decisions);
        let policy = compiler.effective_policy(request);
        if policy.store_sources {
            self.store_context_sources(&augmented, &out, policy.source_ttl_seconds)?;
        }
        if policy.record_events {
            self.record_context_event(request, &out, policy.event_ttl_seconds)?;
        }
        Ok(crate::context::apply_slim(out, policy))
    }

    /// The memories a request's scope pulls in, as compile fragments plus
    /// their id and normalised fused relevance, importance-blended
    /// ([`Self::blend_importance`]) when the policy's weights are active.
    fn context_memories(
        &self,
        request: &CompileRequest,
        importance: &ImportanceWeights,
    ) -> Result<Vec<PulledMemory>, MemoryError> {
        let Some((scope, k)) = scope_and_k(request) else {
            return Ok(Vec::new());
        };
        let filter = scope_filter(scope);
        // The scope's fusion knobs (clamped by from_knobs); absent ones fall
        // back to the crate defaults — raising graph_boost lets a curated
        // relate-chain out-rank lexically-noisy near-misses (see MemoryScope).
        let opts = FusionOptions::from_knobs(scope.hops, scope.graph_boost, None);
        let scored = self.recall_fused_scored(&request.query, k, filter.as_ref(), opts)?;
        let max_fused = scored
            .iter()
            .map(|s| s.fused)
            .fold(f64::MIN, f64::max)
            .max(f64::EPSILON);
        let candidates = scored
            .into_iter()
            .map(|scored| {
                // Sanitise a non-finite fused score to 0 before normalising:
                // `f32::clamp` returns NaN for a NaN input (it does not clamp),
                // which would put a non-`[0, 1]` value — serialising as JSON
                // `null` — into an output sold as deterministic and auditable.
                let fused = if scored.fused.is_finite() {
                    scored.fused
                } else {
                    0.0
                };
                MemoryCandidate {
                    memory_id: scored.recollection.id,
                    base: (fused / max_fused).clamp(0.0, 1.0),
                    vector_norm: scored.vector_norm,
                    graph_weight: scored.graph_weight,
                    metadata: scored.recollection.metadata,
                    content: scored.recollection.content,
                }
            })
            .collect();
        self.blend_importance(candidates, importance)
    }

    /// Memory selection driven by a caller-supplied reranker: the fused
    /// candidate pool (at pool depth, vector + graph) is handed to the
    /// reranker whole, its ordering is truncated to `k`, and relevance is
    /// rank-based (the reranker defines the ranking; the fused ventilation
    /// no longer describes it, so vector/graph read 0 in provenance). The
    /// importance blend then composes with the seam: it re-ranks INSIDE the
    /// reranker-selected pool, exactly as it does over the fused pool.
    fn context_memories_reranked<R: crate::Reranker>(
        &self,
        request: &CompileRequest,
        reranker: &R,
        importance: &ImportanceWeights,
    ) -> Result<Vec<PulledMemory>, MemoryError> {
        let Some((scope, k)) = scope_and_k(request) else {
            return Ok(Vec::new());
        };
        let filter = scope_filter(scope);
        let opts = FusionOptions::from_knobs(scope.hops, scope.graph_boost, None);
        let ranked =
            self.recall_fused_reranked(&request.query, k, filter.as_ref(), opts, reranker)?;
        let count = ranked.len().max(1);
        let candidates = ranked
            .into_iter()
            .enumerate()
            .map(|(rank, recollection)| {
                // Computed in f32 exactly as 0.8.0 did, so inactive weights
                // reproduce the historical relevance bytes.
                #[allow(clippy::cast_precision_loss)] // rank/count are tiny
                let relevance = 1.0 - (rank as f32 / count as f32);
                MemoryCandidate {
                    memory_id: recollection.id,
                    base: f64::from(relevance),
                    vector_norm: 0.0,
                    graph_weight: 0.0,
                    metadata: recollection.metadata,
                    content: recollection.content,
                }
            })
            .collect();
        self.blend_importance(candidates, importance)
    }

    /// Fold usage-driven importance into an already-selected memory pool —
    /// the one ranking the whole engine stack shares (US-002 of EPIC-P-071):
    /// per candidate the key becomes `base + w_c·(confidence − 0.5)·2 +
    /// w_r·recency_norm`, where `base` is the fused (or rank-based)
    /// similarity in `[0, 1]`. Selection is untouched on purpose: confidence
    /// is not relevance, so a reinforced-but-off-topic fact can never buy
    /// its way into the pool here. Inactive weights take the zero-cost path
    /// and reproduce the 0.8.0 output byte for byte (golden-pinned). The
    /// stable sort keeps equal keys in selection order, and no clock is ever
    /// read — recency is min-max normalised within the batch.
    fn blend_importance(
        &self,
        candidates: Vec<MemoryCandidate>,
        weights: &ImportanceWeights,
    ) -> Result<Vec<PulledMemory>, MemoryError> {
        if !importance_active(weights) {
            return Ok(candidates
                .into_iter()
                .map(MemoryCandidate::into_pulled)
                .collect());
        }
        let ids: Vec<u64> = candidates.iter().map(|c| c.memory_id).collect();
        // Raw payloads (reserved keys included): the learned confidence
        // lives under `_veles_rl_confidence`, which caller-facing metadata
        // strips.
        let raw = self.store.get_metadata_batch(&ids)?;
        let recencies = recency_norms(&candidates, weights);
        let mut blended: Vec<(f64, PulledMemory)> = candidates
            .into_iter()
            .zip(raw)
            .zip(recencies)
            .map(|((candidate, payload), recency)| {
                let confidence = payload_confidence(payload.as_ref());
                let score = candidate.base
                    + weights.confidence * (confidence - NEUTRAL_CONFIDENCE) * 2.0
                    + weights.recency * recency;
                let mut pulled = candidate.into_pulled();
                #[allow(clippy::cast_possible_truncation)] // clamped into [0, 1]
                {
                    pulled.relevance = score.clamp(0.0, 1.0) as f32;
                }
                pulled.confidence = confidence;
                pulled.recency = recency;
                pulled.ventilated = true;
                (score, pulled)
            })
            .collect();
        // Stable: equal blended keys keep the selection order.
        blended.sort_by(|a, b| b.0.total_cmp(&a.0));
        Ok(blended.into_iter().map(|(_, pulled)| pulled).collect())
    }

    /// Store every distinct fragment's original as a hub-marked system fact
    /// keyed by its salted handle hash, so its handle can be resolved later.
    /// A fragment carrying media (US-009, PR2) has its base64 payload
    /// persisted alongside the caption under the reserved
    /// [`CTX_SOURCE_MEDIA_FIELD`] key.
    ///
    /// **Identity**: the key mirrors what the compiler mints handles from
    /// (`Analysis::handle_hash` in `context.rs`) — the caption's
    /// [`stable_id`] for text, the raw decoded bytes' hash
    /// ([`media::MediaAnalysis::raw_hash`]) for media, the same identity
    /// PR1's dedup keys on. Keying media on the caption instead was the PR2
    /// review's proven blocker: every captionless image collided onto one
    /// slot and one handle, serving arbitrary wrong bytes back. The slot
    /// stays inside the salted system-fact namespace ([`source_id`] applies
    /// `SOURCE_ID_SALT` to the hash) — same salt, no new namespace. On a
    /// same-key collision (byte-identical images with different captions)
    /// the FIRST occurrence wins, matching the dedup twin the compiler
    /// keeps — a divergent duplicate caption does not survive, exactly as
    /// its decision reason already says.
    ///
    /// Size: [`crate::limits::MAX_MEDIA_BYTES`] /
    /// [`crate::limits::MAX_TOTAL_MEDIA_BYTES`] already bounded every
    /// fragment's `bytes_b64` before `compiler.compile` ever ran (see
    /// `validate_media`, called from `compile`'s `validate`). TEXT is a
    /// different story, and an earlier revision of this comment got it
    /// wrong by claiming no size guard was needed on the write path: those
    /// media caps say nothing about `content`, which a `path` ingestion can
    /// fill up to 1 MiB — so [`Self::source_vector`] caps what it EMBEDS
    /// (the stored content stays whole). The lesson stands: "another layer
    /// already checked" must name which cap, over which field.
    fn store_context_sources(
        &self,
        augmented: &CompileRequest,
        out: &CompiledContext,
        ttl_seconds: Option<u64>,
    ) -> Result<(), MemoryError> {
        let by_hash = index_fragments_by_handle_hash(&augmented.fragments);
        let ttl_seconds = positive_ttl(ttl_seconds);
        for source in &out.sources {
            self.store_one_source(&source.handle, &by_hash, ttl_seconds)?;
        }
        Ok(())
    }

    /// Write the one slot behind `handle`, if this compile owns it.
    ///
    /// A handle whose fragment is no longer in the request (or that does not
    /// parse) is skipped, not an error: `out.sources` is derived from the
    /// same request, so a miss can only mean the source was externalized
    /// under a shape this write path has nothing to store.
    fn store_one_source(
        &self,
        handle: &str,
        by_hash: &BTreeMap<u64, &ContextFragment>,
        ttl_seconds: Option<u64>,
    ) -> Result<(), MemoryError> {
        let Some(hash) = provenance::parse_handle(handle) else {
            return Ok(());
        };
        let Some(fragment) = by_hash.get(&hash) else {
            return Ok(());
        };
        let slot = source_id(hash);
        if !self.prepare_source_slot(slot, ttl_seconds)? {
            return Ok(());
        }
        let (embedding, media_meta) = self.source_vector(fragment, hash)?;
        let mut extra: Vec<(&str, Value)> = vec![(CTX_SOURCE_FIELD, Value::Bool(true))];
        if let Some(media) = media_meta {
            extra.push((CTX_SOURCE_MEDIA_FIELD, media));
        }
        self.store_fact(
            slot,
            fragment.content.as_str(),
            &embedding,
            Some(&system_meta(&extra)),
            ttl_seconds,
        )
    }

    /// Whether `slot` may be written for this compile, clearing a stale point
    /// first when the write upgrades it to permanent.
    ///
    /// A slot never marked as ours is never rewritten: it is a caller fact
    /// squatting the salt preimage, and clobbering it would destroy user
    /// data. A slot already marked as ours holds these exact bytes — sources
    /// are content-addressed — so content and embedding never change; only
    /// durability can, and only upward (never-downgrade TTL upgrade, see
    /// [`Self::should_store_source`]), so a handle sold as permanent never
    /// silently expires just because an earlier compile first wrote it under
    /// a TTL.
    ///
    /// Upgrading to permanent needs the old point *gone*, not merely
    /// overwritten: velesdb-core's store path preserves every `_veles_*` key
    /// from a prior version of a re-stored id unless the new write explicitly
    /// sets it (`semantic_memory.rs`'s `store_internal` carry-forward, so
    /// plain `remember` doesn't silently wipe learned state), and a permanent
    /// write has no expiry to set (`attach_expiry` is a no-op without one) —
    /// so without this delete, `_veles_expires_at` would survive the
    /// "upgrade" untouched. A TTL-to-TTL extension needs no delete: its new
    /// expiry always overwrites the old one.
    fn prepare_source_slot(
        &self,
        slot: u64,
        ttl_seconds: Option<u64>,
    ) -> Result<bool, MemoryError> {
        if !self.should_store_source(slot, ttl_seconds)? {
            return Ok(false);
        }
        if ttl_seconds.is_none() && self.store.get(slot)?.is_some() {
            self.store.delete(slot)?;
        }
        Ok(true)
    }

    /// The vector a source slot is indexed by, plus the media descriptor to
    /// stamp on it when the fragment carries one.
    ///
    /// A media fragment's vector is deterministic and derived from the
    /// DECODED bytes — never the text embedder over `content` (often blank)
    /// nor over the base64 payload itself (opaque, not language). Correct
    /// because `retrieve_context_source` resolves a media source EXCLUSIVELY
    /// by its content-addressed hash/slot, never by vector search: the vector
    /// only has to be well-formed and non-degenerate for the underlying
    /// index, never semantically meaningful. For a media fragment `hash` IS
    /// the raw-bytes hash (see `fragment_handle_hash`), so nothing is
    /// re-decoded here.
    ///
    /// A TEXT fragment is embedded over at most
    /// [`crate::limits::MAX_EMBEDDABLE_TEXT_BYTES`] of its content
    /// ([`super::embeddable_prefix`]) — a `path`-ingested file can be 1 MiB,
    /// far past what the embedding backend accepts, and handing it over
    /// whole surfaced the backend's raw failure (issue #1654's residue,
    /// found on this very path). Truncating the *embedded* text, not the
    /// stored content, is the right trade here: retrieval is hash-addressed
    /// so the source stays whole, and the vector keeps ranking on the head
    /// of the text instead of vanishing from semantic recall.
    fn source_vector(
        &self,
        fragment: &ContextFragment,
        hash: u64,
    ) -> Result<(Vec<f32>, Option<Value>), MemoryError> {
        let Some(media_ref) = &fragment.media else {
            let embeddable = super::embeddable_prefix(fragment.content.as_str());
            return Ok((self.embedder.embed(embeddable)?, None));
        };
        let descriptor = serde_json::to_value(media_ref).unwrap_or(Value::Null);
        Ok((self.media_placeholder_embedding(hash), Some(descriptor)))
    }

    /// Whether [`Self::store_context_sources`] should (re-)write `slot` for
    /// this compile's requested (already [`positive_ttl`]-normalized —
    /// `None` means permanent) TTL.
    ///
    /// - Not marked as ours (absent, or a caller fact squatting the salt
    ///   preimage): store only if the slot is genuinely empty.
    /// - Marked as ours: never re-embed or change content (content-addressed);
    ///   only [`Self::should_upgrade_ttl`] decides whether durability changes.
    fn should_store_source(
        &self,
        slot: u64,
        requested_ttl: Option<u64>,
    ) -> Result<bool, MemoryError> {
        match self.context_source_metadata(slot)? {
            Some(existing) => Ok(Self::should_upgrade_ttl(&existing, requested_ttl)),
            None => Ok(self.store.get(slot)?.is_none()),
        }
    }

    /// Never-downgrade TTL upgrade rule for an already-stored source: permanent
    /// once requested stays permanent, and a TTL only ever extends, never
    /// shortens. The clock read here is fine — this is the storage/expiry
    /// layer, not the clock-free `compile` pipeline.
    fn should_upgrade_ttl(existing: &Metadata, requested_ttl: Option<u64>) -> bool {
        let existing_expiry = existing.get(EXPIRES_AT_FIELD).and_then(Value::as_u64);
        match (requested_ttl, existing_expiry) {
            // Permanent requested, slot still carries a TTL: upgrade.
            (None, Some(_)) => true,
            // Already permanent, or a TTL requested against a permanent slot:
            // never downgrade.
            (None | Some(_), None) => false,
            // Both carry a TTL: extend only if the new one outlives what
            // remains — never shorten.
            (Some(ttl), Some(existing_exp)) => now_unix_secs().saturating_add(ttl) > existing_exp,
        }
    }

    /// A deterministic, non-degenerate embedding for a media source (US-009,
    /// PR2) — see [`Self::store_context_sources`] for why it is bytes-hash
    /// derived rather than text-embedded.
    fn media_placeholder_embedding(&self, raw_hash: u64) -> Vec<f32> {
        let dim = self.embedder.dimension();
        let mut vector = vec![0.0_f32; dim];
        let Ok(dim_u64) = u64::try_from(dim) else {
            return vector;
        };
        if dim_u64 == 0 {
            return vector;
        }
        let bucket = usize::try_from(raw_hash % dim_u64).unwrap_or(0);
        vector[bucket] = 1.0;
        velesdb_core::simd_native::normalize_inplace_native(&mut vector);
        vector
    }

    /// The fact at `slot`'s metadata, when it carries the stored-source
    /// marker (`None` otherwise — absent, or a caller fact squatting the
    /// slot).
    fn context_source_metadata(&self, slot: u64) -> Result<Option<Metadata>, MemoryError> {
        let payloads = self.store.get_metadata_batch(&[slot])?;
        Ok(payloads
            .into_iter()
            .next()
            .flatten()
            .filter(|meta| meta.get(CTX_SOURCE_FIELD) == Some(&Value::Bool(true))))
    }

    /// The original content — and media, when the fragment carried one —
    /// behind a `ctx://source/<hash>` handle.
    ///
    /// # Errors
    /// Returns [`MemoryError::UnknownHandle`] when the handle is malformed
    /// or nothing is stored under it (never stored, expired, or forgotten).
    pub fn retrieve_context_source(&self, handle: &str) -> Result<ContextSource, MemoryError> {
        let unknown = || MemoryError::UnknownHandle(handle.to_owned());
        let hash = provenance::parse_handle(handle).ok_or_else(unknown)?;
        let slot = source_id(hash);
        // Only marker-bearing facts are sources: a caller fact squatting the
        // salted slot is never served back as compiled provenance.
        let meta = self.context_source_metadata(slot)?.ok_or_else(unknown)?;
        let content = self
            .store
            .get(slot)?
            .map(|(content, _embedding)| content)
            .ok_or_else(unknown)?;
        Ok(ContextSource {
            content,
            media: source_media(&meta),
        })
    }

    /// Explain why one fragment of `request` was preserved, abstracted,
    /// externalized, dropped, or cached — the selection primitive the MCP
    /// `explain_compilation` tool delegates to, extracted here so every
    /// adapter (MCP, Node, Python) shares one implementation instead of
    /// reimplementing it. Compilation is deterministic, so `request` is
    /// simply re-compiled — with event/source recording forced off, since an
    /// explanation must not have side effects — and the matching decision is
    /// returned.
    ///
    /// `fragment_index` (0-based position in `request.fragments`), when
    /// given, TAKES PRIORITY over `fragment_id` for locating the decision:
    /// `compile_context` records exactly one decision per input fragment, in
    /// order, so `decisions[fragment_index]` is unambiguous even when
    /// several fragments are byte-identical and therefore share the same
    /// content-addressed `fragment_id` — a plain `fragment_id` lookup always
    /// resolves to the FIRST such decision (the deduplication survivor's),
    /// never a dropped twin's.
    ///
    /// Caveat inherited from re-compiling rather than replaying stored
    /// state: with a `memory_scope` the re-compile recalls from CURRENT
    /// memory, so the decision reflects memory as it is now, not as it was
    /// at the original `compile_context` call; a caller that already
    /// resolved a `path` fragment to `content` is unaffected (this method
    /// does no I/O of its own).
    ///
    /// # Errors
    /// Returns [`MemoryError::FragmentIndexOutOfBounds`] when `fragment_index`
    /// is beyond `request.fragments`, [`MemoryError::FragmentNotFound`] when
    /// no decision matches the selector, or any error [`Self::compile_context`]
    /// itself can return (budget, caps, recall, embedding, storage).
    pub fn explain_compilation(
        &self,
        request: &CompileRequest,
        fragment_id: u64,
        fragment_index: Option<usize>,
    ) -> Result<ContextDecision, MemoryError> {
        if let Some(index) = fragment_index {
            let len = request.fragments.len();
            if index >= len {
                return Err(MemoryError::FragmentIndexOutOfBounds { index, len });
            }
        }
        let mut request = request.clone();
        let mut policy = request.policy.take().unwrap_or_default();
        // Three options neutralised for one reason: an explanation must not
        // inherit the side effects, nor the presentation, of the compilation it
        // explains. The caller asked "why this fragment?", not "compile this".
        policy.record_events = false;
        policy.store_sources = false;
        // `slim_response` empties `sections` and `decisions` to save tokens
        // (see `apply_slim`). Applied here it would not trim the answer, it
        // would DELETE it: `decisions` is cleared, the lookup below finds
        // nothing, and the caller is told `FragmentNotFound` about a fragment
        // that compiled perfectly well (#1745).
        //
        // The option exists to save tokens, so a caller under a tight budget
        // turns it on by default — and lost the audit tool exactly when they
        // most needed it, with a message that sent them looking for a typo in
        // an id that was correct.
        policy.slim_response = false;
        request.policy = Some(policy);
        let compiled =
            self.compile_context(&ContextCompiler::new(CompilePolicy::default()), &request)?;
        let decision = if let Some(index) = fragment_index {
            compiled.decisions.into_iter().nth(index)
        } else {
            compiled
                .decisions
                .into_iter()
                .find(|decision| decision.fragment_id == fragment_id)
        };
        decision.ok_or(MemoryError::FragmentNotFound(fragment_id))
    }

    /// Record one compilation's savings as a metadata-only system fact
    /// (hashes and token counts — never fragment content). Wall-clock time
    /// is stamped here, outside the deterministic compile pipeline.
    fn record_context_event(
        &self,
        request: &CompileRequest,
        out: &CompiledContext,
        ttl_seconds: Option<u64>,
    ) -> Result<(), MemoryError> {
        let occurred_at_nanos = now_nanos();
        // The per-process sequence keeps ids unique even when two compiles
        // land on the same (possibly coarse) clock tick.
        let seq = EVENT_SEQ.fetch_add(1, Ordering::Relaxed);
        let content = format!("{EVENT_ANCHOR} {occurred_at_nanos}-{seq}");
        let id = stable_id(&format!("{EVENT_ID_SALT}{occurred_at_nanos}:{seq}"));
        let embedding = self.embedder.embed(&content)?;
        let meta = event_meta(request, out, occurred_at_nanos);
        self.store_fact(
            id,
            &content,
            &embedding,
            Some(&meta),
            positive_ttl(ttl_seconds),
        )?;
        Ok(())
    }

    /// Aggregate the recorded compilation events, optionally per project.
    /// Sweeps at most [`crate::limits::MAX_RECALL_LIMIT`] events (newest
    /// need not be first — the sweep is similarity-ordered over a constant
    /// anchor, i.e. effectively the whole family until the cap);
    /// [`ContextSavings::truncated`] reports when the cap was hit.
    ///
    /// # Errors
    /// Returns [`MemoryError`] if the underlying filtered recall fails.
    pub fn context_savings(&self, project: Option<&str>) -> Result<ContextSavings, MemoryError> {
        // Filter at the STORAGE layer on the reserved event marker: callers
        // can neither set nor query `_veles_*` keys, so only genuine bridge
        // events can ever match — a caller fact posing as an event counts
        // for nothing.
        let mut filter = Map::new();
        filter.insert(CTX_EVENT_FIELD.to_owned(), Value::Bool(true));
        if let Some(project) = project {
            filter.insert(
                CTX_PROJECT_FIELD.to_owned(),
                Value::String(project.to_owned()),
            );
        }
        let embedding = self.embedder.embed(EVENT_ANCHOR)?;
        let hits =
            self.store
                .query_filtered(&embedding, crate::limits::MAX_RECALL_LIMIT, &filter, 0)?;
        let ids: Vec<u64> = hits.iter().map(|(id, _, _)| *id).collect();
        let payloads = self.store.get_metadata_batch(&ids)?;
        Ok(aggregate_events(&payloads))
    }

    /// Persist `working` under `project` + `session` (idempotent upsert:
    /// saving again replaces the previous state). Returns the system fact id.
    ///
    /// Serialized size is capped at [`crate::limits::MAX_FACT_BYTES`] (1
    /// MiB) — the same ceiling every other stored fact honors — checked
    /// BEFORE anything is written, so an oversized working context is never
    /// partially stored.
    ///
    /// An entirely empty `working` ([`WorkingContext::is_empty`]) is refused.
    /// Because the write is an upsert, saving one would replace — destroy —
    /// the state a previous save stored under the same project and session,
    /// and the one tool whose job is surviving a context loss must not be
    /// able to cause one on a call that carries nothing (issue #1654).
    ///
    /// # Errors
    /// Returns [`MemoryError::EmptyWorkingContext`] if `working` records
    /// nothing, [`MemoryError::WorkingContextCodec`] if serialization fails,
    /// [`MemoryError::ContextOverLimit`] if the serialized `working` exceeds
    /// [`crate::limits::MAX_FACT_BYTES`], or a storage/embedding error.
    pub fn save_working_context(
        &self,
        project: &str,
        session: &str,
        working: &WorkingContext,
    ) -> Result<u64, MemoryError> {
        if working.is_empty() {
            return Err(MemoryError::EmptyWorkingContext);
        }
        let content = serde_json::to_string(working)
            .map_err(|err| MemoryError::WorkingContextCodec(err.to_string()))?;
        if content.len() > crate::limits::MAX_FACT_BYTES {
            return Err(MemoryError::ContextOverLimit(format!(
                "working context of {} bytes exceeds the cap of {} bytes",
                content.len(),
                crate::limits::MAX_FACT_BYTES
            )));
        }
        let id = working_id(project, session);
        let embedding = self
            .embedder
            .embed(&format!("working context {project} {session}"))?;
        let meta = system_meta(&[
            (CTX_WORKING_FIELD, Value::Bool(true)),
            (CTX_PROJECT_FIELD, Value::String(project.to_owned())),
            (CTX_SESSION_FIELD, Value::String(session.to_owned())),
        ]);
        self.store_fact(id, &content, &embedding, Some(&meta), None)?;
        self.update_working_index(project, session)?;
        Ok(id)
    }

    /// The working context previously saved under `project` + `session`,
    /// `None` when there is none.
    ///
    /// Symmetric to [`Self::context_source_metadata`]'s squatter guard: the
    /// slot is only ever served back when its metadata carries the reserved
    /// [`CTX_WORKING_FIELD`] marker (set exclusively by
    /// [`Self::save_working_context`]). A slot occupied by an unmarked caller
    /// fact — one that happened to land on this salted id, or a forged
    /// probe — is indistinguishable from "nothing saved" on purpose: `None`,
    /// never the forged content, and never an error (the caller cannot tell
    /// a squatted slot from a genuinely empty one, which is the point — it
    /// must never learn that *something* occupies this id).
    ///
    /// A pure read: it never writes, never prunes, never heals. Index
    /// convergence happens on the WRITE path
    /// ([`Self::update_working_index`]) — a lookup that rewrites shared state
    /// turns every transient miss into permanent data loss and cannot safely
    /// be retried.
    ///
    /// # Errors
    /// Returns [`MemoryError::WorkingContextCodec`] if the stored payload
    /// does not parse, or if the slot is marked but its body is gone (a torn
    /// fact is corruption — reporting it as "nothing saved" would tell the
    /// caller the one thing that is certainly false), or a storage error.
    pub fn load_working_context(
        &self,
        project: &str,
        session: &str,
    ) -> Result<Option<WorkingContext>, MemoryError> {
        let slot = working_id(project, session);
        let payloads = self.store.get_metadata_batch(&[slot])?;
        let marked = payloads
            .into_iter()
            .next()
            .flatten()
            .is_some_and(|meta| meta.get(CTX_WORKING_FIELD) == Some(&Value::Bool(true)));
        if !marked {
            // The squatter/never-saved guard documented above: silent by
            // design, and the branch a `forget` lands on (deleting a fact
            // removes its metadata with it).
            return Ok(None);
        }
        let Some((content, _)) = self.store.get(slot)? else {
            return Err(MemoryError::WorkingContextCodec(format!(
                "working context for project '{project}', session '{session}' is corrupt: \
                 the reserved marker is present but the stored body is gone"
            )));
        };
        serde_json::from_str(&content)
            .map(Some)
            .map_err(|err| MemoryError::WorkingContextCodec(err.to_string()))
    }

    /// The full resumption envelope for `project` + `session`: what
    /// [`Self::load_working_context`] found, plus the OTHER sessions saved
    /// under the same project so a typo in `session` is recoverable.
    ///
    /// This is the ONE place the three policy rules live:
    ///
    /// 1. `other_sessions` is listed on a HIT too, not just on a miss — a
    ///    typo that lands on another REAL session returns `found: true`, and
    ///    the caller has no other way to notice it resumed the wrong work.
    ///    Costs one extra O(1) index read per successful load.
    /// 2. The requested `session` is never echoed back: the field is named
    ///    `other_sessions`, so returning the requested id would be a
    ///    contradiction the caller cannot act on.
    /// 3. An unreadable index is fatal on a MISS and survivable on a HIT —
    ///    see [`Self::other_sessions_for`].
    ///
    /// Every surface (the `load_working_context` MCP tool and the Node,
    /// Python and WASM bindings) calls this rather than recomposing the
    /// envelope from [`Self::load_working_context`] +
    /// [`Self::list_working_contexts`]: four recompositions are four copies
    /// of those rules, and a copy that stops matching the others fails
    /// silently — the caller still gets a well-formed envelope, just a
    /// different one.
    ///
    /// # Errors
    /// Propagates [`Self::load_working_context`]'s errors (a corrupt or
    /// unparseable stored payload), and [`Self::list_working_contexts`]'s (a
    /// corrupt index, or a storage failure) **on a miss only** — rule 3.
    pub fn resume_working_context(
        &self,
        project: &str,
        session: &str,
    ) -> Result<LoadedWorkingContext, MemoryError> {
        let working = self.load_working_context(project, session)?;
        let other_sessions = self.other_sessions_for(project, session, working.is_some())?;
        Ok(LoadedWorkingContext {
            found: working.is_some(),
            working,
            other_sessions,
        })
    }

    /// The project's OTHER sessions, and what to do when the index that holds
    /// them cannot be read.
    ///
    /// The two answers differ because `other_sessions` plays a different part
    /// on each path:
    ///
    /// - **On a hit** it is a HINT — "you asked for `alpha`, note that
    ///   `alpha-2` also exists, you may have resumed the wrong one". The
    ///   answer the caller actually asked for is already in hand and intact.
    ///   Failing the whole call here would turn a fault in one auxiliary fact
    ///   into a total loss of resumption for EVERY session of the project,
    ///   including the many that read back perfectly — which is why an
    ///   unreadable index degrades to an empty hint instead. Nothing is
    ///   swallowed: the corruption stays loudly reachable through
    ///   [`Self::list_working_contexts`], published on every surface.
    /// - **On a miss** it is the ONLY signal there is. `[]` then reads as the
    ///   positive assertion "nothing else was ever saved under this project",
    ///   and an agent told that starts over on top of work sitting right next
    ///   to where it looked — the exact failure this envelope exists to
    ///   prevent. An assertion we cannot support must not be manufactured, so
    ///   the error propagates.
    ///
    /// # Errors
    /// Propagates [`Self::list_working_contexts`]'s errors when `found` is
    /// false.
    fn other_sessions_for(
        &self,
        project: &str,
        session: &str,
        found: bool,
    ) -> Result<Vec<String>, MemoryError> {
        let listed = match self.list_working_contexts(project) {
            Ok(listed) => listed,
            Err(_) if found => return Ok(Vec::new()),
            Err(err) => return Err(err),
        };
        Ok(listed
            .into_iter()
            .map(|entry| entry.session)
            .filter(|candidate| candidate != session)
            .collect())
    }

    /// The sessions of `sessions` whose working-context fact is still there,
    /// in the same order. One batched metadata lookup for the whole set — not
    /// a store scan, but not free either (see
    /// [`Self::list_working_contexts`]'s cost note).
    ///
    /// Shared by the read path (filter, persist nothing) and the write path
    /// (filter, and persist the result), so both agree on what "alive" means.
    fn live_sessions(
        &self,
        project: &str,
        sessions: Vec<WorkingContextSession>,
    ) -> Result<Vec<WorkingContextSession>, MemoryError> {
        if sessions.is_empty() {
            return Ok(sessions);
        }
        let ids: Vec<u64> = sessions
            .iter()
            .map(|entry| working_id(project, &entry.session))
            .collect();
        let payloads = self.store.get_metadata_batch(&ids)?;
        if payloads.len() != ids.len() {
            // The trait promises one result per id. A backend that breaks
            // that promise must not be silently read as "these sessions are
            // dead" — that would delete real entries on the write path.
            return Err(MemoryError::WorkingContextCodec(format!(
                "storage returned {} metadata rows for {} working-context ids",
                payloads.len(),
                ids.len()
            )));
        }
        Ok(sessions
            .into_iter()
            .zip(payloads)
            .filter(|(_, meta)| {
                meta.as_ref()
                    .is_some_and(|meta| meta.get(CTX_WORKING_FIELD) == Some(&Value::Bool(true)))
            })
            .map(|(entry, _)| entry)
            .collect())
    }

    /// Every session still resumable under `project`'s working-context index
    /// (V2a-1 quick win), most-recently-saved first. Empty when the project
    /// never saved anything — that, and only that, is the empty case.
    ///
    /// Cost: one O(1) index read plus ONE batched metadata lookup of the
    /// listed ids — never a store scan, but no longer a single read either.
    /// The lookup is what drops sessions whose fact was forgotten since;
    /// unlike the previous read-path prune it persists nothing, so a listing
    /// can be retried and a transient miss costs nothing durable.
    ///
    /// # Errors
    /// Returns a storage error if the index fact cannot be read, or
    /// [`MemoryError::WorkingContextCodec`] if it does not parse or is
    /// corrupt (marked, but with no body).
    pub fn list_working_contexts(
        &self,
        project: &str,
    ) -> Result<Vec<WorkingContextSession>, MemoryError> {
        let Some(index) = self.working_index(project)? else {
            // The genuine "this project never saved anything" case — the only
            // one that reaches here now that a corrupt index is an `Err`.
            return Ok(Vec::new());
        };
        let mut sessions = self.live_sessions(project, index.sessions)?;
        sessions.sort_by(|a, b| {
            b.saved_at
                .cmp(&a.saved_at)
                .then_with(|| a.session.cmp(&b.session))
        });
        Ok(sessions)
    }

    /// The raw working-context index fact for `project`, `None` when nothing
    /// was ever saved under it. Symmetric squatter guard to
    /// [`Self::load_working_context`]: a slot occupied without the reserved
    /// [`CTX_WORKING_INDEX_FIELD`] marker is treated as empty, never as a
    /// forged index.
    ///
    /// `None` means "absent". "Corrupt" is an `Err` — collapsing the two
    /// would report a store that lost the index body as a project that never
    /// saved anything, and an agent told that starts over instead of raising
    /// a problem a human could fix.
    fn working_index(&self, project: &str) -> Result<Option<WorkingContextIndex>, MemoryError> {
        let slot = working_index_id(project);
        let payloads = self.store.get_metadata_batch(&[slot])?;
        let marked = payloads
            .into_iter()
            .next()
            .flatten()
            .is_some_and(|meta| meta.get(CTX_WORKING_INDEX_FIELD) == Some(&Value::Bool(true)));
        if !marked {
            return Ok(None);
        }
        match self.store.get(slot)? {
            Some((content, _)) => serde_json::from_str(&content)
                .map(Some)
                .map_err(|err| MemoryError::WorkingContextCodec(err.to_string())),
            None => Err(MemoryError::WorkingContextCodec(format!(
                "working-context index for project '{project}' is corrupt: the index \
                 marker is present but the stored body is gone"
            ))),
        }
    }

    /// Append (or refresh) `session`'s entry in `project`'s working-context
    /// index — called by every [`Self::save_working_context`], so the index
    /// is always current without a separate maintenance step. A resave of
    /// the same project+session updates `saved_at` in place rather than
    /// duplicating the entry.
    ///
    /// This is also where the index CONVERGES: entries whose working-context
    /// fact was forgotten since are dropped here, on the write path, under
    /// the same lock and in the same read-modify-write that was already
    /// paid for. Reads never mutate it.
    fn update_working_index(&self, project: &str, session: &str) -> Result<(), MemoryError> {
        // The index slot's embedding derives from the PROJECT NAME alone,
        // never from the index content, so it is computed here, BEFORE the
        // lock: an embedder can be a network round-trip (or a hung one), and
        // holding the global write lock across it stalls every working-index
        // write in the process behind one slow call. Racing saves may embed
        // concurrently, but they embed the same text, so whichever vector
        // lands is equivalent — no re-check under the lock is needed. The
        // index CONTENT read-modify-write stays entirely under the lock.
        let embedding = self
            .embedder
            .embed(&format!("working context index {project}"))?;
        // Read-modify-write of a single shared fact: held for the whole
        // sequence, otherwise a concurrent save silently erases this entry.
        let _guard = WORKING_INDEX_WRITE.lock();
        // A corrupt index must not brick saving for the whole project. The
        // read path surfaces the error — that is where a human can act on it
        // — but propagating it here would make every future save of every
        // session under this project fail forever, with no way back: the
        // only writer of the index is this function. Rebuild instead.
        let mut index = match self.working_index(project) {
            Ok(index) => index.unwrap_or_default(),
            Err(MemoryError::WorkingContextCodec(_)) => WorkingContextIndex::default(),
            Err(err) => return Err(err),
        };
        let now = now_unix_secs();
        if let Some(entry) = index.sessions.iter_mut().find(|s| s.session == session) {
            entry.saved_at = now;
        } else {
            index.sessions.push(WorkingContextSession {
                session: session.to_owned(),
                saved_at: now,
            });
        }
        // The entry just appended is alive by construction (its fact was
        // stored moments ago, before this call); this only sheds the ones a
        // `forget` orphaned.
        index.sessions = self.live_sessions(project, index.sessions)?;
        let content = serde_json::to_string(&index)
            .map_err(|err| MemoryError::WorkingContextCodec(err.to_string()))?;
        self.write_working_index(project, &content, &embedding)
    }

    /// Persist a serialized index into `project`'s reserved index slot —
    /// always with the [`CTX_WORKING_INDEX_FIELD`] marker, since an index
    /// written without it would be treated as a squatter and read back as
    /// empty. Only [`Self::update_working_index`] (which holds
    /// [`WORKING_INDEX_WRITE`] and supplies the slot `embedding` it computed
    /// before taking that lock) calls this: nothing in here may call the
    /// embedder, or the lock would again be held across a network hop.
    fn write_working_index(
        &self,
        project: &str,
        content: &str,
        embedding: &[f32],
    ) -> Result<(), MemoryError> {
        let slot = working_index_id(project);
        let meta = system_meta(&[
            (CTX_WORKING_INDEX_FIELD, Value::Bool(true)),
            (CTX_PROJECT_FIELD, Value::String(project.to_owned())),
        ]);
        self.store_fact(slot, content, embedding, Some(&meta), None)?;
        Ok(())
    }
}

/// How many memories a scope pulls when it does not say (`k` absent).
const DEFAULT_MEMORY_K: usize = 5;

/// The request's memory scope plus the clamped pull count — `None` when
/// there is no scope or no room: pulled memories must never push the
/// request over the fragment cap (the cap is validated after augmentation,
/// and a rejection there would blame the caller for fragments the bridge
/// itself added).
fn scope_and_k(request: &CompileRequest) -> Option<(&MemoryScope, usize)> {
    let scope = request.memory_scope.as_ref()?;
    let room = crate::limits::MAX_FRAGMENTS.saturating_sub(request.fragments.len());
    let k = crate::limits::clamp_recall_limit(scope.k.unwrap_or(DEFAULT_MEMORY_K)).min(room);
    (k > 0).then_some((scope, k))
}

/// The recall filter a scope narrows to (its project facet), if any.
fn scope_filter(scope: &MemoryScope) -> Option<Metadata> {
    scope.project.as_ref().map(|project| {
        let mut meta = Map::new();
        meta.insert("project".to_owned(), Value::String(project.clone()));
        meta
    })
}

/// One memory the scope pulled in, with its full ranking ventilation.
struct PulledMemory {
    fragment: ContextFragment,
    memory_id: u64,
    /// Fused score normalised over the pulled batch, in `[0, 1]` — the
    /// importance-blended key (clamped) when the blend is active.
    relevance: f32,
    /// Normalised vector term of the fused score.
    vector_norm: f64,
    /// Graph promotion weight of the fused score.
    graph_weight: f64,
    /// Learned RL confidence the blend used (neutral `0.5` when the memory
    /// never received feedback).
    confidence: f64,
    /// Batch-relative recency contribution in `[0, 1]` (`0` when the term
    /// is inactive, the key is absent, or the batch is degenerate).
    recency: f64,
    /// Whether the importance blend ran — drives the extended four-signal
    /// reason ventilation; `false` keeps the exact 0.8.0 reason bytes.
    ventilated: bool,
}

/// A selected memory before the importance blend: its similarity base, its
/// fused ventilation, and the caller-visible metadata the recency term reads.
struct MemoryCandidate {
    memory_id: u64,
    /// Fused-normalised (or rank-based) similarity in `[0, 1]`.
    base: f64,
    vector_norm: f64,
    graph_weight: f64,
    metadata: Option<Metadata>,
    content: String,
}

impl MemoryCandidate {
    /// The unblended [`PulledMemory`] — bytes identical to the 0.8.0 pull.
    fn into_pulled(self) -> PulledMemory {
        #[allow(clippy::cast_possible_truncation)] // base is clamped into [0, 1]
        let relevance = self.base as f32;
        PulledMemory {
            fragment: ContextFragment {
                id: None,
                content: self.content,
                path: None,
                kind: Some("memory".to_owned()),
                priority: None,
                metadata: None,
                media: None,
            },
            memory_id: self.memory_id,
            relevance,
            vector_norm: self.vector_norm,
            graph_weight: self.graph_weight,
            confidence: NEUTRAL_CONFIDENCE,
            recency: 0.0,
            ventilated: false,
        }
    }
}

/// The neutral confidence of a memory with no feedback history — mirrors
/// `reinforce::RL_NEUTRAL_CONFIDENCE`, whose module is `persistence`-gated:
/// its contribution to the blend is exactly `0`.
const NEUTRAL_CONFIDENCE: f64 = 0.5;

/// The learned RL confidence off a raw payload, in `[0, 1]`. Without the
/// `persistence` feature the RL module (and thus `feedback`) does not exist,
/// so every memory reads neutral.
#[cfg(feature = "persistence")]
fn payload_confidence(payload: Option<&Metadata>) -> f64 {
    f64::from(payload.map_or(
        super::reinforce::RL_NEUTRAL_CONFIDENCE,
        super::reinforce::read_confidence,
    ))
}

/// See the `persistence` twin: no RL module, always neutral.
#[cfg(not(feature = "persistence"))]
fn payload_confidence(_payload: Option<&Metadata>) -> f64 {
    NEUTRAL_CONFIDENCE
}

/// Whether the policy's importance weights change anything at all: a
/// non-zero confidence weight, or a non-zero recency weight WITH a field to
/// read. Zero weights must cost nothing and change nothing (0.8.0 parity).
#[allow(
    clippy::float_cmp,
    reason = "an exact zero weight is the documented off switch; any non-zero weight, however small, is active"
)]
fn importance_active(weights: &ImportanceWeights) -> bool {
    weights.confidence != 0.0 || (weights.recency != 0.0 && weights.recency_field.is_some())
}

/// The batch-relative recency contribution of every candidate, in `[0, 1]`:
/// min-max over the candidates that carry the policy's `recency_field` as a
/// number (one monotone scale per batch — `YYYYMMDD` or an epoch, the
/// caller's choice). A candidate without the key contributes `0` (never
/// penalised), and a degenerate batch (`max == min`) contributes `0` for
/// all. No clock: recency is relative to the newest of the batch.
#[allow(
    clippy::float_cmp,
    reason = "an exact zero weight is the documented off switch for the recency term"
)]
fn recency_norms(candidates: &[MemoryCandidate], weights: &ImportanceWeights) -> Vec<f64> {
    let field = weights
        .recency_field
        .as_ref()
        .filter(|_| weights.recency != 0.0);
    let Some(field) = field else {
        return vec![0.0; candidates.len()];
    };
    let values: Vec<Option<f64>> = candidates
        .iter()
        .map(|candidate| {
            candidate
                .metadata
                .as_ref()
                .and_then(|meta| meta.get(field.as_str()))
                .and_then(Value::as_f64)
                .filter(|value| value.is_finite())
        })
        .collect();
    let (min, max) = values
        .iter()
        .flatten()
        .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| {
            (lo.min(v), hi.max(v))
        });
    if max <= min {
        return vec![0.0; candidates.len()];
    }
    values
        .into_iter()
        .map(|value| value.map_or(0.0, |v| ((v - min) / (max - min)).clamp(0.0, 1.0)))
        .collect()
}

/// Stamp pulled memories into the compiled provenance: their decisions and
/// sources gain the backing `memory_id`, the decision's relevance becomes
/// the normalised (importance-blended, when active) ranking score, and the
/// reason spells out the full score ventilation — vector and graph always,
/// plus confidence and recency when the blend ran — so `why this memory` is
/// answerable from the decision alone.
fn annotate_memory_provenance(out: &mut CompiledContext, pulled: &BTreeMap<u64, PulledMemory>) {
    for decision in &mut out.decisions {
        if let Some(memory) = pulled.get(&decision.content_hash) {
            decision.memory_id = Some(memory.memory_id);
            decision.relevance = memory.relevance;
            decision.reason = if memory.ventilated {
                format!(
                    "{} — pulled from memory {} (vector {:.2}, graph {:.2}, confidence {:.2}, recency {:.2})",
                    decision.reason,
                    memory.memory_id,
                    memory.vector_norm,
                    memory.graph_weight,
                    memory.confidence,
                    memory.recency
                )
            } else {
                format!(
                    "{} — pulled from memory {} (vector {:.2}, graph {:.2})",
                    decision.reason, memory.memory_id, memory.vector_norm, memory.graph_weight
                )
            };
        }
    }
    for source in &mut out.sources {
        if let Some(hash) = provenance::parse_handle(&source.handle) {
            if let Some(memory) = pulled.get(&hash) {
                source.memory_id = Some(memory.memory_id);
            }
        }
    }
}

/// Base metadata of every bridge-stored system fact: hub-marked (invisible
/// to normal recall) plus the given extra keys.
fn system_meta(extra: &[(&str, Value)]) -> Metadata {
    let mut meta = Map::new();
    meta.insert(HUB_FIELD.to_owned(), Value::Bool(true));
    for (key, value) in extra {
        meta.insert((*key).to_owned(), value.clone());
    }
    meta
}

/// The metadata of one compilation event — counts and identifiers only,
/// every key reserved.
fn event_meta(request: &CompileRequest, out: &CompiledContext, nanos: u128) -> Metadata {
    let mut extra: Vec<(&str, Value)> = vec![
        (CTX_EVENT_FIELD, Value::Bool(true)),
        (
            CTX_TOKENS_IN_FIELD,
            Value::Number(out.insights.tokens_in.into()),
        ),
        (
            CTX_TOKENS_OUT_FIELD,
            Value::Number(out.insights.tokens_out.into()),
        ),
        (
            CTX_TOKENS_SAVED_FIELD,
            Value::Number(out.insights.tokens_saved.into()),
        ),
        (
            CTX_AT_FIELD,
            Value::Number(Number::from(
                u64::try_from(nanos / 1_000_000_000).unwrap_or(u64::MAX),
            )),
        ),
    ];
    if let Some(project) = &request.project {
        extra.push((CTX_PROJECT_FIELD, Value::String(project.clone())));
    }
    if let Some(model) = &request.target_model {
        extra.push((CTX_MODEL_FIELD, Value::String(model.clone())));
    }
    if let (Some(micros), Some(currency)) = (
        out.insights.estimated_cost_saved_micros,
        out.insights.currency.as_ref(),
    ) {
        extra.push((CTX_COST_FIELD, Value::Number(micros.into())));
        extra.push((CTX_CURRENCY_FIELD, Value::String(currency.clone())));
    }
    system_meta(&extra)
}

/// Fold raw event payloads (reserved keys included) into one
/// [`ContextSavings`]. Every accumulation saturates — an aggregate must
/// never panic, whatever the stored numbers.
fn aggregate_events(payloads: &[Option<Metadata>]) -> ContextSavings {
    let mut savings = ContextSavings {
        events: payloads.len() as u64,
        truncated: payloads.len() >= crate::limits::MAX_RECALL_LIMIT,
        ..ContextSavings::default()
    };
    for payload in payloads {
        let Some(meta) = payload else { continue };
        savings.tokens_in = savings
            .tokens_in
            .saturating_add(meta_u64(meta, CTX_TOKENS_IN_FIELD));
        savings.tokens_out = savings
            .tokens_out
            .saturating_add(meta_u64(meta, CTX_TOKENS_OUT_FIELD));
        savings.tokens_saved = savings
            .tokens_saved
            .saturating_add(meta_u64(meta, CTX_TOKENS_SAVED_FIELD));
        if let (Some(Value::String(currency)), micros) =
            (meta.get(CTX_CURRENCY_FIELD), meta_u64(meta, CTX_COST_FIELD))
        {
            if micros > 0 {
                let entry = savings
                    .cost_saved_micros_by_currency
                    .entry(currency.clone())
                    .or_insert(0);
                *entry = entry.saturating_add(micros);
            }
        }
    }
    savings
}

/// A `u64` metadata field, `0` when absent or non-numeric.
fn meta_u64(meta: &Metadata, key: &str) -> u64 {
    meta.get(key).and_then(Value::as_u64).unwrap_or(0)
}

/// The salted system-fact id of a stored source.
fn source_id(content_hash: u64) -> u64 {
    stable_id(&format!("{SOURCE_ID_SALT}{content_hash}"))
}

/// The handle-identity hash of one request fragment — the bridge-side twin
/// of `Analysis::handle_hash` in `context.rs` (kept in lockstep; the two
/// must key the same identity or stored slots and minted handles drift
/// apart): raw decoded media bytes for a media fragment, caption/content
/// [`stable_id`] otherwise.
fn fragment_handle_hash(fragment: &ContextFragment) -> u64 {
    fragment.media.as_ref().map_or_else(
        || stable_id(&fragment.content),
        |media_ref| media::analyze(media_ref).raw_hash,
    )
}

/// Index a request's fragments by the hash their `ctx://source/` handle is
/// built from, so a handle can be resolved back to the fragment that produced
/// it. First occurrence wins (see the identity note on
/// `store_context_sources`): `entry` + `or_insert`, never a blind overwrite.
fn index_fragments_by_handle_hash(
    fragments: &[ContextFragment],
) -> BTreeMap<u64, &ContextFragment> {
    let mut by_hash: BTreeMap<u64, &ContextFragment> = BTreeMap::new();
    for fragment in fragments {
        by_hash
            .entry(fragment_handle_hash(fragment))
            .or_insert(fragment);
    }
    by_hash
}

/// A stored source's media payload (US-009, PR2), when its metadata carries
/// one — absent (or malformed, which should never happen for a payload this
/// bridge wrote itself) round-trips as `None` rather than an error, so a
/// media decode hiccup degrades to "text-only", never breaks the whole
/// retrieval.
fn source_media(meta: &Metadata) -> Option<MediaRef> {
    meta.get(CTX_SOURCE_MEDIA_FIELD)
        .cloned()
        .and_then(|value| serde_json::from_value(value).ok())
}

/// The salted, deterministic system-fact id of a working context.
fn working_id(project: &str, session: &str) -> u64 {
    stable_id(&format!("{WORKING_ID_SALT}{project}\u{1f}{session}"))
}

/// The salted, deterministic system-fact id of a project's working-context
/// index — one per project, so every save updates the same slot.
fn working_index_id(project: &str) -> u64 {
    stable_id(&format!("{WORKING_INDEX_ID_SALT}{project}"))
}

#[cfg(all(test, feature = "persistence"))]
#[path = "memory_bridge_tests.rs"]
mod tests;