topodb 0.1.0

//! Production hybrid recall: reciprocal-rank fusion of the text, vector,
//! and graph read paths. The engine owns the MECHANICS only — query
//! vectors and term expansions arrive pre-resolved from the host (see the
//! spec: graph-native data, engine mechanics, host policy).

use crate::ids::NodeId;
use crate::read::TimeAxis;

/// Standard RRF constant — dampens the head so one leg's #1 can't drown
/// out consistent mid-rank agreement across legs.
pub(crate) const RRF_K: f32 = 60.0;

/// Fuses per-leg rankings: each list is `(weight, ids best-first)`; a
/// node's fused score is `Σ weight / (RRF_K + rank)` over the lists it
/// appears in (rank is 1-based). Output is sorted score-desc with
/// ascending-id tie-break — the same determinism contract as search.
pub(crate) fn rrf_fuse(lists: &[(f32, Vec<NodeId>)]) -> Vec<(NodeId, f32)> {
    use std::collections::HashMap;
    let mut scores: HashMap<NodeId, f32> = HashMap::new();
    for (weight, ids) in lists {
        for (i, id) in ids.iter().enumerate() {
            *scores.entry(*id).or_insert(0.0) += weight / (RRF_K + (i as f32 + 1.0));
        }
    }
    let mut out: Vec<(NodeId, f32)> = scores.into_iter().collect();
    out.sort_by(|a, b| {
        b.1.partial_cmp(&a.1)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then_with(|| a.0.cmp(&b.0))
    });
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    fn id(n: u128) -> NodeId {
        NodeId::from_u128(n)
    }

    #[test]
    fn agreement_across_legs_beats_single_leg_top_rank() {
        // B is #2 in both legs; A and C are #1 in one leg each.
        // 2/(60+2) = 0.0323 > 1/(60+1) = 0.0164 — agreement wins.
        let fused = rrf_fuse(&[
            (1.0, vec![id(1), id(2)]), // text: A, B
            (1.0, vec![id(3), id(2)]), // vector: C, B
        ]);
        assert_eq!(fused[0].0, id(2), "B (agreement) must rank first");
    }

    #[test]
    fn weights_scale_leg_contributions() {
        let fused = rrf_fuse(&[(1.0, vec![id(1)]), (0.5, vec![id(2)])]);
        assert_eq!(fused[0].0, id(1));
        assert!((fused[0].1 - 1.0 / 61.0).abs() < 1e-6);
        assert!((fused[1].1 - 0.5 / 61.0).abs() < 1e-6);
    }

    #[test]
    fn ties_break_by_ascending_id() {
        let fused = rrf_fuse(&[(1.0, vec![id(9)]), (1.0, vec![id(3)])]);
        assert_eq!(fused[0].0, id(3), "equal scores: lower id first");
    }

    #[test]
    fn empty_input_fuses_to_empty() {
        assert!(rrf_fuse(&[]).is_empty());
        assert!(rrf_fuse(&[(1.0, vec![])]).is_empty());
    }
}

use crate::error::TopoError;
use crate::fts::SearchOptions;
use crate::ids::ScopeSet;
use crate::props::PropValue;
use crate::state::NodeRecord;
use crate::Db;

/// How deep each leg ranks before fusion: enough depth that RRF has real
/// lists to agree over even for small `k`, capped so a huge `k` cannot
/// turn every leg into a full-corpus scan.
pub(crate) fn leg_depth(k: usize) -> usize {
    (3 * k).clamp(30, 50)
}

