tidaldb/tidal/src/query/executor/pipeline.rs

//! RETRIEVE query execution pipeline.
//!
//! Contains the `execute()` method that drives the 6-stage pipeline.
//! Stage 3 (signal scoring) is delegated to `stage3_score` in `mod.rs`.

#![allow(clippy::too_many_lines)]

use std::{collections::HashSet, time::Instant};

use super::{RetrieveExecutor, candidate_gen, post_filter, user_filter};
use crate::{
    query::{
        retrieve::{Cursor, QueryError, Results, Retrieve, RetrieveResult, Signal},
        stats::QueryStats,
    },
    ranking::{
        diversity::DiversitySelector,
        profile::{CandidateStrategy, Sort},
    },
    schema::{EntityId, Timestamp},
    storage::indexes::{
        bitmap::BitmapIndex,
        filter::{FilterEvaluator, FilterResult},
        range::RangeIndex,
    },
};

/// ANN candidate over-fetch multiplier (m12p2): fetch `limit * this` nearest
/// neighbours so Stage 2/2.5 filtering and Stage 4 diversity can trim back to
/// `limit` without starving. Mirrors the SEARCH ANN over-fetch.
const ANN_OVERFETCH: usize = 10;

/// Minimum ANN candidate count regardless of `limit`, so a tiny-limit feed still
/// seeds a usable candidate pool.
const ANN_CANDIDATE_FLOOR: usize = 200;