/// One production hybrid-recall request. The engine fuses; the HOST
/// resolves — `vector` is a pre-computed query embedding, `expansions`
/// are pre-resolved synonym terms. The engine neither knows nor cares
/// where either came from (spec: engine mechanics, host policy).
#[derive(Debug, Clone)]
pub struct RecallQuery {
    pub scopes: ScopeSet,
    pub query: String,
    pub k: usize,
    /// Host-computed query embedding with its model name. `None` = no
    /// vector leg. An empty vector is `Rejected` (a host bug should be
    /// loud); an unknown model name is just an empty leg (legitimately no
    /// data under that namespace).
    pub vector: Option<(String, Vec<f32>)>,
    /// Host-resolved term expansions, applied to the text leg at
    /// `fts::FUZZY_DISCOUNT`. Depth-1 only — the engine never chains them.
    pub expansions: Vec<(String, Vec<String>)>,
    /// Two-stage graph signal: 1-hop neighbors of the top preliminary
    /// seeds join as a third, half-weight list.
    pub graph_boost: bool,
    /// Recency + fuzzy knobs. Recency is applied ONCE, post-fusion (the
    /// legs run recency-free so the decay can't compound). `options.now_ms`
    /// pins BOTH clocks a call needs to be deterministic: the post-fusion
    /// recency decay's "now" (see above), AND the graph leg's `as_of`
    /// traversal time (`TraversalQuery::as_of`, passed through verbatim) — a
    /// test or replay that fixes `now_ms` gets a single consistent instant
    /// for every time-sensitive part of one `recall` call, not two that
    /// could drift apart.
    pub options: SearchOptions,
    /// Post-fusion label allowlist: when `Some`, only nodes carrying one of
    /// these labels survive fusion. `None` = unfiltered (today's behavior).
    /// `Some(vec![])` is `Rejected` — an empty allowlist admits nothing, so
    /// it is almost certainly a caller bug; omit the field to search
    /// unfiltered instead.
    pub labels: Option<Vec<String>>,
    /// Tombstone filter: a fused candidate is dropped if ANY of these props
    /// holds an integer value `<=` the query's effective `now`
    /// (`options.now_ms`, else wall clock). Used to exclude memories a caller
    /// has marked superseded/forgotten, while an `as_of`/`now_ms` set BEFORE
    /// the mark still sees them — the marker is a timestamp, not a delete.
    /// Empty (the default) = no tombstone filtering. Generic on purpose: the
    /// engine never names the props.
    pub tombstone_props: Vec<String>,
    /// Text leg's RRF weight. Defaults to `WEIGHT_TEXT`.
    pub text_weight: f32,
    /// Vector leg's RRF weight. Defaults to `WEIGHT_VECTOR`.
    pub vector_weight: f32,
    /// Graph leg's RRF weight. Defaults to `WEIGHT_GRAPH`.
    pub graph_weight: f32,
    /// Opt-in post-fusion access boost (0.0-1.0, default 0 = off): each
    /// hit's fused score is multiplied by
    /// `1 + w·ln(1+count)/(1+ln(1+count))` using NON-BUMPING reads of the
    /// access counters — neutral for never-recalled nodes, log-damped,
    /// bounded below `1+w`. Live counters are db state: like wall-clock
    /// recency, results may shift as counters move.
    pub access_weight: f32,
    /// Opt-in post-fusion corroboration boost (0.0-1.0, default 0 = off):
    /// each hit's fused score is multiplied by `1 + w·(legs_hit − 1)/2`,
    /// where `legs_hit` counts how many legs that actually RAN contained
    /// it (text, vector, graph — a zero-weight leg is skipped entirely and
    /// never counts; expansions live inside the text leg and are not a
    /// leg). The graph leg counts a hit when the PPR list holds it OR when
    /// it is a seed adjacent to ANOTHER seed — `ppr_over_subgraph` excludes
    /// its seeds by contract, and without the co-seed rule the top
    /// `GRAPH_SEEDS` hits (exactly the ones a tie-breaker is for) could
    /// never be graph-corroborated while their own 1-hop neighbors could.
    /// Counting only: fusion inputs are untouched. Exactly 1 for
    /// single-leg hits, bounded `[1, 1+w]` (three legs max). A mild
    /// re-ranker that breaks near-ties toward corroborated hits — RRF's
    /// additive term already rewards agreement. The engine default stays
    /// 0.0; the host default is policy (same split as recency).
    pub corroboration_weight: f32,
    /// Post-fusion score multipliers by node label. Each `(label, weight)`
    /// applies a multiplicative factor to fused scores of matching nodes.
    /// Empty (the default) changes nothing. Each weight is validated
    /// finite and within 0.0..=10.0, same as the leg weights. Duplicate
    /// labels keep the first match. The engine is policy-free: this is
    /// mechanism only; policy defaults arrive from the host/MCP server.
    pub label_weights: Vec<(String, f32)>,
}

pub(crate) const WEIGHT_TEXT: f32 = 1.0;
pub(crate) const WEIGHT_VECTOR: f32 = 1.0;
/// Half weight for the graph leg: adjacency is corroboration, not
/// relevance — a 1-hop neighbor should never outrank a genuine text or
/// vector hit purely by being linked.
pub(crate) const WEIGHT_GRAPH: f32 = 0.5;
/// How many top preliminary-fusion nodes seed the graph leg's traversal —
/// bounded so `graph_boost` costs a handful of 1-hop reads, not one per
/// candidate in a potentially deep leg list.
pub(crate) const GRAPH_SEEDS: usize = 5;

impl RecallQuery {
    /// A query with every tunable at its default: stock leg weights, no
    /// label filter, no access boost, no corroboration boost, graph boost
    /// on, no vector leg, no expansions, default search options. Prefer
    /// struct-update over this
    /// (`RecallQuery { vector: …, ..RecallQuery::new(…) }`) so future
    /// tunables don't break your construction site.
    pub fn new(scopes: ScopeSet, query: impl Into<String>, k: usize) -> Self {
        Self {
            scopes,
            query: query.into(),
            k,
            vector: None,
            expansions: Vec::new(),
            graph_boost: true,
            options: SearchOptions::default(),
            labels: None,
            tombstone_props: Vec::new(),
            text_weight: WEIGHT_TEXT,
            vector_weight: WEIGHT_VECTOR,
            graph_weight: WEIGHT_GRAPH,
            access_weight: 0.0,
            corroboration_weight: 0.0,
            label_weights: Vec::new(),
        }
    }

    /// Validates the tunable fields (weights, access_weight, labels shape).
    /// `Rejected` on any violation — checked by `recall` before any leg
    /// runs. Validation is over the WEIGHTS, not the effective legs: a
    /// caller can zero every leg that actually runs and legitimately get
    /// an empty result (degenerate-but-honest; see the design spec).
    pub(crate) fn validate_tuning(&self) -> Result<(), TopoError> {
        let weight_ok = |w: f32| w.is_finite() && (0.0..=10.0).contains(&w);
        for (name, w) in [
            ("text_weight", self.text_weight),
            ("vector_weight", self.vector_weight),
            ("graph_weight", self.graph_weight),
        ] {
            if !weight_ok(w) {
                return Err(TopoError::Rejected(format!(
                    "{name} must be finite and within 0.0..=10.0, got {w}"
                )));
            }
        }
        if self.text_weight == 0.0 && self.vector_weight == 0.0 && self.graph_weight == 0.0 {
            return Err(TopoError::Rejected(
                "at least one leg weight must be > 0.0".into(),
            ));
        }
        if !(self.access_weight.is_finite() && (0.0..=1.0).contains(&self.access_weight)) {
            return Err(TopoError::Rejected(format!(
                "access_weight must be finite and within 0.0..=1.0, got {}",
                self.access_weight
            )));
        }
        if !(self.corroboration_weight.is_finite()
            && (0.0..=1.0).contains(&self.corroboration_weight))
        {
            return Err(TopoError::Rejected(format!(
                "corroboration_weight must be finite and within 0.0..=1.0, got {}",
                self.corroboration_weight
            )));
        }
        if let Some(labels) = &self.labels {
            if labels.is_empty() {
                return Err(TopoError::Rejected(
                    "labels must not be empty when present — an empty allowlist admits nothing; \
                     omit it to search unfiltered"
                        .into(),
                ));
            }
        }
        for (label, w) in &self.label_weights {
            if label.is_empty() {
                return Err(TopoError::Rejected(
                    "label_weights: label must not be empty".into(),
                ));
            }
            if !weight_ok(*w) {
                return Err(TopoError::Rejected(format!(
                    "label_weights[\"{}\"] must be finite and within 0.0..=10.0, got {w}",
                    label
                )));
            }
        }
        Ok(())
    }
}