impl RetrieveExecutor<'_> {
    /// Execute a RETRIEVE query through the 6-stage pipeline.
    ///
    /// # Errors
    ///
    /// Returns `QueryError` on validation failure, missing profile, or
    /// unsupported candidate strategy.
    #[allow(clippy::too_many_lines)]
    pub fn execute(&self, query: &Retrieve) -> Result<Results, QueryError> {
        let query_start = Instant::now();
        let mut stats = QueryStats::new(query.profile.name.clone());
        stats.degradation_level = self.degradation_level as u8;

        // Validate the query against the profile registry.
        query.validate(self.profile_registry)?;

        // Compute the combined filter ONCE and reuse it across every stage.
        // `combined_filter()` clones the entire `filters` Vec into a fresh
        // `And` node on each call, and the pipeline consults it at 6+ points.
        let combined_filter = query.combined_filter();

        // Reject filter shapes with no sound semantics BEFORE any stage runs, so
        // a bad query fails loudly instead of silently returning the inverse set
        // (negated deferred post-filters) or AND-instead-of-OR (any deferred
        // post-filter under an OR — they resolve to the full universe in Stage 2
        // and are then applied conjunctively, so the OR would collapse to AND).
        if let Some(ref filter_expr) = combined_filter {
            user_filter::reject_negated_deferred_filters(filter_expr)?;
            user_filter::reject_or_of_deferred_filters(filter_expr)?;
        }

        tracing::trace!(
            profile = %query.profile.name,
            limit = query.limit,
            has_filter = combined_filter.is_some(),
            has_diversity = query.diversity.is_some(),
            "retrieve query start"
        );

        // Resolve the profile.
        #[allow(clippy::option_if_let_else)]
        let profile = match query.profile.version {
            Some(v) => self
                .profile_registry
                .get_version(&query.profile.name, v)
                .map_err(|_| QueryError::ProfileNotFound(query.profile.name.clone()))?,
            None => self
                .profile_registry
                .get(&query.profile.name)
                .map_err(|_| QueryError::ProfileNotFound(query.profile.name.clone()))?,
        };

        let mut warnings: Vec<String> = Vec::new();

        // ── Stage 1: Candidate Generation ───────────────────────────────
        // When for_creator is specified, restrict candidates to that creator's
        // items using the CreatorItemsBitmap. This is O(k) where k = the
        // creator's item count, avoiding the O(N) scan cap that would miss
        // items at high IDs in a large catalog.
        //
        // The two `for_creator` arms below are kept SEPARATE on purpose. A single
        // `if let creator_id && let creator_items` chain would make the whole
        // condition false when `for_creator` is set but `creator_items` is absent,
        // falling through to the full-universe scan and SILENTLY dropping the
        // creator restriction (returning every creator's items). Instead, when the
        // scope is requested but cannot be satisfied, we warn and return an EMPTY
        // candidate set — the requested scope yields no items rather than the
        // inverse (all items). Mirrors the warn-on-missing-index pattern in Stage 2.
        let has_user_context = query.for_user.is_some();
        let mut candidates = if let Some(creator_id) = query.for_creator {
            if let Some(creator_items) = self.creator_items {
                creator_items
                    .get(creator_id.as_u64())
                    .map_or_else(Vec::new, |bm| {
                        bm.iter()
                            .map(|id_u32| EntityId::new(u64::from(id_u32)))
                            .collect()
                    })
            } else {
                warnings.push(
                    "FOR CREATOR scope requested but the creator-items index is unavailable; \
                     returning no candidates (refusing to fall back to a full-universe scan \
                     that would ignore the creator restriction)"
                        .to_string(),
                );
                tracing::warn!(
                    creator_id = creator_id.as_u64(),
                    "for_creator scope unsatisfiable: creator_items index absent; returning empty"
                );
                Vec::new()
            }
        } else {
            match &profile.candidate_strategy {
                CandidateStrategy::Scan { .. } => {
                    candidate_gen::scan_candidates(self.universe, query.limit, has_user_context)
                }
                CandidateStrategy::SignalRanked { signal, .. } => {
                    // m12p2: candidates come from the cached per-signal-type top-K
                    // (O(K)). When the signal is absent from the schema, or no
                    // writes have landed yet, that yields nothing — fall back to a
                    // scan so the read still works (e.g. trending on a fresh corpus
                    // with no views yet) rather than returning an empty feed.
                    let ranked =
                        candidate_gen::signal_ranked_candidates(self.ledger, signal, query.limit);
                    if ranked.is_empty() {
                        warnings.push(format!(
                            "SignalRanked('{signal}') produced no candidates \
                             (signal absent or no writes yet); falling back to scan"
                        ));
                        candidate_gen::scan_candidates(self.universe, query.limit, has_user_context)
                    } else {
                        ranked
                    }
                }
                CandidateStrategy::Ann {
                    limit: ann_limit, ..
                } => {
                    // (`top_clusters` is consulted by the db layer when it resolves
                    // the fan-out set into `ann_query_vectors`; the executor just
                    // runs whatever vectors it was handed.)
                    // m12p2: ANN candidate generation in RETRIEVE — O(ef_search)
                    // nearest neighbours of the resolved query vector (the user's
                    // preference vector for `for_you`, the seed item's embedding
                    // for `similar_to`), over-fetched so Stage 2/2.5 + diversity
                    // have room. Degrades to a scan (with a caller- AND
                    // operator-visible note) when no registry/query vector is
                    // available — an anonymous read or a user with no preference
                    // vector yet — so the read keeps working instead of erroring.
                    // Over-fetch limit×10, floored so tiny feeds still seed a pool,
                    // and capped by the profile's declared `Ann.limit` so a huge
                    // page size can't drive an unbounded ANN beam.
                    let k = query
                        .limit
                        .saturating_mul(ANN_OVERFETCH)
                        .max(ANN_CANDIDATE_FLOOR)
                        .min((*ann_limit).max(ANN_CANDIDATE_FLOOR));
                    // Prefer the multi-vector fan-out set (warm `for_you`) when the
                    // db layer resolved one; fall back to the single query vector
                    // (`similar_to`, cold-start). Both are BORROWED — no per-request
                    // clone of the (up to 3 × dim) centroids on the serve path.
                    // `ann_candidates_multi` is byte-identical to the single search
                    // when the slice has ≤1 element, so a single-cluster user pays
                    // no fan-out overhead.
                    let single_fallback;
                    let fan_out: &[Vec<f32>] = match &self.ann_query_vectors {
                        Some(vs) if !vs.is_empty() => vs.as_slice(),
                        _ => match &self.ann_query_vector {
                            Some(v) => {
                                single_fallback = std::slice::from_ref(v);
                                single_fallback
                            }
                            None => &[],
                        },
                    };
                    if let Some(registry) = self.embedding_registry
                        && !fan_out.is_empty()
                    {
                        let slot = self.item_embedding_slot.unwrap_or("content");
                        // ef_search=0 → the index default (ANN_DEFAULT_EF_SEARCH),
                        // matching the m12p2 single-vector path. TODO(tuning):
                        // thread the per-query ef_search the m12p3 knob resolves so
                        // the fan-out can trade beam width per cluster (the M-sweep
                        // open question). Pinned to the default for now.
                        let ann =
                            candidate_gen::ann_candidates_multi(registry, slot, fan_out, k, 0);
                        if ann.is_empty() {
                            warnings.push(
                                "ANN candidate generation returned no candidates; \
                                 falling back to scan"
                                    .to_string(),
                            );
                            tracing::warn!(
                                profile = %query.profile.name,
                                "ANN returned no candidates (empty/absent slot); scan fallback"
                            );
                            candidate_gen::scan_candidates(
                                self.universe,
                                query.limit,
                                has_user_context,
                            )
                        } else {
                            ann
                        }
                    } else {
                        warnings.push(
                            "ANN candidate strategy: no query vector resolvable \
                             (anonymous read or no preference vector yet); falling back to scan"
                                .to_string(),
                        );
                        tracing::debug!(
                            profile = %query.profile.name,
                            has_registry = self.embedding_registry.is_some(),
                            "ANN strategy without a query vector; scan fallback"
                        );
                        candidate_gen::scan_candidates(self.universe, query.limit, has_user_context)
                    }
                }
                CandidateStrategy::Relationship => {
                    // M3: source candidates from the user's followed creators.
                    // Uses the follows index in user_state + creator_items bitmap to build
                    // a candidate set from items produced by followed creators.
                    if let (Some(user_id), Some(user_state), Some(creator_items)) =
                        (query.for_user, self.user_state, self.creator_items)
                    {
                        let followed = user_state.followed_creators_vec(user_id);
                        if followed.is_empty() {
                            warnings.push(
                            "Relationship strategy: user follows no creators; falling back to scan"
                                .to_string(),
                        );
                            candidate_gen::scan_candidates(self.universe, query.limit, true)
                        } else {
                            let bitmap = creator_items.union_for(&followed);
                            if bitmap.is_empty() {
                                warnings.push(
                                "Relationship strategy: followed creators have no items; falling back to scan"
                                    .to_string(),
                            );
                                candidate_gen::scan_candidates(self.universe, query.limit, true)
                            } else {
                                bitmap
                                    .iter()
                                    .map(|id_u32| EntityId::new(u64::from(id_u32)))
                                    .collect()
                            }
                        }
                    } else {
                        warnings.push(
                            "Relationship strategy requires FOR USER clause; falling back to scan"
                                .to_string(),
                        );
                        candidate_gen::scan_candidates(self.universe, query.limit, false)
                    }
                }
                other => {
                    return Err(QueryError::UnsupportedStrategy(format!(
                        "{other:?} requires M4+ infrastructure"
                    )));
                }
            }
        };

        // Apply exclude list.
        if !query.exclude.is_empty() {
            let exclude_set: HashSet<u64> = query.exclude.iter().map(|id| id.as_u64()).collect();
            candidates.retain(|id| !exclude_set.contains(&id.as_u64()));
        }

        // M7p2: ReducedCandidates — cap scan set to reduce Stage 3 scoring work.
        //
        // A blind `truncate(cap)` keeps the FIRST `cap` candidates in scan order.
        // `scan_candidates` iterates the universe bitmap in ASCENDING entity-ID
        // order, so a blind truncate keeps the LOWEST IDs. For an ID-ordered
        // profile that ranks by DESCENDING entity ID (e.g. `New`), the highest IDs
        // are exactly the items that should rank at the top — so a blind truncate
        // drops the top-ranked items before scoring and corrupts the "new"
        // ranking. We therefore truncate by the profile's primary ordering key:
        // for `New`, keep the `cap` HIGHEST IDs via an O(N) `select_nth_unstable`
        // partition (no full sort; Stage 3 still orders the survivors). Other
        // profiles score on signal state rather than scan position, so their
        // truncation order is not load-bearing and a plain truncate is retained.
        if self.degradation_level.reduces_candidates() {
            let cap = (query.limit * 4).max(100);
            if candidates.len() > cap {
                if matches!(profile.sort, Some(Sort::New)) {
                    // Partition so the `cap` highest IDs occupy [0, cap): the items
                    // that rank highest under descending-ID order survive.
                    candidates.select_nth_unstable_by(cap - 1, |a, b| b.as_u64().cmp(&a.as_u64()));
                }
                candidates.truncate(cap);
            }
        }

        stats.candidates_considered = candidates.len();

        tracing::trace!(
            candidates = candidates.len(),
            "stage 1: candidates generated"
        );

        // ── Stage 2: Filter Evaluation ──────────────────────────────────
        if let Some(ref filter_expr) = combined_filter {
            // Resolve the universe up front, BEFORE building the evaluator. The
            // `FilterEvaluator` resolves deferred variants
            // (`MinSignal`/`MaxSignal`/`NearLocation`/`InCollection`/user-state)
            // to `self.universe` and intersects every `And` child against it.
            // That is only sound when the universe is a populated, trustworthy
            // bitmap. When it is unavailable — not wired (`None`) or the read
            // lock is poisoned — the empty fallback would make the intersection
            // silently drop EVERY candidate (including ones that should pass)
            // before the Stage 2.2/2.3/2.4 deferred post-filters ever run. The
            // actual deferred filtering happens in those later stages, so the
            // correct degraded behavior — mirroring the SEARCH pipeline's
            // `apply_metadata_filter` (W28) — is to SKIP the index intersection
            // entirely (retain all) and surface a warning, rather than collapse
            // to an empty result set.
            #[allow(clippy::option_if_let_else)]
            let universe_guard = match self.universe {
                Some(u) => {
                    if let Ok(guard) = u.read() {
                        Some(guard)
                    } else {
                        warnings.push(
                            "universe lock poisoned; metadata index pre-filter degraded — \
                         relying on post-filters"
                                .to_string(),
                        );
                        None
                    }
                }
                None => None,
            };

            if let Some(universe_ref) = universe_guard.as_deref() {
                // Build a per-query FilterEvaluator from whatever indexes are
                // available. When an index is `None` (no items have been written
                // to it yet), the empty fallback causes filters on that field to
                // match nothing -- which is the correct semantic: if no items
                // have a "category" field indexed, then `CategoryEq("jazz")`
                // should return zero results.
                let empty_bitmap = BitmapIndex::new("_empty");
                let empty_dur = RangeIndex::<u32>::new("_empty");
                let empty_ts = RangeIndex::<u64>::new("_empty");

                let cat = self.category_index.unwrap_or_else(|| {
                    tracing::debug!(
                        "category_index unavailable; CategoryEq/In filters will return no matches"
                    );
                    &empty_bitmap
                });
                let fmt = self.format_index.unwrap_or_else(|| {
                    tracing::debug!(
                        "format_index unavailable; FormatEq/In filters will return no matches"
                    );
                    &empty_bitmap
                });
                let cre = self.creator_index.unwrap_or(&empty_bitmap);
                let tag = self.tag_index.unwrap_or(&empty_bitmap);
                let dur = self.duration_index.unwrap_or(&empty_dur);
                let ts = self.created_at_index.unwrap_or(&empty_ts);

                let evaluator = FilterEvaluator::new(cat, fmt, cre, tag, dur, ts, universe_ref);
                match evaluator.evaluate(filter_expr) {
                    FilterResult::Bitmap(bitmap) => {
                        // INCLUSION test against the match bitmap. The match bitmap
                        // holds only 32-bit ids, so a candidate id that overflows
                        // `u32` can never be a legitimate member and must be DROPPED
                        // — a raw `id as u32` truncation would alias `2^32 + k` onto
                        // the low slot `k`, which may be present in the bitmap, and
                        // wrongly RETAIN an item that does not satisfy the filter.
                        // `SignalRanked` candidates come straight from the ledger,
                        // which keys on full-u64 ids, so a `> u32::MAX` id is valid
                        // input that reaches here. Route through the checked
                        // `entity_as_u32` (drop-on-overflow), matching the
                        // InCollection/SocialGraph arms and the Stage 2.5 helper.
                        candidates.retain(|id| {
                            super::entity_as_u32(*id).is_some_and(|i| bitmap.contains(i))
                        });
                    }
                    FilterResult::Predicate(pred) => {
                        candidates.retain(|id| pred(id.as_u64()));
                    }
                }
            }
        }

        tracing::trace!(candidates = candidates.len(), "stage 2: filter applied");

        // ── Stage 2.2/2.3/2.4: Deferred Post-Filters ────────────────────
        // InCollection (M6p4), MinSignal/MaxSignal thresholds (M6p3), and
        // NearLocation geo (M6p3). These evaluate to the full universe at the
        // FilterEvaluator level and must be applied here against the live
        // collection index, ledger, and item metadata. Shared verbatim with the
        // SEARCH pipeline via `post_filter::apply_deferred_post_filters` so the
        // two surfaces cannot drift (a missing arm there silently leaked items
        // that violated the filter).
        if let Some(ref filter_expr) = combined_filter {
            let ctx = post_filter::PostFilterCtx {
                ledger: self.ledger,
                collection_index: self.collection_index,
                items_storage: self.items_storage,
                degradation_level: self.degradation_level,
            };
            post_filter::apply_deferred_post_filters(filter_expr, &mut candidates, &ctx)?;
        }

        // ── Stage 2.5: User-Context Filtering (M3) ─────────────────────
        // When FOR USER is specified, exclude seen/hidden items, items from
        // blocked creators, and hard negatives, then apply user-state inclusion
        // filters. The whole block is shared verbatim with the SEARCH pipeline
        // via `super::apply_user_context_suppression` so the two surfaces cannot
        // drift (a missing arm on one side silently leaks an item the other
        // suppresses).
        if let Some(user_id) = query.for_user {
            super::apply_user_context_suppression(
                &mut candidates,
                user_id,
                self.user_state,
                self.creator_items,
                self.hard_negatives,
                combined_filter.as_ref(),
            );

            tracing::trace!(
                candidates = candidates.len(),
                user_id = user_id,
                "stage 2.5: user-context filter applied"
            );
        }

        stats.candidates_after_filter = candidates.len();
        stats.filters_applied = query.filters.len();

        // ── DateSaved validation (M6p3) ─────────────────────────────────
        // DateSaved sort requires FOR USER context. Fail early with a clear
        // error rather than silently returning bad ordering.
        if matches!(profile.sort, Some(Sort::DateSaved)) && query.for_user.is_none() {
            return Err(QueryError::InvalidFilter {
                field: "sort".to_string(),
                reason: "DateSaved sort requires FOR USER context".to_string(),
            });
        }

        // ── Stage 3: Signal Scoring ─────────────────────────────────────
        let scoring_start = Instant::now();
        let now = Timestamp::now();
        // `item_metadata` is the per-candidate metadata map Stage 3 pre-loaded
        // once from `items_storage`. It is retained (not discarded) so Stage 4.5
        // can resolve `creator_id` from it instead of re-reading storage per
        // candidate inside `apply_notification_caps` (see the re-population below).
        let (scored, item_metadata) = self.stage3_score(
            &candidates,
            profile,
            query,
            combined_filter.as_ref(),
            now,
            &mut stats,
            &mut warnings,
        )?;
        let total_scored = scored.len();
        stats.scoring_time_us = scoring_start.elapsed().as_micros() as u64;
        tracing::trace!(scored = scored.len(), "stage 3: scored");

        // ── Stage 4: Diversity Enforcement ──────────────────────────────
        // Use explicit query diversity if set; otherwise fall back to the profile's
        // DiversitySpec so that built-in profiles (e.g. notification: max_per_creator=1)
        // are always enforced even when the caller does not set query.diversity.
        let profile_diversity: Option<crate::ranking::diversity::DiversityConstraints> = {
            let spec = &profile.diversity;
            if spec.max_per_creator.is_some() || spec.format_mix_max_fraction.is_some() {
                Some(crate::ranking::diversity::DiversityConstraints {
                    max_per_creator: spec.max_per_creator,
                    format_mix_max_fraction: spec.format_mix_max_fraction,
                    ..crate::ranking::diversity::DiversityConstraints::default()
                })
            } else {
                None
            }
        };
        let effective_diversity = query.diversity.as_ref().or(profile_diversity.as_ref());
        // M7p2: NoDiversity — skip diversity enforcement entirely under load.
        let diversity_start = Instant::now();
        let (final_candidates, constraints_satisfied) = if self.degradation_level.skips_diversity()
        {
            warnings.push("diversity enforcement skipped due to load degradation".to_string());
            (scored, true)
        } else if let Some(diversity) = effective_diversity {
            // Pass the full scored set (not query.limit) so diversity constraints are
            // applied before pagination slicing. This ensures the page boundary does
            // not artificially inflate or reduce the diversity-enforced count.
            let result = DiversitySelector::select(&scored, diversity, scored.len());
            let satisfied = result.constraints_satisfied;
            if !satisfied {
                for v in &result.violations {
                    warnings.push(format!(
                        "diversity constraint '{}' relaxed: {}",
                        v.constraint, v.detail
                    ));
                }
            }
            (result.selected, satisfied)
        } else {
            (scored, true)
        };

        stats.diversity_time_us = diversity_start.elapsed().as_micros() as u64;
        stats.candidates_after_diversity = final_candidates.len();

        tracing::trace!(
            final_count = final_candidates.len(),
            "stage 4: diversity enforced"
        );

        // ── Stage 4.5: Notification Cap FILTER (M6p6) ───────────────────
        // Trims the post-diversity set against the daily caps WITHOUT recording.
        // Recording is deferred to Stage 5 so it counts only the delivered page,
        // not items pagination drops (C-CRITICAL).
        let final_candidates =
            if let (Some(caps), Some(user_id)) = (query.notification_caps, query.for_user) {
                let mut final_candidates = final_candidates;
                // Backfill `creator_id` from the metadata Stage 3 already loaded so
                // the per-creator cap check never re-reads storage per candidate.
                // Stage 3 populates `creator_id` on the *scored* set, but only when
                // the field was `None`; backfilling again here is cheap (a HashMap
                // lookup) and closes the case where diversity selection or a later
                // pass left a survivor with an unresolved creator. Without this, the
                // `resolve_candidate_creator_id` fallback issues a storage `get()`
                // for every still-unresolved candidate. Shares the
                // one `backfill_creator_ids` helper with Stage 3 so the two passes
                // cannot drift on the parse rule.
                super::backfill_creator_ids(&mut final_candidates, &item_metadata);
                self.apply_notification_caps(final_candidates, caps, user_id, &query.profile.name)
            } else {
                final_candidates
            };

        tracing::trace!(
            final_count = final_candidates.len(),
            "stage 4.5: notification cap filter applied (recording deferred to stage 5)"
        );

        // ── Stage 5: Result Assembly ────────────────────────────────────
        // `Cursor::offset` is fallible: a keyset cursor used as an offset (or a
        // 32-bit-overflowing value) is a loud `InvalidCursor` rather than a silent
        // misread.
        let offset = query.cursor.as_ref().map_or(Ok(0), Cursor::offset)?;
        // Guard against cursor-offset overflow: an attacker-supplied cursor can
        // carry an `offset` near `usize::MAX`, and `offset + limit` would wrap.
        // `saturating_add` clamps to `usize::MAX`; the `.min(len)` then keeps the
        // slice bounds valid and the empty-page branch below handles `offset >= len`.
        let end = offset
            .saturating_add(query.limit)
            .min(final_candidates.len());
        let page_slice = if offset < final_candidates.len() {
            &final_candidates[offset..end]
        } else {
            &[]
        };

        // ── Notification delivery recording (post-pagination) ───────────────
        // Record deliveries against ONLY the page actually returned to the caller,
        // not the full post-diversity set. Stage 4.5 filtered without recording;
        // this records the delivered page through the tracker's atomic
        // check-and-record so the daily cap is consumed by exactly the
        // notifications the caller receives — never items trimmed by pagination
        // (C-CRITICAL: notification-cap recording must follow pagination). For
        // non-notification profiles or no tracker this returns the page unchanged.
        let recorded_page: Vec<crate::ranking::executor::ScoredCandidate> =
            if let (Some(caps), Some(user_id)) = (query.notification_caps, query.for_user) {
                self.record_notification_deliveries(page_slice, caps, user_id, &query.profile.name)
            } else {
                page_slice.to_vec()
            };
        let page: &[crate::ranking::executor::ScoredCandidate] = &recorded_page;

        let items: Vec<RetrieveResult> = page
            .iter()
            .enumerate()
            .map(|(i, c)| {
                let signals: Vec<Signal> = c
                    .signal_snapshot
                    .iter()
                    .map(|(name, value)| Signal {
                        name: name.as_str().to_owned(),
                        value: *value,
                        source: "decay_score".to_string(),
                    })
                    .collect();

                let reasons = crate::ranking::reason::select_top_reasons(c.reasons.clone());

                RetrieveResult {
                    entity_id: c.entity_id,
                    score: c.score,
                    rank: offset + i + 1,
                    signals,
                    reasons,
                }
            })
            .collect();

        // M2: offset-based pagination. The cursor encodes a list position rather
        // than a (score, entity_id) keyset. This is stable only when the ranked
        // list does not change between page requests (no concurrent signal writes
        // during iteration). Replace with keyset pagination in M3.
        let next_cursor = if end < final_candidates.len() {
            Some(Cursor::from_offset(end))
        } else {
            None
        };

        tracing::trace!(
            returned = items.len(),
            has_next_cursor = next_cursor.is_some(),
            "stage 5: results assembled"
        );

        stats.total_time_us = query_start.elapsed().as_micros() as u64;

        Ok(Results {
            items,
            next_cursor,
            total_candidates: total_scored,
            constraints_satisfied,
            warnings,
            session_snapshot: self.session_snapshot.clone(),
            degradation_level: self.degradation_level,
            stats,
            // Local queries carry no governance metadata; community-overlay
            // queries (M9p2+) populate this.
            policy_metadata: crate::query::retrieve::types::PolicyMetadata::default(),
        })
    }
}

// ── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::time::Duration;

    // `Retrieve`, `CandidateStrategy`, `EntityId`, and `Timestamp` are in scope
    // via `use super::*`; `FilterExpr` is test-only here (the production pipeline
    // body no longer names it directly), so it is imported explicitly to keep the
    // module-level imports free of an otherwise-unused name.
    use super::*;
    use crate::{
        ranking::{
            profile::{DiversitySpec, RankingProfile},
            registry::ProfileRegistry,
        },
        schema::{DecaySpec, EntityKind, SchemaBuilder, Window as RankWindow},
        signals::{NoopWalWriter, SignalLedger},
        storage::indexes::FilterExpr,
    };

    /// Schema with a single `view` signal so a `min_signal("view", …)` deferred
    /// filter resolves and the `SignalRanked` strategy can generate candidates.
    fn view_schema() -> crate::schema::Schema {
        let mut builder = SchemaBuilder::new();
        let _ = builder
            .signal(
                "view",
                EntityKind::Item,
                DecaySpec::Exponential {
                    half_life: Duration::from_secs(7 * 24 * 3600),
                },
            )
            .windows(&[RankWindow::AllTime])
            .add();
        builder.build().unwrap()
    }

    /// Registry holding one `SignalRanked` profile. Unlike the built-in `Scan`
    /// profiles, `SignalRanked` sources candidates from the ledger, NOT the
    /// universe bitmap — exactly what the W28 "no universe wired" path needs so
    /// candidate generation does not vanish before the deferred post-filter runs.
    fn signal_ranked_registry() -> ProfileRegistry {
        let mut reg = ProfileRegistry::new();
        reg.register(RankingProfile {
            name: "view_ranked".to_owned(),
            version: 1,
            candidate_strategy: CandidateStrategy::SignalRanked {
                signal: "view".to_owned(),
                window: RankWindow::AllTime,
            },
            boosts: vec![],
            decay: None,
            gates: vec![],
            penalties: vec![],
            excludes: vec![],
            diversity: DiversitySpec::default(),
            exploration: 0.0,
            sort: None,
            is_builtin: false,
        })
        .unwrap();
        reg
    }

    /// W28 regression: a deferred `MinSignal` filter with NO universe wired must
    /// still return candidates from `SignalRanked` retrieval and let the Stage
    /// 2.3 post-filter trim them — NOT collapse to an empty set.
    ///
    /// Before the fix, RETRIEVE Stage 2 fell back to an EMPTY `empty_universe`
    /// when `self.universe` was `None`/poisoned, ran the `FilterEvaluator`
    /// (which resolves `MinSignal` to that empty universe), and `candidates.retain`
    /// dropped EVERY candidate before the post-filter could run. SEARCH already
    /// skipped the index intersection in that case; this mirrors the same fix for
    /// RETRIEVE.
    #[test]
    fn retrieve_min_signal_with_no_universe_still_postfilters() {
        let schema = view_schema();
        let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
        let profile_reg = signal_ranked_registry();

        // Item 1 has 5 views (>= 3), item 2 has 1 view (< 3). Both have a `view`
        // signal so both become SignalRanked candidates.
        let now = Timestamp::now();
        for _ in 0..5 {
            ledger
                .record_signal("view", EntityId::new(1), 1.0, now)
                .unwrap();
        }
        ledger
            .record_signal("view", EntityId::new(2), 1.0, now)
            .unwrap();

        // NOTE: `universe = None` — the exact condition that previously dropped
        // every candidate in Stage 2.
        let exec = RetrieveExecutor::new(
            &ledger,
            &profile_reg,
            None, // category
            None, // format
            None, // creator
            None, // tag
            None, // duration
            None, // created_at
            None, // universe
        );
        let query = Retrieve::builder()
            .profile("view_ranked")
            .filter(FilterExpr::min_signal("view", 3.0))
            .limit(10)
            .build()
            .unwrap();

        let results = exec.execute(&query).unwrap();
        let mut ids: Vec<u64> = results.items.iter().map(|r| r.entity_id.as_u64()).collect();
        ids.sort_unstable();
        assert_eq!(
            ids,
            vec![1],
            "with no universe wired, min_signal(view,3) must keep item 1 and drop item 2 via the \
             post-filter; before the fix the empty-universe intersection dropped BOTH"
        );
    }

    /// Regression: the user-context suppression stage shared by RETRIEVE and
    /// SEARCH (`super::apply_user_context_suppression`) must (a) exclude seen
    /// items, (b) retain only saved items when a `Saved` filter is present, and
    /// (c) treat a `> u32::MAX` id correctly under the u32 conversion — RETAIN it
    /// against an exclusion bitmap (it can never be a member, and the old raw
    /// `id as u32` truncation would have aliased it onto a low suppressed id and
    /// dropped the wrong item).
    #[test]
    fn shared_suppression_excludes_seen_and_keeps_overflow_ids() {
        use crate::entities::{HardNegIndex, UserStateIndex};

        let user_state = UserStateIndex::new();
        // User 42 has seen item 1.
        user_state.mark_seen(42, 1);
        let hard_neg = HardNegIndex::new();

        // Candidate 1 is seen (must drop). Candidate 2 is unseen (must keep).
        // Candidate `u64::from(u32::MAX) + 1` overflows u32: it can never be the
        // seen bitmap's member, so it MUST be retained — a raw truncation would
        // alias it onto slot 0 (== item 1, which IS seen) and wrongly drop it.
        let overflow_id = u64::from(u32::MAX) + 1;
        let mut candidates = vec![
            EntityId::new(1),
            EntityId::new(2),
            EntityId::new(overflow_id),
        ];

        crate::query::executor::apply_user_context_suppression(
            &mut candidates,
            42,
            Some(&user_state),
            None,
            Some(&hard_neg),
            None,
        );

        let mut ids: Vec<u64> = candidates.iter().map(|id| id.as_u64()).collect();
        ids.sort_unstable();
        assert_eq!(
            ids,
            vec![2, overflow_id],
            "seen item 1 dropped; unseen item 2 and the overflow id (un-aliasable, so retained) kept"
        );
    }

    /// Regression: a `Saved` inclusion filter routed through the shared
    /// suppression helper keeps only saved items, and an overflow id is DROPPED
    /// against the inclusion bitmap (it cannot satisfy membership).
    #[test]
    fn shared_suppression_saved_inclusion_drops_unsaved_and_overflow() {
        use crate::entities::UserStateIndex;

        let user_state = UserStateIndex::new();
        user_state.add_save(42, 2); // only item 2 is saved

        let overflow_id = u64::from(u32::MAX) + 1;
        let mut candidates = vec![
            EntityId::new(1),
            EntityId::new(2),
            EntityId::new(overflow_id),
        ];

        crate::query::executor::apply_user_context_suppression(
            &mut candidates,
            42,
            Some(&user_state),
            None,
            None,
            Some(&FilterExpr::Saved(42)),
        );

        assert_eq!(
            candidates,
            vec![EntityId::new(2)],
            "Saved(42) keeps only item 2; item 1 (unsaved) and the overflow id (cannot be an \
             inclusion-bitmap member) are both dropped"
        );
    }

    /// C-CRITICAL regression (Stage 2 inclusion intersection): a `SignalRanked`
    /// candidate whose id is `(1<<32) + k` must be DROPPED by an index-backed
    /// `CategoryEq` filter even when slot `k` IS present in the match bitmap. A raw
    /// `id as u32` truncation would alias the overflow id onto slot `k` and wrongly
    /// RETAIN it; the checked `entity_as_u32` (drop-on-overflow) prevents the leak.
    #[test]
    fn stage2_index_filter_drops_overflow_alias_candidate() {
        use roaring::RoaringBitmap;

        use crate::storage::indexes::bitmap::BitmapIndex;

        let schema = view_schema();
        let ledger = SignalLedger::new(schema, Box::new(NoopWalWriter));
        let profile_reg = signal_ranked_registry();

        // Real low id `5` is in category "jazz"; the overflow id `(1<<32)+5`
        // aliases onto slot 5 under truncation but must NOT be retained.
        let k: u32 = 5;
        let overflow_id = (1u64 << 32) + u64::from(k);

        let now = Timestamp::now();
        ledger
            .record_signal("view", EntityId::new(u64::from(k)), 1.0, now)
            .unwrap();
        ledger
            .record_signal("view", EntityId::new(overflow_id), 1.0, now)
            .unwrap();

        // Category index: slot 5 ∈ "jazz". The overflow item is NOT indexed under
        // jazz; only its truncated alias (slot 5) would falsely match.
        let category = BitmapIndex::new("category");
        category.insert(k, "jazz");

        // Universe must be Some so Stage 2 runs the index intersection. It only
        // gates entry to the block; CategoryEq resolves to the category bitmap.
        let mut universe_bm = RoaringBitmap::new();
        universe_bm.insert(k);
        let universe = std::sync::RwLock::new(universe_bm);

        let exec = RetrieveExecutor::new(
            &ledger,
            &profile_reg,
            Some(&category), // category
            None,            // format
            None,            // creator
            None,            // tag
            None,            // duration
            None,            // created_at
            Some(&universe), // universe
        );
        let query = Retrieve::builder()
            .profile("view_ranked")
            .filter(FilterExpr::CategoryEq("jazz".into()))
            .limit(10)
            .build()
            .unwrap();

        let results = exec.execute(&query).unwrap();
        let ids: Vec<u64> = results.items.iter().map(|r| r.entity_id.as_u64()).collect();
        assert_eq!(
            ids,
            vec![u64::from(k)],
            "only the genuine low id 5 may survive CategoryEq(jazz); the overflow id \
             {overflow_id} must be dropped, not aliased onto slot {k} and retained"
        );
    }
}