impl Db {
    /// Hybrid recall: BM25 text (+ expansions), cosine vector, and 1-hop
    /// graph legs, RRF-fused (`rrf_fuse`), recency-weighted post-fusion,
    /// truncated to `k`. Legs run as sequential read transactions against
    /// the single-applier engine — see the spec for why that is
    /// acceptable. Validation mirrors `search_text_with`.
    pub fn recall(&self, q: &RecallQuery) -> Result<Vec<(NodeRecord, f32)>, TopoError> {
        // search_text_with rejects k == 0 internally too, but only because
        // depth is clamped to >= 30 — recall's own k must be checked
        // explicitly or a k == 0 request would silently run at depth 30.
        if q.k == 0 {
            return Err(TopoError::Rejected("recall requires k > 0".into()));
        }
        q.validate_tuning()?;
        // Check the CALLER's recency options before the leg call zeroes the
        // weight — see SearchOptions::validate_recency for why.
        q.options.validate_recency()?;
        q.options.validate_prop_retain()?;
        q.options.validate_created_range()?;
        if let Some((_, v)) = &q.vector {
            if v.is_empty() {
                return Err(TopoError::Rejected(
                    "recall query vector is empty (host must not send an empty embedding)".into(),
                ));
            }
        }
        let depth = leg_depth(q.k);

        // A zero-weight leg contributes nothing to a fused score but its ids
        // would still ride along at score 0.0 if the list were included
        // unconditionally (`rrf_fuse` does `entry().or_insert(0.0) += w /
        // rank`, which inserts an entry even at `w == 0.0`) — so "weight
        // says on, nothing actually ran" degenerates to Ok(empty)/ghost-free
        // rather than a ghost result. Applied symmetrically across all
        // three legs: skip the underlying read entirely when its weight is
        // 0.0, not just the fusion push — a leg that cannot contribute
        // shouldn't be paid for either.
        let mut records: std::collections::HashMap<crate::NodeId, NodeRecord> =
            std::collections::HashMap::new();
        let mut lists: Vec<(f32, Vec<crate::NodeId>)> = Vec::new();

        // Text leg runs recency-free: recency applies once, post-fusion.
        if q.text_weight > 0.0 {
            let mut leg_options = q.options.clone();
            leg_options.recency_weight = 0.0;
            let text_hits =
                self.search_text_expanded(&q.scopes, &q.query, depth, &leg_options, &q.expansions)?;
            let text_ids: Vec<crate::NodeId> = text_hits.iter().map(|(n, _)| n.id).collect();
            for (n, _) in text_hits {
                records.entry(n.id).or_insert(n);
            }
            lists.push((q.text_weight, text_ids));
        }

        // Vector leg: cosine over the scoped clusters for the named model.
        // An unknown model or a scope with no vectors is an EMPTY leg —
        // legitimately no data — never an error (contrast the empty-vector
        // rejection above, which is a host bug).
        if q.vector_weight > 0.0 {
            if let Some((model, vector)) = &q.vector {
                let vhits = self.search_vector(&crate::VectorQuery {
                    scopes: q.scopes.clone(),
                    model: model.clone(),
                    vector: vector.clone(),
                    k: depth,
                    candidates: None,
                })?;
                let vids: Vec<crate::NodeId> = vhits.iter().map(|(n, _)| n.id).collect();
                for (n, _) in vhits {
                    records.entry(n.id).or_insert(n);
                }
                lists.push((q.vector_weight, vids));
            }
        }
        // Graph leg (spec 2026-07-19): preliminary text+vector fusion picks
        // GRAPH_SEEDS seeds; ONE GRAPH_HOPS-bounded Both-direction traversal
        // (1 hop — eval-tuned, see ppr.rs) materializes their joint neighborhood;
        // PPR with teleport weighted by each
        // seed's preliminary score ranks it. Connectivity now orders the
        // list — a node multiple seeds converge on outranks a node dangling
        // off one seed — replacing the old flat seed-rank concatenation.
        // Seeds stay excluded from the list (ppr_over_subgraph's contract);
        // half weight as ever: adjacency is corroboration, not relevance.
        // Seeds the graph leg vouches for despite `ppr_over_subgraph`'s
        // seed-exclusion contract — counted by `apply_corroboration` below,
        // never fused.
        let mut co_seed_ids: Vec<crate::NodeId> = Vec::new();
        if q.graph_boost && q.graph_weight > 0.0 {
            let prelim = rrf_fuse(&lists);
            let seeds: Vec<(crate::NodeId, f32)> = prelim
                .iter()
                .take(GRAPH_SEEDS)
                .map(|(id, score)| (*id, *score))
                .collect();
            if !seeds.is_empty() {
                let sg = self.traverse(&crate::TraversalQuery {
                    scopes: q.scopes.clone(),
                    seeds: seeds.iter().map(|(id, _)| *id).collect(),
                    max_hops: crate::ppr::GRAPH_HOPS,
                    edge_types: None,
                    direction: crate::Direction::Both,
                    as_of: q.options.now_ms,
                    time_axis: TimeAxis::Valid,
                })?;
                let scored = crate::ppr::ppr_over_subgraph(&sg, &seeds);
                // Corroboration counting (review amendment, spec
                // 2026-08-11): the PPR list excludes its seeds by contract,
                // so a top-GRAPH_SEEDS hit could never earn graph
                // corroboration — exactly the hits the boost exists to
                // separate — while its own 1-hop neighbors could, letting
                // the boost promote a rank-6 neighbor over the top hits it
                // rode in on. For COUNTING only (fusion is untouched), a
                // seed earns the graph leg when it is adjacent to ANOTHER
                // seed: two independently strong hits vouching for each
                // other.
                let seed_set: std::collections::HashSet<crate::NodeId> =
                    seeds.iter().map(|(id, _)| *id).collect();
                for e in &sg.edges {
                    if e.from != e.to && seed_set.contains(&e.from) && seed_set.contains(&e.to) {
                        co_seed_ids.push(e.from);
                        co_seed_ids.push(e.to);
                    }
                }
                let mut by_id: std::collections::HashMap<crate::NodeId, NodeRecord> =
                    sg.nodes.into_iter().map(|n| (n.id, n)).collect();
                let mut graph_ids: Vec<crate::NodeId> = Vec::new();
                for (id, _) in scored {
                    if let Some(rec) = by_id.remove(&id) {
                        graph_ids.push(id);
                        records.entry(id).or_insert(rec);
                    }
                }
                if !graph_ids.is_empty() {
                    lists.push((q.graph_weight, graph_ids));
                }
            }
        }
        let mut fused = rrf_fuse(&lists);
        // Corroboration boost (spec 2026-08-11): mild multiplicative
        // re-ranker toward hits present in multiple legs that actually ran
        // (`lists` holds exactly those — a zero-weight leg was never
        // pushed). Composes with the recency/access/label adjustments
        // below; multiplication commutes, so application order can't change
        // the ranking. Weight 0 skips everything, keeping defaults
        // byte-identical. The co-seed ids ride as their OWN counting entry:
        // they are disjoint from the PPR list (which excludes seeds), so no
        // node can double-count the graph leg — and with only seed-to-seed
        // edges in the neighborhood the PPR list is empty and was never
        // pushed, yet the evidence is real.
        if q.corroboration_weight > 0.0 {
            let mut corrob_lists = lists;
            if !co_seed_ids.is_empty() {
                corrob_lists.push((q.graph_weight, co_seed_ids));
            }
            apply_corroboration(&mut fused, &corrob_lists, q.corroboration_weight);
        }

        let mut out: Vec<(NodeRecord, f32)> = fused
            .into_iter()
            .filter_map(|(id, score)| records.remove(&id).map(|n| (n, score)))
            .collect();
        // Post-fusion label allowlist (spec: filter BEFORE adjustments so
        // counter reads aren't spent on filtered nodes; leg depth >> k, so
        // filtering rarely starves k — and legitimately may).
        if let Some(labels) = &q.labels {
            out.retain(|(n, _)| labels.iter().any(|l| n.label == l.as_str()));
        }
        // Post-fusion prop-retain (the labels-retain slot): the text leg
        // already filters via options, but vector/graph-leg candidates
        // never pass through text search — this catches them.
        if let Some(retain) = &q.options.prop_retain {
            out.retain(|(n, _)| retain.keeps(&n.props));
        }
        // Post-fusion created-range (same slot as prop_retain): the text leg
        // already filters via options, but vector/graph-leg candidates never
        // pass through text search — this catches them.
        if let Some(range) = &q.options.created_range {
            out.retain(|(n, _)| range.keeps(n.id.timestamp_ms() as i64));
        }
        // Tombstone filter: drop candidates the caller marked superseded as of
        // this query's "now". The marker is a timestamp, so an as_of/now_ms set
        // BEFORE the mark keeps the node (its tombstone is in that query's
        // future) — supersession dates a fact, it doesn't erase its history.
        if !q.tombstone_props.is_empty() {
            let now = q.options.now_ms.unwrap_or_else(|| {
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_millis() as i64)
                    .unwrap_or(0)
            });
            out.retain(|(n, _)| {
                !q.tombstone_props.iter().any(|prop| {
                    matches!(n.props.get(prop.as_str()), Some(PropValue::Int(ts)) if *ts <= now)
                })
            });
        }
        apply_adjustments(
            &mut out,
            &q.options,
            q.access_weight,
            &q.label_weights,
            &|id| self.access_count_unbumped(id),
        );
        out.truncate(q.k);
        Ok(out)
    }
}

/// Post-fusion corroboration factor: `1 + weight·(legs_hit − 1)/2`.
/// Exactly 1 for single-leg hits (and for nodes absent from every leg,
/// which cannot occur post-fusion but costs nothing to handle); bounded
/// within `[1, 1 + weight]` since three legs is the maximum. See the
/// design spec's framing: a tie-breaker, not a recall-quality claim.
pub(crate) fn corroboration_factor(weight: f32, legs_hit: usize) -> f32 {
    if weight <= 0.0 || legs_hit <= 1 {
        return 1.0;
    }
    1.0 + weight * (legs_hit as f32 - 1.0) / 2.0
}

/// Applies the corroboration boost to fused scores and re-sorts (score
/// desc, id asc — the same determinism contract as `rrf_fuse`). `lists`
/// are the per-leg COUNTING lists: the lists that were fused, plus at most
/// one extra graph entry for co-seed evidence (see `recall` — disjoint
/// from the fused graph list, so a node still counts each leg once). A leg
/// that never ran is absent and can never count; a zero-weight list is
/// skipped here too for the same reason. Weight 0 skips the code path
/// entirely so results stay byte-identical to today — same pattern as the
/// recency knob.
pub(crate) fn apply_corroboration(
    fused: &mut [(NodeId, f32)],
    lists: &[(f32, Vec<NodeId>)],
    weight: f32,
) {
    if weight <= 0.0 {
        return;
    }
    use std::collections::{HashMap, HashSet};
    let mut legs_hit: HashMap<NodeId, usize> = HashMap::new();
    for (leg_weight, ids) in lists {
        if *leg_weight <= 0.0 {
            continue;
        }
        // A leg counts a node at most once, even if a list ever carried a
        // duplicate id.
        for id in ids.iter().copied().collect::<HashSet<NodeId>>() {
            *legs_hit.entry(id).or_insert(0) += 1;
        }
    }
    for (id, score) in fused.iter_mut() {
        *score *= corroboration_factor(weight, legs_hit.get(id).copied().unwrap_or(0));
    }
    fused.sort_by(|a, b| {
        b.1.partial_cmp(&a.1)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then_with(|| a.0.cmp(&b.0))
    });
}

/// Post-fusion access factor: neutral at count 0, log-damped (recall's own
/// reads bump the counters this reads — damping keeps the loop from
/// running away), bounded below `1 + weight`. See the design spec.
pub(crate) fn access_factor(weight: f32, count: u64) -> f32 {
    if weight <= 0.0 || count == 0 {
        return 1.0;
    }
    let l = ((count as f32) + 1.0).ln();
    1.0 + weight * l / (1.0 + l)
}

/// Combined post-fusion adjustment: recency (the same
/// `(1-w) + w·2^(-age/half_life)` factor `search_text_with` uses), the
/// opt-in access boost (`access_factor`), and label-based score multipliers
/// multiply into one factor per candidate, applied once to fused scores, then
/// re-sorted (score desc, id asc). No-op — and no counter reads — when
/// recency weight is 0, access_weight is 0, AND label_weights is empty,
/// preserving today's early-return shape and the byte-identical-defaults
/// requirement.
pub(crate) fn apply_adjustments(
    out: &mut [(NodeRecord, f32)],
    options: &SearchOptions,
    access_weight: f32,
    label_weights: &[(String, f32)],
    counts: &dyn Fn(crate::NodeId) -> u64,
) {
    let w = options.recency_weight;
    if w <= 0.0 && access_weight <= 0.0 && label_weights.is_empty() {
        return;
    }
    let now = options.now_ms.unwrap_or_else(|| {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("system clock before UNIX epoch")
            .as_millis() as i64
    });
    for (rec, score) in out.iter_mut() {
        let mut factor = 1.0f32;
        if w > 0.0 {
            let age = (now - rec.id.timestamp_ms() as i64).max(0) as f32;
            let half_life = options.half_life_for(rec) as f32;
            factor *= (1.0 - w) + w * (-(age / half_life)).exp2();
        }
        factor *= access_factor(
            access_weight,
            if access_weight > 0.0 {
                counts(rec.id)
            } else {
                0
            },
        );
        // Apply label-based score multiplier: first match wins, duplicate
        // labels keep the first.
        if let Some((_, wl)) = label_weights.iter().find(|(l, _)| rec.label == l.as_str()) {
            factor *= wl;
        }
        *score *= factor;
    }
    out.sort_by(|a, b| {
        b.1.partial_cmp(&a.1)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then_with(|| a.0.id.cmp(&b.0.id))
    });
}

#[cfg(test)]
mod adjustment_tests {
    use super::*;
    use crate::ids::Scope;
    use smol_str::SmolStr;

    fn node_with_label(label: &str, id_num: u128) -> NodeRecord {
        NodeRecord {
            id: crate::NodeId::from_u128(id_num),
            scope: Scope::Id(crate::ids::ScopeId::new()),
            label: SmolStr::new(label),
            props: std::collections::BTreeMap::new(),
            embedding: None,
        }
    }

    fn apply_adjustments_with_labels(
        out: &mut [(NodeRecord, f32)],
        label_weights: &[(String, f32)],
    ) {
        apply_adjustments(out, &SearchOptions::default(), 0.0, label_weights, &|_| 0);
    }

    #[test]
    fn access_factor_is_neutral_monotone_bounded() {
        let f = |w: f32, count: u64| access_factor(w, count);
        assert_eq!(f(1.0, 0), 1.0, "neutral at zero count");
        assert_eq!(f(0.0, 1_000_000), 1.0, "neutral at zero weight");
        assert!(f(1.0, 1) > f(1.0, 0));
        assert!(f(1.0, 100) > f(1.0, 10));
        assert!(f(1.0, u64::MAX) < 2.0, "bounded below 1 + weight");
        assert!(f(0.5, u64::MAX) < 1.5);
    }

    #[test]
    fn label_weight_downranks_matching_label() {
        // Two candidates with adjacent fused scores; the down-weighted label's
        // node must drop below the other after adjustment.
        let mut out = vec![
            (node_with_label("Entity", 1), 0.020f32),
            (node_with_label("Memory", 2), 0.019f32),
        ];
        apply_adjustments_with_labels(&mut out, &[("Entity".to_string(), 0.5)]);
        assert_eq!(
            out[0].0.label.to_string(),
            "Memory",
            "down-weighted Entity must drop"
        );
        assert!(
            (out[1].1 - 0.010).abs() < 1e-6,
            "factor applied multiplicatively"
        );
    }

    #[test]
    fn empty_label_weights_change_nothing() {
        // Same input twice: empty label_weights must be byte-identical to the
        // pre-change behavior (scores AND order).
        let mut out = vec![
            (node_with_label("Entity", 1), 0.020f32),
            (node_with_label("Memory", 2), 0.019f32),
        ];
        let original_order = out.iter().map(|(n, s)| (n.id, *s)).collect::<Vec<_>>();
        let original_scores = out.iter().map(|(_, s)| *s).collect::<Vec<_>>();

        apply_adjustments_with_labels(&mut out, &[]);

        let new_order = out.iter().map(|(n, s)| (n.id, *s)).collect::<Vec<_>>();
        let new_scores = out.iter().map(|(_, s)| *s).collect::<Vec<_>>();

        assert_eq!(
            original_order, new_order,
            "order must not change with empty label_weights"
        );
        assert_eq!(
            original_scores, new_scores,
            "scores must not change with empty label_weights"
        );
    }
}

#[cfg(test)]
mod query_tests {
    use super::*;
    use crate::ids::ScopeSet;

    fn q() -> RecallQuery {
        RecallQuery::new(ScopeSet::of(&[crate::ids::ScopeId::new()]), "hello", 5)
    }

    #[test]
    fn new_carries_todays_defaults() {
        let q = q();
        assert_eq!(q.text_weight, WEIGHT_TEXT);
        assert_eq!(q.vector_weight, WEIGHT_VECTOR);
        assert_eq!(q.graph_weight, WEIGHT_GRAPH);
        assert_eq!(q.access_weight, 0.0);
        assert!(q.labels.is_none());
        assert!(q.graph_boost);
        assert!(q.vector.is_none());
    }

    #[test]
    fn weight_validation_rejects_bad_values() {
        for build in [
            |mut x: RecallQuery| {
                x.text_weight = -0.1;
                x
            },
            |mut x: RecallQuery| {
                x.vector_weight = f32::NAN;
                x
            },
            |mut x: RecallQuery| {
                x.graph_weight = 10.1;
                x
            },
            |mut x: RecallQuery| {
                x.access_weight = 1.1;
                x
            },
            |mut x: RecallQuery| {
                x.access_weight = f32::INFINITY;
                x
            },
            |mut x: RecallQuery| {
                x.text_weight = 0.0;
                x.vector_weight = 0.0;
                x.graph_weight = 0.0;
                x
            },
            |mut x: RecallQuery| {
                x.labels = Some(vec![]);
                x
            },
        ] {
            let bad = build(q());
            assert!(
                matches!(bad.validate_tuning(), Err(crate::TopoError::Rejected(_))),
                "must reject: {bad:?}"
            );
        }
    }

    #[test]
    fn label_weights_validation_rejects_bad_factors() {
        // NaN, negative, and 10.1 all rejected; 0.0 and 10.0 accepted.
        for build in [
            |mut x: RecallQuery| {
                x.label_weights = vec![("Entity".to_string(), -0.1)];
                x
            },
            |mut x: RecallQuery| {
                x.label_weights = vec![("Entity".to_string(), f32::NAN)];
                x
            },
            |mut x: RecallQuery| {
                x.label_weights = vec![("Entity".to_string(), 10.1)];
                x
            },
            |mut x: RecallQuery| {
                x.label_weights = vec![("".to_string(), 0.5)];
                x
            },
        ] {
            let bad = build(q());
            assert!(
                matches!(bad.validate_tuning(), Err(crate::TopoError::Rejected(_))),
                "must reject: {bad:?}"
            );
        }

        // Valid edge cases should pass.
        assert!(RecallQuery {
            label_weights: vec![("Entity".to_string(), 0.0)],
            ..q()
        }
        .validate_tuning()
        .is_ok());
        assert!(RecallQuery {
            label_weights: vec![("Entity".to_string(), 10.0)],
            ..q()
        }
        .validate_tuning()
        .is_ok());
        assert!(RecallQuery {
            label_weights: vec![("Entity".to_string(), 0.5), ("Memory".to_string(), 1.5)],
            ..q()
        }
        .validate_tuning()
        .is_ok());
    }

    #[test]
    fn default_tuning_validates() {
        assert!(q().validate_tuning().is_ok());
    }
}

#[cfg(test)]
mod corroboration_tests {
    use super::*;
    use crate::ids::ScopeSet;

    fn id(n: u128) -> NodeId {
        NodeId::from_u128(n)
    }

    fn q() -> RecallQuery {
        RecallQuery::new(ScopeSet::of(&[crate::ids::ScopeId::new()]), "hello", 5)
    }

    #[test]
    fn corroboration_weight_defaults_to_zero() {
        assert_eq!(
            q().corroboration_weight,
            0.0,
            "off by default: engine is mechanism"
        );
    }

    #[test]
    fn corroboration_factor_neutral_and_bounded() {
        assert_eq!(
            corroboration_factor(1.0, 0),
            1.0,
            "absent from every leg: neutral"
        );
        assert_eq!(
            corroboration_factor(1.0, 1),
            1.0,
            "single-leg hit: exactly 1"
        );
        assert_eq!(corroboration_factor(0.0, 3), 1.0, "weight 0: neutral");
        assert!((corroboration_factor(0.2, 2) - 1.1).abs() < 1e-6);
        assert_eq!(
            corroboration_factor(1.0, 3),
            2.0,
            "three legs at weight 1: the 1 + w ceiling"
        );
        assert!(
            corroboration_factor(0.5, 3) <= 1.5,
            "bounded within [1, 1 + w]"
        );
    }

    #[test]
    fn corroboration_weight_validation_matches_access_weight_envelope() {
        for bad in [-0.1, 1.1, f32::NAN, f32::INFINITY] {
            let query = RecallQuery {
                corroboration_weight: bad,
                ..q()
            };
            assert!(
                matches!(query.validate_tuning(), Err(crate::TopoError::Rejected(_))),
                "must reject corroboration_weight {bad}"
            );
        }
        for good in [0.0, 0.2, 1.0] {
            let query = RecallQuery {
                corroboration_weight: good,
                ..q()
            };
            assert!(
                query.validate_tuning().is_ok(),
                "must accept corroboration_weight {good}"
            );
        }
    }

    #[test]
    fn corroboration_zero_weight_is_byte_identical() {
        let lists = vec![(1.0, vec![id(1), id(2), id(3)]), (0.5, vec![id(3), id(2)])];
        let mut fused = rrf_fuse(&lists);
        let before = fused.clone();
        apply_corroboration(&mut fused, &lists, 0.0);
        assert_eq!(before, fused, "weight 0 skips the code path entirely");
    }

    #[test]
    fn corroborated_hit_wins_near_tie_only_when_weight_positive() {
        // A is #1 in the text leg; B is #2 there AND the barely-weighted
        // vector leg's #1. At w = 0, A's 1/61 edges out B's
        // 1/62 + 0.01/61 — a near-tie resolved toward the single-leg hit.
        // Any positive weight breaks it the other way: B's two-leg factor
        // (1.1 at w = 0.2) overtakes while A's single-leg factor stays 1.
        let lists = vec![
            (1.0, vec![id(1), id(2)]), // text: A, B
            (0.01, vec![id(2)]),       // vector: B only
        ];
        let mut fused = rrf_fuse(&lists);
        assert_eq!(fused[0].0, id(1), "boost off: near-tie resolves to A");
        apply_corroboration(&mut fused, &lists, 0.0);
        assert_eq!(fused[0].0, id(1), "weight 0 must not touch the ordering");
        apply_corroboration(&mut fused, &lists, 0.2);
        assert_eq!(fused[0].0, id(2), "corroborated B overtakes once boosted");
    }

    #[test]
    fn zero_weight_legs_do_not_count_toward_corroboration() {
        // B rides in a second list whose leg weight is 0.0 — that leg never
        // ran, so B stays a single-leg hit and every factor is exactly 1.
        let ran = vec![(1.0, vec![id(1), id(2)])];
        let mut fused = rrf_fuse(&ran);
        let before = fused.clone();
        let with_dead_leg = vec![(1.0, vec![id(1), id(2)]), (0.0, vec![id(2)])];
        apply_corroboration(&mut fused, &with_dead_leg, 1.0);
        assert_eq!(before, fused, "a zero-weight leg must not create a boost");
    }
}