triplets_core/sampler/

mod.rs

mod backends;

use crate::chunking::{ChunkingAlgorithm, SlidingWindowChunker};
use chrono::Duration;
use indexmap::IndexMap;
use nohash_hasher::IntSet;
use rand::prelude::*;
use rayon::prelude::*;
use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

use crate::config::{
    ChunkingStrategy, NegativeStrategy, SamplerConfig, Selector, TextRecipe, TripletRecipe,
};
use crate::constants::sampler::AUTO_INJECTED_LONG_SECTION_CHUNK_PAIR_RECIPE_NAME;
use crate::constants::sampler::{
    ANCHOR_POSITIVE_SWAP_MASK, EPOCH_SEED_OFFSET, EXHAUSTION_RETRY_LIMIT, NEG_REASON_SAME_RECORD,
    NEG_REASON_WRONG_ARTICLE, NEG_REASON_WRONG_DATE, NEG_REASON_WRONG_QA, PREFETCHER_SOURCE_ID,
    PREFETCHER_STOPPED_REASON, RECIPE_LABEL_TEXT, RECIPE_LABEL_TRIPLETS,
    RECIPE_ORDER_MAX_WEIGHT_MULTIPLIER, ROLE_LABEL_ANCHOR, ROLE_LABEL_CONTEXT,
    SAME_SELECTOR_PAIR_RETRY_LIMIT,
};
use crate::data::{
    ChunkView, DataRecord, PairLabel, RecordChunk, RecordSection, SampleBatch, SamplePair,
    SampleTriplet, SectionRole, TextBatch, TextSample, TripletBatch,
};
use crate::epoch::EpochTracker;
use crate::errors::SamplerError;
use crate::hash::{derive_epoch_seed, stable_hash_key, stable_hash_key_section, stable_hash_str};
use crate::ingestion::IngestionManager;
use crate::metadata::{META_FIELD_DATE, MetadataKey};
use crate::metrics::{chunk_proximity_score, window_index_proximity};
use crate::source::DataSource;
use crate::splits::{
    EpochStateStore, PersistedSamplerState, SamplerStateStore, SplitLabel, SplitStore,
};
use crate::tokenizer::{Tokenizer, WhitespaceTokenizer};
use crate::types::{RecipeKey, RecordKey, SourceId};
use crate::utils::platform_newline;

// AUTO-RECIPE HANDLING OVERVIEW (end-to-end):
// Stage A: Source-level injection eligibility ("should this source even get the recipe?")
//   - `sync_records_from_cache` -> `record_has_long_anchor_or_context_section` marks
//     `long_section_counts`.
//   - `resolve_source_triplet_plan` -> `should_auto_inject_chunk_pair_recipe` appends
//     `source_chunk_pair_recipe` for eligible sources.
// Stage B: Record-level execution eligibility ("can this specific record run it now?")
//   - `make_auto_chunk_pair_triplet_with_anchor` checks
//     `record_has_at_least_two_window_chunks_for_selector` before sampling.
// Stage C: Actual chunk window layout ("how are windows formed?")
//   - `materialize_chunks` is the source of truth for token window construction.

#[derive(Debug, Clone)]
/// Small deterministic RNG used for reproducible sampler behavior.
struct DeterministicRng {
    state: u64,
}

impl DeterministicRng {
    fn new(seed: u64) -> Self {
        Self { state: seed }
    }

    fn from_state(state: u64) -> Self {
        Self { state }
    }

    fn state(&self) -> u64 {
        self.state
    }

    fn next_u64_internal(&mut self) -> u64 {
        let mut z = self.state.wrapping_add(0x9E3779B97F4A7C15);
        self.state = z;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
        z ^ (z >> 31)
    }
}

impl rand::RngCore for DeterministicRng {
    fn next_u32(&mut self) -> u32 {
        self.next_u64_internal() as u32
    }

    fn next_u64(&mut self) -> u64 {
        self.next_u64_internal()
    }

    fn fill_bytes(&mut self, dest: &mut [u8]) {
        let mut offset = 0;
        while offset < dest.len() {
            let value = self.next_u64_internal();
            let bytes = value.to_le_bytes();
            let remaining = dest.len() - offset;
            let copy_len = remaining.min(bytes.len());
            dest[offset..offset + copy_len].copy_from_slice(&bytes[..copy_len]);
            offset += copy_len;
        }
    }
}

/// Public helper so callers (and examples) can compute the same chunk weighting the sampler uses.
///
/// Definitions:
/// - `trust` (range 0.0-1.0): comes from `RecordChunk.quality.trust`; scales the contribution of a chunk.
/// - `base`: window chunks use proximity-to-head (`1 / (index + 1)`); summary chunks use their configured `summary_fallback_weight`.
/// - `chunk_weight_floor`: minimum weight applied after scaling.
///
/// Formula: `max(chunk_weight_floor, base * trust)`.
pub fn chunk_weight(strategy: &ChunkingStrategy, chunk: &RecordChunk) -> f32 {
    let floor = strategy.chunk_weight_floor;
    let trust = chunk.quality.trust.clamp(0.0, 1.0);
    let base = match &chunk.view {
        ChunkView::Window { index, .. } => window_index_proximity(*index),
        ChunkView::SummaryFallback { weight, .. } => *weight,
    };
    (base * trust).max(floor)
}

/// Public sampling interface for pair, triplet, and text batch generation.
pub trait Sampler {
    /// Returns a batch of pairs. Consumes the shared epoch cursor for anchor selection.
    ///
    /// Samples all sources uniformly. Prefer [`Sampler::next_pair_batch_with_weights`]
    /// with an explicit per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call next_pair_batch_with_weights(split, &weights) with an explicit per-source weight map to honor a data mixture"
    )]
    fn next_pair_batch(&self, split: SplitLabel) -> Result<SampleBatch, SamplerError> {
        self.next_pair_batch_with_weights(split, &HashMap::new())
    }
    /// Returns a batch of text samples. Consumes the shared epoch cursor for anchor selection.
    ///
    /// Samples all sources uniformly. Prefer [`Sampler::next_text_batch_with_weights`]
    /// with an explicit per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call next_text_batch_with_weights(split, &weights) with an explicit per-source weight map to honor a data mixture"
    )]
    fn next_text_batch(&self, split: SplitLabel) -> Result<TextBatch, SamplerError> {
        self.next_text_batch_with_weights(split, &HashMap::new())
    }
    /// Returns a batch of triplets. Consumes the shared epoch cursor for anchor selection.
    ///
    /// Samples all sources uniformly. Prefer [`Sampler::next_triplet_batch_with_weights`]
    /// with an explicit per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call next_triplet_batch_with_weights(split, &weights) with an explicit per-source weight map to honor a data mixture"
    )]
    fn next_triplet_batch(&self, split: SplitLabel) -> Result<TripletBatch, SamplerError> {
        self.next_triplet_batch_with_weights(split, &HashMap::new())
    }
    /// Returns a batch of pairs with per-call source weights.
    fn next_pair_batch_with_weights(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<SampleBatch, SamplerError>;
    /// Returns a batch of text samples with per-call source weights.
    fn next_text_batch_with_weights(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<TextBatch, SamplerError>;
    /// Returns a batch of triplets with per-call source weights.
    fn next_triplet_batch_with_weights(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<TripletBatch, SamplerError>;
    /// Persist sampler and split runtime state for restart-resume.
    ///
    /// Default is a no-op, allowing mock/test implementations to skip persistence.
    fn save_sampler_state(&self, _save_to: Option<&std::path::Path>) -> Result<(), SamplerError> {
        Ok(())
    }
}

/// Background prefetcher that fills a bounded queue with sample batches.
pub struct BatchPrefetcher<T> {
    receiver: Option<mpsc::Receiver<Result<T, SamplerError>>>,
    handle: Option<thread::JoinHandle<()>>,
    stats: Arc<PrefetcherStats>,
}

#[derive(Default)]
/// Prefetcher runtime counters.
struct PrefetcherStats {
    queued: AtomicUsize,
    produced: AtomicUsize,
    errors: AtomicUsize,
}

impl<T: Send + 'static> BatchPrefetcher<T> {
    fn new<F>(capacity: usize, mut producer: F) -> Self
    where
        F: FnMut() -> Result<T, SamplerError> + Send + 'static,
    {
        let (sender, receiver) = mpsc::sync_channel(capacity.max(1));
        let stats = Arc::new(PrefetcherStats::default());
        let stats_thread = Arc::clone(&stats);
        let handle = thread::spawn(move || {
            loop {
                let result = producer();
                if result.is_err() {
                    stats_thread.errors.fetch_add(1, Ordering::Relaxed);
                }
                if sender.send(result).is_err() {
                    return;
                }
                stats_thread.queued.fetch_add(1, Ordering::Relaxed);
                stats_thread.produced.fetch_add(1, Ordering::Relaxed);
            }
        });
        Self {
            receiver: Some(receiver),
            handle: Some(handle),
            stats,
        }
    }

    /// Block until the next prefetched batch is available.
    pub fn next(&self) -> Result<T, SamplerError> {
        let receiver = self
            .receiver
            .as_ref()
            .ok_or_else(|| SamplerError::SourceUnavailable {
                source_id: PREFETCHER_SOURCE_ID.into(),
                reason: PREFETCHER_STOPPED_REASON.into(),
            })?;
        let result = receiver.recv().unwrap_or_else(|_| {
            Err(SamplerError::SourceUnavailable {
                source_id: PREFETCHER_SOURCE_ID.into(),
                reason: PREFETCHER_STOPPED_REASON.into(),
            })
        });
        self.stats
            .queued
            .try_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
                Some(value.saturating_sub(1))
            })
            .ok();
        result
    }

    /// Number of prefetched batches currently queued.
    pub fn queue_len(&self) -> usize {
        self.stats.queued.load(Ordering::Relaxed)
    }

    /// Total number of batches produced by the background worker.
    pub fn produced_count(&self) -> usize {
        self.stats.produced.load(Ordering::Relaxed)
    }

    /// Total number of errors produced by the background worker.
    pub fn error_count(&self) -> usize {
        self.stats.errors.load(Ordering::Relaxed)
    }
}

impl<T> Drop for BatchPrefetcher<T> {
    fn drop(&mut self) {
        self.receiver.take();
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

/// Sampler that draws anchors from a single shared epoch cursor and then
/// selects chunks from those records. Ingestion happens on demand when sampling.
pub struct TripletSampler<S: SplitStore + EpochStateStore + SamplerStateStore + 'static> {
    inner: Mutex<TripletSamplerInner<S>>,
}

/// Internal sampler state implementation guarded by `TripletSampler`.
struct TripletSamplerInner<S: SplitStore + EpochStateStore + SamplerStateStore + 'static> {
    /// Immutable sampler configuration (seed, batch size, recipes, splits, etc.).
    config: SamplerConfig,
    /// Active chunking implementation used to materialize section chunks.
    chunker: Arc<dyn ChunkingAlgorithm>,
    /// Split store backing train/val/test assignments and persisted sampler state.
    split_store: Arc<S>,
    /// On-demand ingestion manager that fills the batch-sized buffer.
    ingestion: IngestionManager,
    /// Current in-memory record pool keyed by composite record identity.
    /// Composite keys (not bare ids) so overlapping ids from different
    /// sources coexist instead of overwriting each other.
    records: IndexMap<RecordKey, Arc<DataRecord>>,
    /// Deficit counters for weighted source selection (smooth weighted
    /// round-robin over `source_order`). Only engaged when the caller passes
    /// explicit non-uniform per-source weights; uniform/empty maps take the
    /// legacy uniform-cycling path bit-identically. New sources start at 0.0
    /// via entry-or-insert (no sync hook needed); stale entries are pruned
    /// with the source index. Intentionally NOT persisted: deficits
    /// self-correct within a few draws, and no consumer persists sampler
    /// state across processes with weights engaged.
    source_deficits: HashMap<SourceId, f32>,
    /// Deterministic RNG for per-batch shuffles and sampling.
    rng: DeterministicRng,
    /// Config-level triplet recipes used when sources do not supply their own.
    triplet_recipes: Vec<TripletRecipe>,
    /// Config-level text recipes used when sources do not supply their own.
    text_recipes: Vec<TextRecipe>,
    /// Per-source triplet recipes keyed by source id.
    source_triplet_recipes: HashMap<SourceId, Vec<TripletRecipe>>,
    /// Per-source count of pooled records containing at least one section
    /// larger than the chunk window. A source is eligible for the
    /// auto-injected chunk-pair recipe while its count is greater than zero.
    /// Maintained incrementally on record insert/evict (strict O(Δ)) instead
    /// of rescanning the pool on every sync.
    long_section_counts: HashMap<SourceId, usize>,
    /// Per-source text recipes keyed by source id.
    source_text_recipes: HashMap<SourceId, Vec<TextRecipe>>,
    /// True if triplet recipes came from config (no source defaults).
    using_config_triplet_recipes: bool,
    /// True if text recipes came from config (no derivation).
    using_config_text_recipes: bool,
    /// Last observed ingestion counter for cache updates.
    last_observed_ingest: u64,
    /// Epoch tracker for split-aware deterministic sampling.
    epoch_tracker: EpochTracker,
    /// Per-record, per-section chunk cursor to rotate through chunk windows.
    chunk_cursors: HashMap<(RecordKey, usize), usize>,
    /// Per-record, per-role cursor to rotate through role-specific sections.
    role_cursors: HashMap<(RecordKey, String), usize>,
    /// Pluggable negative-selection backend (uniform-random or BM25).
    negative_backend: Box<dyn backends::NegativeBackend>,
    /// Chunk key to parent record key lookup (used by epoch tracker).
    chunk_index: HashMap<RecordKey, RecordKey>,
    /// Round-robin order of source ids (deterministic).
    source_order: Vec<SourceId>,
    /// Current index into `source_order` for shuffled-cycle sampling.
    source_cycle_idx: usize,
    /// True once persisted source state is loaded.
    source_state_loaded: bool,
    /// True once persisted ingestion cursors are loaded.
    ingestion_cursors_loaded: bool,
    /// Marks whether source state needs persistence.
    source_state_dirty: bool,
    /// Record keys per source for round-robin within a source.
    source_record_indices: HashMap<SourceId, Vec<RecordKey>>,
    /// Per-source cursor into `source_record_indices`.
    source_record_cursors: HashMap<SourceId, usize>,
    /// Round-robin index for triplet recipe cycling.
    triplet_recipe_rr_idx: usize,
    /// Hashes of text already emitted by the text batch path while the record
    /// pool is stable.  Pruned on sync_records_from_cache only when records
    /// are evicted.  Prevents the same text from being sampled again within
    /// the same ingestion window, regardless of source wrapping or epoch.
    ///
    /// Per-record tracking lets us surgically remove hashes only for evicted
    /// records, preserving cross-batch dedup across small window advances
    /// (e.g. window=512, batch_size=4 where only 4/512 records are swapped).
    emitted_text_hashes: HashMap<RecordKey, HashSet<u64>>,
    /// Round-robin index for text recipe cycling.
    text_recipe_rr_idx: usize,
    /// Epoch counter for per-source deterministic shuffling (seed ^ epoch).
    epoch: u64,

    /// Tracks whether each source has wrapped its cursor in the current epoch.
    source_wrapped: HashMap<SourceId, bool>,

    /// Last sync version watermark for incremental delta synchronization.
    last_sync_version: u64,

    /// Monotonic counter advanced only when record-pool membership actually
    /// changes (genuinely new IDs added or IDs truly evicted). Pure version
    /// refreshes of already-pooled IDs do not advance it, so the cached
    /// `records_by_split` map below stays valid across steady-state advances.
    pool_generation: u64,

    /// Cached `records_by_split` output, valid while
    /// `cached_records_by_split_gen == pool_generation`. Avoids rebuilding the
    /// full per-split map (O(N) clones + label lookups) on every batch.
    cached_records_by_split: Option<HashMap<SplitLabel, Vec<RecordKey>>>,
    /// Pool generation the cached map was built from.
    cached_records_by_split_gen: u64,

    /// Last observed force-refresh generation. When the ingestion manager
    /// reports a newer force-refresh generation, the next sync falls back to
    /// a full pool resync (exact full-snapshot semantics) instead of the
    /// O(Δ) delta path, because a force refresh replaces the cache contents
    /// wholesale. Steady-state advances keep the delta fast path.
    last_force_refresh_generation: u64,

    /// Persistent cache of split labels keyed by composite record identity.
    /// Eliminates redundant DataStore::read() calls for split label lookups.
    split_labels: HashMap<RecordKey, SplitLabel>,
}

impl<S: SplitStore + EpochStateStore + SamplerStateStore + 'static> TripletSamplerInner<S> {
    fn new(config: SamplerConfig, split_store: Arc<S>) -> Self {
        Self::new_with_chunker(config, split_store, Arc::new(SlidingWindowChunker))
    }

    fn new_with_chunker(
        config: SamplerConfig,
        split_store: Arc<S>,
        chunker: Arc<dyn ChunkingAlgorithm>,
    ) -> Self {
        let buffer_size = config.ingestion_max_records.max(config.batch_size).max(2);
        let using_config_triplet_recipes = !config.recipes.is_empty();
        let using_config_text_recipes = !config.text_recipes.is_empty();
        let triplet_recipes = if using_config_triplet_recipes {
            config.recipes.clone()
        } else {
            Vec::new()
        };
        let text_recipes = if using_config_text_recipes {
            config.text_recipes.clone()
        } else if !triplet_recipes.is_empty() {
            Self::build_derived_text_recipes(&triplet_recipes)
        } else {
            Vec::new()
        };
        let ingestion = IngestionManager::new(buffer_size, config.clone());
        let epoch_backend = Some(Arc::clone(&split_store) as Arc<dyn EpochStateStore>);
        let epoch_tracker = EpochTracker::new(
            true,
            epoch_backend,
            derive_epoch_seed(config.seed, EPOCH_SEED_OFFSET),
        );
        let mut sampler = Self {
            rng: DeterministicRng::new(config.seed),
            config,
            chunker,
            split_store,
            ingestion,
            records: IndexMap::new(),
            triplet_recipes,
            text_recipes,
            source_triplet_recipes: HashMap::new(),
            long_section_counts: HashMap::new(),
            source_text_recipes: HashMap::new(),
            using_config_triplet_recipes,
            using_config_text_recipes,
            last_observed_ingest: 0,
            epoch_tracker,
            chunk_cursors: HashMap::new(),
            role_cursors: HashMap::new(),
            negative_backend: {
                #[cfg(feature = "bm25-mining")]
                {
                    Box::new(backends::Bm25Backend::new())
                }
                #[cfg(not(feature = "bm25-mining"))]
                {
                    Box::new(backends::DefaultBackend)
                }
            },
            chunk_index: HashMap::new(),
            source_order: Vec::new(),
            source_cycle_idx: 0,
            source_deficits: HashMap::new(),
            source_state_loaded: false,
            ingestion_cursors_loaded: false,
            source_state_dirty: false,
            source_record_indices: HashMap::new(),
            source_record_cursors: HashMap::new(),
            emitted_text_hashes: HashMap::new(),
            triplet_recipe_rr_idx: 0,
            text_recipe_rr_idx: 0,
            epoch: 0,
            source_wrapped: HashMap::new(),
            last_sync_version: 0,
            last_force_refresh_generation: 0,
            pool_generation: 0,
            cached_records_by_split: None,
            cached_records_by_split_gen: 0,
            split_labels: HashMap::new(),
        };
        if !sampler.using_config_text_recipes {
            sampler.rebuild_derived_text_recipes();
        }
        sampler
    }

    fn text_recipes(&self) -> &[TextRecipe] {
        &self.text_recipes
    }

    /// Current epoch-adjusted seed: mixes `epoch` into `config.seed` so every epoch
    /// produces a distinct permutation across all seed-dependent operations.
    fn epoch_seed(&self) -> u64 {
        derive_epoch_seed(self.config.seed, self.epoch)
    }

    /// Get or compute the split label for a record, using the persistent cache.
    fn get_or_insert_split_label(&mut self, key: &RecordKey) -> Result<SplitLabel, SamplerError> {
        if let Some(&label) = self.split_labels.get(key) {
            return Ok(label);
        }
        let label = match self.split_store.label_for(key) {
            Some(label) => label,
            None => self.split_store.ensure(key.clone())?,
        };
        self.split_labels.insert(key.clone(), label);
        Ok(label)
    }

    fn register_source(
        &mut self,
        source: Box<dyn DataSource + 'static>,
    ) -> Result<(), SamplerError> {
        let source_id = source.id().to_string();
        if !self.using_config_triplet_recipes {
            let triplets = source.default_triplet_recipes();
            if !triplets.is_empty() {
                self.source_triplet_recipes
                    .insert(source_id.clone(), triplets.clone());
                if !self.using_config_text_recipes {
                    let derived = Self::build_derived_text_recipes(&triplets);
                    self.source_text_recipes
                        .insert(source_id.clone(), derived.clone());
                    self.extend_text_recipes_unique(&derived);
                }
            }
        }
        self.ingestion.register_source(source)?;
        Ok(())
    }

    fn set_epoch(&mut self, epoch: u64) -> Result<(), SamplerError> {
        self.epoch_tracker.ensure_loaded()?;
        self.epoch_tracker.force_epoch(epoch);
        self.epoch = epoch;
        self.ingestion.set_epoch(epoch);
        self.ingestion.reset_stream_cursors();
        // Reset epoch step at epoch boundary so each epoch
        // starts with step=0, giving deterministic step sequences.
        self.ingestion.reset_epoch_step();
        self.source_record_cursors.clear();
        self.source_cycle_idx = 0;
        for source in &self.source_order {
            self.source_wrapped.insert(source.clone(), false);
        }
        self.rebuild_source_index()?;
        self.source_state_dirty = self.source_order.len() > 1;
        Ok(())
    }

    fn next_chunk_from_pool(
        &mut self,
        key: &RecordKey,
        section_idx: usize,
        pool: Vec<RecordChunk>,
    ) -> Option<RecordChunk> {
        if pool.is_empty() {
            return None;
        }
        let map_key = (key.clone(), section_idx);
        if !self.chunk_cursors.contains_key(&map_key) {
            // First touch should **not** always start at window 0.
            // We derive a deterministic per-(record,section) starting offset so:
            // - repeated runs with the same seed/epoch are reproducible,
            // - first sampled window is spread across records/sections,
            // - subsequent calls still rotate cyclically through the pool.
            let cursor_key = stable_hash_key_section(self.epoch_seed(), key, section_idx);
            let start = (cursor_key as usize) % pool.len();
            self.chunk_cursors.insert(map_key.clone(), start);
        }
        let cursor = self.chunk_cursors.entry(map_key).or_insert(0);
        if *cursor >= pool.len() {
            *cursor = 0;
        }
        let chunk = pool.get(*cursor).cloned();
        *cursor = (*cursor + 1) % pool.len();
        chunk
    }

    fn prune_cursor_state(&mut self) {
        if self.chunk_cursors.is_empty()
            && self.role_cursors.is_empty()
            && self.negative_backend.cursors_empty()
        {
            return;
        }
        let valid_keys: HashSet<RecordKey> = self.records.keys().cloned().collect();
        self.chunk_cursors
            .retain(|(record_key, _), _| valid_keys.contains(record_key));
        self.role_cursors
            .retain(|(record_key, _), _| valid_keys.contains(record_key));
        self.negative_backend.prune_cursors(&valid_keys);
    }

    /// Ordered removal of a record id from one source's index vector.
    /// Empty source entries are left in place; the caller prunes them after
    /// all evictions and additions are processed (mirroring
    /// `rebuild_source_index`, which only contains sources with at least one
    /// allowed-split record).
    fn remove_id_from_source_index(&mut self, source: &SourceId, record_key: &RecordKey) {
        if let Some(ids) = self.source_record_indices.get_mut(source)
            && let Some(pos) = ids.iter().position(|key| key == record_key)
        {
            ids.remove(pos);
        }
    }

    /// Insert a record key into one source's index vector, preserving the
    /// deterministic `stable_hash_key(epoch_seed, key)` sort order via binary
    /// search. Records whose split is not allowed are not indexed, mirroring
    /// `rebuild_source_index`. Creates the source entry when needed; the
    /// caller keeps `source_order` sorted afterwards.
    fn insert_id_into_source_index(
        &mut self,
        source: SourceId,
        record_key: RecordKey,
        label: SplitLabel,
        allowed: &HashSet<SplitLabel>,
    ) {
        if !allowed.contains(&label) {
            return;
        }
        let seed = self.epoch_seed();
        let key = stable_hash_key(seed, &record_key);
        // Upper-bound insertion: with distinct hashes this is the unique
        // sorted position; on a hash collision it places the key after
        // existing equals, matching stable-sort semantics.
        let pos = match self.source_record_indices.get(&source) {
            Some(ids) => ids.partition_point(|k| stable_hash_key(seed, k) <= key),
            None => 0,
        };
        self.source_record_indices
            .entry(source)
            .or_default()
            .insert(pos, record_key);
    }

    /// Increment the long-section record count for a source.
    fn increment_long_section_count(&mut self, source: &SourceId) {
        *self.long_section_counts.entry(source.clone()).or_default() += 1;
    }

    /// Decrement the long-section record count for a source, removing the
    /// entry when no counted records remain.
    fn decrement_long_section_count(&mut self, source: &SourceId) {
        let remove = if let Some(count) = self.long_section_counts.get_mut(source) {
            *count = count.saturating_sub(1);
            *count == 0
        } else {
            false
        };
        if remove {
            self.long_section_counts.remove(source);
        }
    }

    fn rebuild_chunk_index(&mut self) {
        self.chunk_index.clear();
        for record in self.records.values() {
            let key = RecordKey::from(record);
            self.chunk_index.insert(key.clone(), key);
        }
    }

    fn rebuild_source_index(&mut self) -> Result<(), SamplerError> {
        self.source_record_indices.clear();
        let allowed = self.allowed_target_splits();
        let allowed_set: HashSet<SplitLabel> = allowed.into_iter().collect();
        // Collect record keys first to avoid borrowing issues
        let record_keys: Vec<RecordKey> = self.records.values().map(RecordKey::of_arc).collect();
        for record_key in record_keys {
            let label = self.get_or_insert_split_label(&record_key)?;
            if !allowed_set.contains(&label) {
                continue;
            }
            self.source_record_indices
                .entry(record_key.source.clone())
                .or_default()
                .push(record_key);
        }

        let shuffle_seed = self.epoch_seed();
        for record_keys in self.source_record_indices.values_mut() {
            record_keys.sort_by_key(|key| stable_hash_key(shuffle_seed, key));
        }

        self.source_order = self.source_record_indices.keys().cloned().collect();
        self.source_order.sort();
        self.refresh_source_wrapped();
        // Drop deficit counters for sources that left the pool.
        self.source_deficits
            .retain(|s, _| self.source_record_indices.contains_key(s));

        self.source_record_cursors
            .retain(|source, _| self.source_record_indices.contains_key(source));
        if self.source_state_loaded {
            if self.source_order.is_empty() {
                self.source_cycle_idx = 0;
            }
            self.source_state_dirty = self.source_order.len() > 1;
        }
        Ok(())
    }

    fn refresh_source_wrapped(&mut self) {
        self.source_wrapped.clear();
        for source in &self.source_order {
            let len = self
                .source_record_indices
                .get(source)
                .map(|items| items.len())
                .unwrap_or(0);
            if len == 0 {
                self.source_wrapped.insert(source.clone(), false);
                continue;
            }
            let cursor = self.source_record_cursors.get(source).copied().unwrap_or(0);
            let wrapped = cursor > 0 && cursor % len == 0;
            self.source_wrapped.insert(source.clone(), wrapped);
        }
    }

    fn shuffled_source_cycle(&self, cycle: u64) -> Vec<SourceId> {
        let mut sources = self.source_order.clone();
        let seed = self.epoch_seed() ^ cycle;
        sources.sort_by_key(|source| stable_hash_str(seed, source));
        sources
    }

    /// True when `weights` carries no selective signal: empty or all equal.
    /// Uniform maps take the legacy uniform-cycling path bit-identically.
    fn weights_are_selective(weights: &HashMap<SourceId, f32>) -> bool {
        if weights.is_empty() {
            return false;
        }
        let first = *weights.values().next().unwrap();
        weights.values().any(|&w| w != first)
    }

    /// Effective quantum for `source`: its weight when finite and
    /// non-negative, else 0. Unlisted sources default to 1.0 (mirrors the
    /// ingestion drain).
    fn weight_quantum(weights: &HashMap<SourceId, f32>, source: &str) -> f32 {
        weights.get(source).copied().unwrap_or(1.0).max(0.0)
    }

    /// Weighted source choice: smooth weighted round-robin over `candidates`.
    /// Credits each eligible source its weight quantum, picks the highest
    /// deficit, deducts the round total. Deterministic given pool + deficits
    /// (no RNG): same state, same choice. New sources start at 0.0 via
    /// entry-or-insert. Returns `None` when no candidate has records.
    fn select_weighted_source(
        &mut self,
        candidates: &[SourceId],
        weights: &HashMap<SourceId, f32>,
    ) -> Option<SourceId> {
        let eligible: Vec<&SourceId> = candidates
            .iter()
            .filter(|s| {
                self.source_record_indices
                    .get(*s)
                    .map(|v| !v.is_empty())
                    .unwrap_or(false)
            })
            .collect();
        if eligible.is_empty() {
            return None;
        }
        let mut total = 0.0f32;
        for source in &eligible {
            let q = Self::weight_quantum(weights, source);
            total += q;
            *self.source_deficits.entry((*source).clone()).or_insert(0.0) += q;
        }
        if total <= 0.0 {
            return None;
        }
        let mut best: Option<(f32, SourceId)> = None;
        for source in &eligible {
            let d = self.source_deficits.get(*source).copied().unwrap_or(0.0);
            if best.as_ref().map(|(bd, _)| d > *bd).unwrap_or(true) {
                best = Some((d, (*source).clone()));
            }
        }
        let (_, source) = best?;
        *self
            .source_deficits
            .get_mut(&source)
            .expect("deficit credited above") -= total;
        Some(source)
    }

    fn ensure_source_state(&mut self) -> Result<(), SamplerError> {
        if self.source_state_loaded {
            return Ok(());
        }
        let persisted = self.split_store.load_sampler_state()?;
        self.source_cycle_idx = persisted
            .as_ref()
            .map(|state| state.source_cycle_idx as usize)
            .unwrap_or(0);
        if let Some(state) = persisted {
            for (source, cursor) in state.source_record_cursors {
                if self.source_record_indices.contains_key(&source) {
                    self.source_record_cursors.insert(source, cursor as usize);
                }
            }
            self.epoch = state.epoch;
            self.ingestion.set_epoch(state.epoch);
            self.rng = DeterministicRng::from_state(state.rng_state);
            self.triplet_recipe_rr_idx = state.triplet_recipe_rr_idx as usize;
            self.text_recipe_rr_idx = state.text_recipe_rr_idx as usize;
        }
        self.refresh_source_wrapped();
        self.source_state_loaded = true;
        self.source_state_dirty = true;
        Ok(())
    }

    fn persist_source_state(&mut self, save_to: Option<&Path>) -> Result<(), SamplerError> {
        if !self.source_state_loaded {
            return Ok(());
        }
        let state = PersistedSamplerState {
            source_cycle_idx: self.source_cycle_idx as u64,
            source_record_cursors: self
                .source_record_cursors
                .iter()
                .map(|(source, cursor)| (source.clone(), *cursor as u64))
                .collect(),
            epoch: self.epoch,
            epoch_step: self.ingestion.epoch_step(),
            rng_state: self.rng.state(),
            triplet_recipe_rr_idx: self.triplet_recipe_rr_idx as u64,
            text_recipe_rr_idx: self.text_recipe_rr_idx as u64,
            source_stream_cursors: self.ingestion.snapshot_cursors(),
        };
        self.split_store.save_sampler_state(&state, save_to)?;
        self.source_state_dirty = false;
        Ok(())
    }

    fn rebuild_derived_text_recipes(&mut self) {
        if self.using_config_text_recipes {
            return;
        }
        if self.triplet_recipes.is_empty() {
            self.text_recipes.clear();
        } else {
            self.text_recipes = Self::build_derived_text_recipes(&self.triplet_recipes);
        }
    }

    fn extend_text_recipes_unique(&mut self, recipes: &[TextRecipe]) {
        for recipe in recipes {
            if self
                .text_recipes
                .iter()
                .any(|existing| existing.name == recipe.name)
            {
                continue;
            }
            self.text_recipes.push(recipe.clone());
        }
    }

    fn configured_triplet_recipes_for_source<'a>(&'a self, source: &str) -> &'a [TripletRecipe] {
        if self.using_config_triplet_recipes {
            return &self.triplet_recipes;
        }
        self.source_triplet_recipes
            .get(source)
            .map(|recipes| recipes.as_slice())
            .unwrap_or(&[])
    }

    /// Returns true when `recipes` already contains the long-section auto recipe.
    fn contains_auto_chunk_pair_recipe(recipes: &[TripletRecipe]) -> bool {
        recipes
            .iter()
            .any(|recipe| recipe.name.as_ref() == AUTO_INJECTED_LONG_SECTION_CHUNK_PAIR_RECIPE_NAME)
    }

    fn source_supports_chunk_pair_recipe(&self, source: &str) -> bool {
        if self.config.chunking.max_window_tokens == 0 {
            return false;
        }
        self.long_section_counts
            .get(source)
            .is_some_and(|&count| count > 0)
    }

    /// Stage A (source-level): decide whether to append auto recipe for `source`.
    ///
    /// Decision criteria:
    /// - source must be eligible (`source_supports_chunk_pair_recipe`), and
    /// - configured pool must not already include the auto recipe name.
    fn should_auto_inject_chunk_pair_recipe(
        &self,
        source: &str,
        recipes: &[TripletRecipe],
    ) -> bool {
        self.source_supports_chunk_pair_recipe(source)
            && !Self::contains_auto_chunk_pair_recipe(recipes)
    }

    /// Build the auto-injected long-section recipe.
    ///
    /// Semantics:
    /// - Anchor selector: `Context`
    /// - Positive selector: `Context`
    /// - Negative selector: `Context` on a different record (`WrongArticle`)
    ///
    /// Anchor and positive are selected by two independent `select_chunk` calls
    /// from the same context chunk candidate pool for the chosen record.
    /// They are not concatenated and one is not derived from the other's text.
    fn source_chunk_pair_recipe() -> TripletRecipe {
        TripletRecipe {
            name: Cow::Borrowed(AUTO_INJECTED_LONG_SECTION_CHUNK_PAIR_RECIPE_NAME),
            anchor: Selector::Role(SectionRole::Context),
            positive_selector: Selector::Role(SectionRole::Context),
            negative_selector: Selector::Role(SectionRole::Context),
            negative_strategy: NegativeStrategy::WrongArticle,
            weight: 1.0,
            instruction: None,
            allow_same_anchor_positive: false,
        }
    }

    /// Resolve the source triplet plan.
    ///
    /// This is Stage A (source-level injection), not record-level execution.
    ///
    /// Algorithm for the auto-triplet injection path:
    /// 1) Start from the configured source recipe pool (`configured_triplet_recipes_for_source`).
    /// 2) Check auto-injection eligibility (`should_auto_inject_chunk_pair_recipe`), which is true only when:
    ///    - the source has long sections discovered during ingest sync, and
    ///    - the pool does not already include the auto recipe name.
    /// 3) If eligible, append `source_chunk_pair_recipe`.
    /// 4) Return both the effective recipe pool and whether step (3) happened.
    fn resolve_source_triplet_plan(&self, source: &str) -> (Vec<TripletRecipe>, bool) {
        let mut recipes = self.configured_triplet_recipes_for_source(source).to_vec();
        let mut auto_injected = false;
        if self.should_auto_inject_chunk_pair_recipe(source, &recipes) {
            recipes.push(Self::source_chunk_pair_recipe());
            auto_injected = true;
        }
        (recipes, auto_injected)
    }

    #[cfg(test)]
    fn triplet_recipes_for_source(&self, source: &str) -> Vec<TripletRecipe> {
        self.resolve_source_triplet_plan(source).0
    }

    fn triplet_recipe_count_for_source(&self, source: &str) -> usize {
        let (recipes, _auto_injected) = self.resolve_source_triplet_plan(source);
        recipes.len()
    }

    fn text_recipes_for_source<'a>(&'a self, source: &str) -> &'a [TextRecipe] {
        if self.using_config_text_recipes || self.using_config_triplet_recipes {
            return &self.text_recipes;
        }
        self.source_text_recipes
            .get(source)
            .map(|recipes| recipes.as_slice())
            .unwrap_or(&[])
    }

    /// Build a weighted, shuffled selection order from a slice of recipe weights.
    ///
    /// Each recipe with `weight > 0.0` receives a number of slots proportional to its weight
    /// relative to the smallest positive weight, capped at
    /// [`RECIPE_ORDER_MAX_WEIGHT_MULTIPLIER`].  The slot list is then shuffled so that
    /// deterministic round-robin cycling naturally draws recipes at the requested frequency:
    ///
    /// * `weight = [1.0, 1.0, 1.0]` → one slot each — identical to uniform round-robin.
    /// * `weight = [3.0, 1.0]`       → three slots for recipe 0, one for recipe 1.
    /// * `weight = [1.0, 0.0]`       → recipe 1 is excluded entirely (zero slots).
    fn recipe_order_weighted_shuffled(
        &mut self,
        weights: &[f32],
        rng: &mut DeterministicRng,
    ) -> Vec<usize> {
        weighted_recipe_order(weights, rng)
    }

    /// Weighted shuffle rotated by `rr_idx` for deterministic round-robin cycling.
    ///
    /// Calls [`Self::recipe_order_weighted_shuffled`] and rotates the result by
    /// `rr_idx % order.len()` so successive batches start from different recipes,
    /// preserving both the proportional frequency and the cross-batch cycling.
    fn recipe_order_weighted_cycled(
        &mut self,
        weights: &[f32],
        rr_idx: usize,
        rng: &mut DeterministicRng,
    ) -> Vec<usize> {
        let base = self.recipe_order_weighted_shuffled(weights, rng);
        if base.is_empty() {
            return base;
        }
        let start = rr_idx % base.len();
        let mut order = Vec::with_capacity(base.len());
        order.extend_from_slice(&base[start..]);
        order.extend_from_slice(&base[..start]);
        order
    }

    /// Weighted shuffled selection order for text recipes — same algorithm as
    /// [`Self::recipe_order_weighted_shuffled`], kept separate so the text and triplet
    /// round-robin counters can advance independently.
    fn text_recipe_order_weighted_shuffled(
        &mut self,
        weights: &[f32],
        rng: &mut DeterministicRng,
    ) -> Vec<usize> {
        weighted_recipe_order(weights, rng)
    }

    fn text_recipe_order_weighted_cycled(
        &mut self,
        weights: &[f32],
        rr_idx: usize,
        rng: &mut DeterministicRng,
    ) -> Vec<usize> {
        let base = self.text_recipe_order_weighted_shuffled(weights, rng);
        if base.is_empty() {
            return base;
        }
        let start = rr_idx % base.len();
        let mut order = Vec::with_capacity(base.len());
        order.extend_from_slice(&base[start..]);
        order.extend_from_slice(&base[..start]);
        order
    }

    fn allowed_target_splits(&self) -> Vec<SplitLabel> {
        self.config.allowed_splits.clone()
    }

    fn ensure_split_allowed(&self, split: SplitLabel) -> Result<(), SamplerError> {
        let allowed = self.allowed_target_splits();
        if allowed.contains(&split) {
            return Ok(());
        }
        Err(SamplerError::Configuration(format!(
            "requested split {:?} is not in allowed_splits {:?}",
            split, allowed
        )))
    }

    fn ensure_split_has_records(&mut self, target_split: SplitLabel) -> Result<(), SamplerError> {
        self.refresh_records_by_split_cache()?;
        let has_records = self
            .cached_records_by_split
            .as_ref()
            .and_then(|map| map.get(&target_split))
            .is_some_and(|records| !records.is_empty());
        if has_records {
            return Ok(());
        }
        Err(SamplerError::Exhausted(
            "no records available for target split".into(),
        ))
    }

    /// Rebuild the cached `records_by_split` map when the record pool changed
    /// since the last build. Steady-state batches hit the cache (O(1)) instead
    /// of rebuilding the full map (O(N) clones + label lookups).
    fn refresh_records_by_split_cache(&mut self) -> Result<(), SamplerError> {
        if self.cached_records_by_split_gen == self.pool_generation
            && self.cached_records_by_split.is_some()
        {
            return Ok(());
        }
        let mut map: HashMap<SplitLabel, Vec<RecordKey>> = HashMap::new();
        // Collect chunk_index entries first to avoid borrowing issues
        let entries: Vec<(RecordKey, RecordKey)> = self
            .chunk_index
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        for (chunk_key, record_key) in entries {
            if !self.records.contains_key(&record_key) {
                continue;
            }
            let label = self.get_or_insert_split_label(&record_key)?;
            map.entry(label).or_default().push(chunk_key);
        }
        self.cached_records_by_split = Some(map);
        self.cached_records_by_split_gen = self.pool_generation;
        Ok(())
    }

    /// Owned-map convenience used by tests that hand-mutate the pool.
    /// Production paths use [`Self::refresh_records_by_split_cache`] plus a
    /// direct reference to avoid the per-call clone.
    #[cfg(test)]
    fn records_by_split(&mut self) -> Result<HashMap<SplitLabel, Vec<RecordKey>>, SamplerError> {
        self.refresh_records_by_split_cache()?;
        Ok(self.cached_records_by_split.clone().unwrap_or_default())
    }

    fn choose_anchor_record(
        &mut self,
        source: Option<&str>,
        split: SplitLabel,
    ) -> Option<Arc<DataRecord>> {
        if let Some(source) = source {
            let record_ids = self.source_record_indices.get(source)?;
            if record_ids.is_empty() {
                return None;
            }
            let len = record_ids.len();
            let mut cursor = *self.source_record_cursors.get(source).unwrap_or(&0);
            let cycle = cursor / len;
            let offset_seed = self.epoch_seed() ^ (cycle as u64);
            let offset = (stable_hash_str(offset_seed, source) as usize) % len;
            let mut wrapped = false;
            let mut selected_key: Option<RecordKey> = None;

            for _ in 0..len {
                let pos = (cursor % len).saturating_add(offset) % len;
                let record_key = &record_ids[pos];
                cursor = cursor.saturating_add(1);
                if cursor.is_multiple_of(len) {
                    wrapped = true;
                }
                if self.split_labels.get(record_key) == Some(&split) {
                    selected_key = Some(record_key.clone());
                    break;
                }
            }

            self.source_record_cursors
                .insert(source.to_string(), cursor);
            if wrapped {
                self.mark_source_wrapped(source);
            }

            selected_key.and_then(|key| self.records.get(&key).map(Arc::clone))
        } else {
            while let Some(chunk_key) = self.epoch_tracker.next_record(split) {
                if let Some(record_key) = self.chunk_index.get(&chunk_key)
                    && let Some(record) = self.records.get(record_key)
                {
                    return Some(Arc::clone(record));
                }
            }
            None
        }
    }

    fn save_sampler_state(&mut self, save_to: Option<&Path>) -> Result<(), SamplerError> {
        if self.epoch_tracker.is_enabled() {
            self.epoch_tracker.persist()?;
        }
        self.persist_source_state(save_to)?;
        Ok(())
    }

    fn mark_source_wrapped(&mut self, source: &str) {
        self.source_wrapped.insert(source.to_string(), true);
        if self.source_order.is_empty() {
            return;
        }
        let all_wrapped = self
            .source_order
            .iter()
            .all(|name| self.source_wrapped.get(name).copied().unwrap_or(false));
        if all_wrapped {
            self.advance_epoch();
        }
    }

    fn advance_epoch(&mut self) {
        // The epoch advanced, changing the sampling permutation.
        // emitted_text_hashes is NOT cleared: the record pool is unchanged,
        // so texts already emitted are still off-limits.  The epoch
        // changes the shuffling permutation, not the dedup window.
        // Advance the epoch counter so source refreshes receive a different
        // permutation seed (seed ^ epoch), producing a fresh traversal order.
        self.epoch = self.epoch.saturating_add(1);
        self.ingestion.set_epoch(self.epoch);
        // Do NOT reset the epoch step here: it tracks the number of
        // refresh batches, which is orthogonal to the sampling epoch.
        // Resetting it would corrupt the persisted cursor value on resume.
        // Reset per-source record cursors so the new epoch starts from the
        // beginning of each source's permuted index space (not from where the
        // previous epoch left off).
        self.source_record_cursors.clear();
        self.source_cycle_idx = 0;
        // Reset the wrapped-tracking so all sources are eligible for sampling
        // in the new epoch.
        for source in &self.source_order {
            self.source_wrapped.insert(source.clone(), false);
        }
        let _ = self.rebuild_source_index();
        self.source_state_dirty = self.source_order.len() > 1;
    }

    fn select_temporal_neighbor(
        &'_ self,
        record: &DataRecord,
        offset_days: i32,
    ) -> Option<Arc<DataRecord>> {
        let target = record.created_at + Duration::days(offset_days.into());
        let key = record.taxonomy.first().cloned();
        let record_split = self.split_store.label_for(&RecordKey::from(record))?;
        self.records
            .values()
            .filter(|candidate| {
                RecordKey::from(*candidate) != RecordKey::from(record)
                    && self
                        .split_store
                        .label_for(&RecordKey::from(*candidate))
                        .map(|label| label == record_split)
                        .unwrap_or(false)
                    && (candidate.source == record.source
                        || key
                            .as_ref()
                            .zip(candidate.taxonomy.first())
                            .map(|(a, b)| a == b)
                            .unwrap_or(false))
            })
            .min_by_key(|candidate| (candidate.created_at - target).num_seconds().abs())
            .cloned()
    }

    fn select_negative_record(
        &self,
        anchor_record: &DataRecord,
        strategy: &NegativeStrategy,
        anchor_query_text: Option<&str>,
        rng: &mut dyn rand::RngCore,
    ) -> Option<(Arc<DataRecord>, bool)> {
        let anchor_split = self
            .split_store
            .label_for(&RecordKey::from(anchor_record))?;

        let in_anchor_split = |candidate: &DataRecord| {
            self.split_store
                .label_for(&RecordKey::from(candidate))
                .map(|label| label == anchor_split)
                .unwrap_or(false)
        };

        match strategy {
            NegativeStrategy::WrongArticle => {
                let anchor_date =
                    taxonomy_value(anchor_record, META_FIELD_DATE).map(|d| d.to_string());
                let mut same_date: Vec<Arc<DataRecord>> = self
                    .records
                    .values()
                    .filter(|candidate| {
                        candidate.source == anchor_record.source
                            && candidate.id != anchor_record.id
                            && in_anchor_split(candidate)
                    })
                    .filter(|candidate| {
                        anchor_date
                            .as_deref()
                            .zip(taxonomy_value(candidate, META_FIELD_DATE))
                            .map(|(a, b)| a == b)
                            .unwrap_or(false)
                    })
                    .cloned()
                    .collect();
                if same_date.is_empty() {
                    same_date = self
                        .records
                        .values()
                        .filter(|candidate| {
                            candidate.source == anchor_record.source
                                && candidate.id != anchor_record.id
                                && in_anchor_split(candidate)
                        })
                        .cloned()
                        .collect();
                }
                if !same_date.is_empty() {
                    return self.negative_backend.choose_negative(
                        anchor_record,
                        anchor_split,
                        same_date,
                        false,
                        anchor_query_text,
                        rng,
                    );
                }
                let pool = self
                    .records
                    .values()
                    .filter(|candidate| {
                        candidate.id != anchor_record.id && in_anchor_split(candidate)
                    })
                    .cloned()
                    .collect::<Vec<_>>();
                self.negative_backend.choose_negative(
                    anchor_record,
                    anchor_split,
                    pool,
                    true,
                    anchor_query_text,
                    rng,
                )
            }
            NegativeStrategy::WrongPublicationDate => {
                let anchor_date =
                    taxonomy_value(anchor_record, META_FIELD_DATE).map(|d| d.to_string());
                let pool: Vec<Arc<DataRecord>> = self
                    .records
                    .values()
                    .filter(|candidate| {
                        candidate.source == anchor_record.source
                            && candidate.id != anchor_record.id
                            && in_anchor_split(candidate)
                    })
                    .filter(|candidate| {
                        match (
                            anchor_date.as_deref(),
                            taxonomy_value(candidate, META_FIELD_DATE),
                        ) {
                            (Some(anchor), Some(candidate_date)) => anchor != candidate_date,
                            (Some(_), None) => true,
                            (None, Some(_)) => true,
                            (None, None) => false,
                        }
                    })
                    .cloned()
                    .collect();
                if pool.is_empty() {
                    // Fallback to any other record in the same split so split boundaries
                    // remain strictly isolated.
                    let fallback_pool = self
                        .records
                        .values()
                        .filter(|candidate| {
                            candidate.id != anchor_record.id && in_anchor_split(candidate)
                        })
                        .cloned()
                        .collect::<Vec<_>>();

                    return self.negative_backend.choose_negative(
                        anchor_record,
                        anchor_split,
                        fallback_pool,
                        true,
                        anchor_query_text,
                        rng,
                    );
                }

                self.negative_backend.choose_negative(
                    anchor_record,
                    anchor_split,
                    pool,
                    false,
                    anchor_query_text,
                    rng,
                )
            }
            NegativeStrategy::QuestionAnswerMismatch => {
                let pool: Vec<Arc<DataRecord>> = self
                    .records
                    .values()
                    .filter(|candidate| {
                        candidate.source == anchor_record.source
                            && candidate.id != anchor_record.id
                            && in_anchor_split(candidate)
                    })
                    .cloned()
                    .collect();
                if pool.is_empty() {
                    // Fallback to any other record in the same split so split boundaries
                    // remain strictly isolated.
                    let fallback_pool = self
                        .records
                        .values()
                        .filter(|candidate| {
                            candidate.id != anchor_record.id && in_anchor_split(candidate)
                        })
                        .cloned()
                        .collect::<Vec<_>>();

                    return self.negative_backend.choose_negative(
                        anchor_record,
                        anchor_split,
                        fallback_pool,
                        true,
                        anchor_query_text,
                        rng,
                    );
                }

                self.negative_backend.choose_negative(
                    anchor_record,
                    anchor_split,
                    pool,
                    false,
                    anchor_query_text,
                    rng,
                )
            }
            NegativeStrategy::SameRecord => {
                // Return the anchor record itself.  The caller
                // (finalize_triplet_with_negative) will call select_chunk on
                // this record using the recipe's negative_selector, which
                // picks from Context sections.  The dedup guard rejects
                // triplets where any two slots share identical text.
                Some((Arc::new(anchor_record.clone()), false))
            }
        }
    }

    /// True when the recipe is the special auto-injected long-section chunk-pair recipe.
    fn is_auto_chunk_pair_recipe(recipe: &TripletRecipe) -> bool {
        recipe.name.as_ref() == AUTO_INJECTED_LONG_SECTION_CHUNK_PAIR_RECIPE_NAME
    }

    /// Select anchor/positive chunk pair from selectors for a single record.
    ///
    /// When selectors are the same, retries until a valid pair is found or retry limit is hit.
    fn select_anchor_positive_pair(
        &mut self,
        record: &DataRecord,
        anchor_selector: &Selector,
        positive_selector: &Selector,
        enforce_window_pair: bool,
    ) -> Option<(RecordChunk, RecordChunk)> {
        let mut anchor_chunk = self.select_chunk(record, anchor_selector)?;
        let mut positive_chunk = self.select_chunk(record, positive_selector)?;
        if anchor_selector == positive_selector {
            let mut retries = 0usize;
            while !same_selector_pair_is_valid(&anchor_chunk, &positive_chunk, enforce_window_pair)
                && retries < SAME_SELECTOR_PAIR_RETRY_LIMIT
            {
                let Some(redraw_anchor) = self.select_chunk(record, anchor_selector) else {
                    break;
                };
                let Some(redraw_positive) = self.select_chunk(record, positive_selector) else {
                    break;
                };
                anchor_chunk = redraw_anchor;
                positive_chunk = redraw_positive;
                retries += 1;
            }
            if !same_selector_pair_is_valid(&anchor_chunk, &positive_chunk, enforce_window_pair) {
                return None;
            }
        }
        Some((anchor_chunk, positive_chunk))
    }

    /// Select two distinct *window* chunks for the auto-injected recipe using recipe selectors.
    ///
    /// This method intentionally uses the recipe's selectors (no hardcoded selector values).
    /// For the current auto recipe those selectors are both `Role(Context)`, but the behavior
    /// remains tied to recipe configuration, not this helper body.
    ///
    /// NOTE: this does *not* decide whether the auto recipe is enabled for a source.
    /// Source-level enablement happens in `resolve_source_triplet_plan` via
    /// `should_auto_inject_chunk_pair_recipe`.
    fn select_distinct_window_pair_for_auto_recipe(
        &mut self,
        recipe: &TripletRecipe,
        record: &DataRecord,
    ) -> Option<(RecordChunk, RecordChunk)> {
        if recipe.anchor != recipe.positive_selector {
            return None;
        }
        self.select_anchor_positive_pair(record, &recipe.anchor, &recipe.positive_selector, true)
    }

    /// Stage B (record-level): guard for the auto-injected recipe.
    ///
    /// Requires at least two `ChunkView::Window` candidates across sections addressed by
    /// `selector`. This makes the auto recipe eligibility check explicit at execution time.
    fn record_has_at_least_two_window_chunks_for_selector(
        &self,
        record: &DataRecord,
        selector: &Selector,
    ) -> bool {
        let section_indices: Vec<usize> = match selector {
            Selector::Role(role) => record
                .sections
                .iter()
                .enumerate()
                .filter(|(_, section)| roles_match(role, &section.role))
                .map(|(idx, _)| idx)
                .collect(),
            Selector::Paragraph(idx) => {
                if *idx < record.sections.len() {
                    vec![*idx]
                } else {
                    Vec::new()
                }
            }
            Selector::Random => (0..record.sections.len()).collect(),
            Selector::TemporalOffset(_) => return false,
        };

        let mut window_count = 0usize;
        for section_idx in section_indices {
            let Some(section) = record.sections.get(section_idx) else {
                continue;
            };
            let chunks = self.materialize_chunks(record, section_idx, section);
            window_count += chunks
                .iter()
                .filter(|chunk| matches!(chunk.view, ChunkView::Window { .. }))
                .count();
            if window_count >= 2 {
                return true;
            }
        }
        false
    }

    /// Serial candidate builder with per-batch negative-text dedup (Option A).
    ///
    /// Draws the anchor/positive pair once, hoists the 50 % swap + (A) guard
    /// out of the retry loop, then redraws **only the negative** up to
    /// `NEG_DEDUP_RETRIES` times. The coin-flip swap (one RNG draw, negative
    /// unaffected) defeats positional shortcuts so the model treats both
    /// positive slots symmetrically. All validation reads use references;
    /// ownership of `final_anchor`/`final_positive` moves into the triplet
    /// only after a passing negative is found (then return). Text guard (A)
    /// (`a == p` when forbidden) is checked before the loop — unrecoverable.
    /// Loop checks (B) (`fresh_tuple`, `fresh_neg`, `n == a/p`) on references.
    fn make_triplet_with_unique_negative(
        &mut self,
        recipe: &TripletRecipe,
        record: &DataRecord,
        rng: &mut DeterministicRng,
        seen: &mut HashSet<(RecordKey, RecordKey, RecordKey)>,
        seen_negatives: &mut IntSet<u64>,
    ) -> Option<SampleTriplet> {
        // Draw anchor/positive once (serial cursor semantics).
        let (mut anchor_chunk, mut positive_chunk) = if Self::is_auto_chunk_pair_recipe(recipe) {
            if !self.record_has_at_least_two_window_chunks_for_selector(record, &recipe.anchor) {
                return None;
            }
            self.select_distinct_window_pair_for_auto_recipe(recipe, record)?
        } else {
            self.select_anchor_positive_pair(
                record,
                &recipe.anchor,
                &recipe.positive_selector,
                false,
            )?
        };
        let anchor_raw_text = anchor_chunk.text.clone();
        self.decorate_chunk(record, &mut anchor_chunk, rng);
        self.decorate_chunk(record, &mut positive_chunk, rng);

        // Hoisted swap (single RNG draw) + guard (A).
        let (final_anchor, final_positive) = if rng.next_u64() & ANCHOR_POSITIVE_SWAP_MASK == 0 {
            (positive_chunk, anchor_chunk)
        } else {
            (anchor_chunk, positive_chunk)
        };
        if !recipe.allow_same_anchor_positive && final_anchor.text == final_positive.text {
            return None;
        }

        // Negative-only retries; references only, no anchor/positive clones.
        let mut accepted: Option<(RecordChunk, f32, bool)> = None;
        for _ in 0..NEG_DEDUP_RETRIES {
            let (negative_record, fallback_used) = self.select_negative_record(
                record,
                &recipe.negative_strategy,
                Some(&anchor_raw_text),
                rng,
            )?;
            let mut negative_chunk =
                match self.select_chunk(&negative_record, &recipe.negative_selector) {
                    Some(c) => c,
                    None => continue,
                };
            self.decorate_chunk(&negative_record, &mut negative_chunk, rng);
            if negative_chunk.text == final_positive.text
                || negative_chunk.text == final_anchor.text
            {
                continue;
            }
            let key = (
                RecordKey::of_chunk(&final_anchor),
                RecordKey::of_chunk(&final_positive),
                RecordKey::of_chunk(&negative_chunk),
            );
            let neg_h = negative_text_hash(&negative_chunk);
            // Atomic check-then-commit: no partial `seen` mutation on reject.
            if seen.contains(&key) || seen_negatives.contains(&neg_h) {
                continue;
            }
            let chunk_weight =
                self.triplet_chunk_weight(recipe, &final_anchor, &final_positive, &negative_chunk);
            let weight = recipe.weight * chunk_weight;
            seen.insert(key);
            seen_negatives.insert(neg_h);
            accepted = Some((negative_chunk, weight, fallback_used));
            break;
        }
        let (negative_chunk, weight, fallback_used) = accepted?;
        let recipe_name = if fallback_used {
            format!("{}_fallback_same_split", recipe.name)
        } else {
            recipe.name.to_string()
        };
        Some(SampleTriplet {
            recipe: recipe_name,
            anchor: final_anchor,
            positive: final_positive,
            negative: negative_chunk,
            weight,
            instruction: recipe.instruction.as_ref().map(|s| s.to_string()),
        })
    }

    fn make_text_sample_for_split(
        &mut self,
        recipe: &TextRecipe,
        source: Option<&str>,
        split: SplitLabel,
        rng: &mut DeterministicRng,
    ) -> Option<TextSample> {
        let record = self.choose_anchor_record(source, split)?;
        let mut chunk = self.select_chunk(&record, &recipe.selector)?;
        self.decorate_chunk(&record, &mut chunk, rng);
        let weight = recipe.weight * self.chunk_weight(&chunk);
        Some(TextSample {
            recipe: recipe.name.to_string(),
            chunk,
            weight,
            instruction: recipe.instruction.as_ref().map(|s| s.to_string()),
        })
    }

    fn chunk_weight(&self, chunk: &RecordChunk) -> f32 {
        chunk_weight(&self.config.chunking, chunk)
    }

    fn triplet_chunk_weight(
        &self,
        recipe: &TripletRecipe,
        anchor: &RecordChunk,
        positive: &RecordChunk,
        negative: &RecordChunk,
    ) -> f32 {
        let floor = self.config.chunking.chunk_weight_floor;
        let negative_weight = negative.quality.trust.clamp(0.0, 1.0).max(floor);
        if Self::is_auto_chunk_pair_recipe(recipe) {
            // For the auto long-section recipe, use one coherence signal (proximity)
            // for the anchor/positive pair.
            let pair_trust = ((anchor.quality.trust.clamp(0.0, 1.0)
                + positive.quality.trust.clamp(0.0, 1.0))
                / 2.0)
                .clamp(0.0, 1.0);
            let pair_weight = (chunk_proximity_score(anchor, positive) * pair_trust).max(floor);
            // Keep negative weighting simple in this recipe: trust + floor only.
            return (pair_weight + pair_weight + negative_weight) / 3.0;
        }
        // Non-auto recipes also apply anchor-positive proximity, while keeping
        // negative weighting trust-only.
        let pair_proximity = chunk_proximity_score(anchor, positive);
        let anchor_weight = (self.chunk_weight(anchor) * pair_proximity).max(floor);
        let positive_weight = (self.chunk_weight(positive) * pair_proximity).max(floor);
        (anchor_weight + positive_weight + negative_weight) / 3.0
    }

    fn decorate_chunk(
        &mut self,
        record: &DataRecord,
        chunk: &mut RecordChunk,
        rng: &mut DeterministicRng,
    ) {
        chunk.kvp_meta = record
            .meta_prefix
            .as_ref()
            .map(|s| s.all_metadata())
            .unwrap_or_default();
        if let Some(spec) = record.meta_prefix.as_ref()
            && let Some(prefix) = spec.sample(rng)
        {
            let body_tokens: Vec<&str> = WhitespaceTokenizer.tokenize(&chunk.text);
            let prefix_tokens: Vec<&str> = WhitespaceTokenizer.tokenize(&prefix);
            let total_tokens = prefix_tokens.len() + body_tokens.len();
            let max_window = self.config.chunking.max_window_tokens;
            if max_window > 0 && total_tokens > max_window {
                if prefix_tokens.len() >= max_window {
                    chunk.text = prefix_tokens
                        .into_iter()
                        .take(max_window)
                        .collect::<Vec<_>>()
                        .join(" ");
                    chunk.tokens_estimate = max_window;
                } else {
                    let remaining = max_window - prefix_tokens.len();
                    let trimmed_body: Vec<&str> = body_tokens.into_iter().take(remaining).collect();
                    chunk.text =
                        format!("{}{}{}", prefix, platform_newline(), trimmed_body.join(" "));
                    chunk.tokens_estimate = max_window;
                }
            } else {
                chunk.text = format!("{}{}{}", prefix, platform_newline(), chunk.text);
                chunk.tokens_estimate = total_tokens;
            }
        }
    }

    // ── parallel-batch helpers (&self, rng-based, no cursor maps) ────────────

    /// Select a chunk using rng instead of cursor maps (for parallel execution).
    fn select_chunk_parallel(
        &self,
        record: &DataRecord,
        selector: &Selector,
        rng: &mut DeterministicRng,
    ) -> Option<RecordChunk> {
        match selector {
            Selector::Role(role) => self.select_role_parallel(record, role, rng),
            Selector::Paragraph(idx) => record.sections.get(*idx).and_then(|section| {
                let pool = self.materialize_chunks(record, *idx, section);
                if pool.is_empty() {
                    return None;
                }
                let i = rng.random_range(0..pool.len());
                pool.into_iter().nth(i)
            }),
            Selector::TemporalOffset(offset) => self
                .select_temporal_neighbor(record, *offset)
                .and_then(|neighbor| {
                    self.select_role_parallel(&neighbor, &SectionRole::Context, rng)
                }),
            Selector::Random => {
                if record.sections.is_empty() {
                    return None;
                }
                let idx = rng.random_range(0..record.sections.len());
                record.sections.get(idx).and_then(|section| {
                    let pool = self.materialize_chunks(record, idx, section);
                    if pool.is_empty() {
                        return None;
                    }
                    let i = rng.random_range(0..pool.len());
                    pool.into_iter().nth(i)
                })
            }
        }
    }

    /// Select a chunk by role using rng (parallel path — no role_cursors written).
    fn select_role_parallel(
        &self,
        record: &DataRecord,
        role: &SectionRole,
        rng: &mut DeterministicRng,
    ) -> Option<RecordChunk> {
        let indices: Vec<usize> = record
            .sections
            .iter()
            .enumerate()
            .filter(|(_, s)| roles_match(role, &s.role))
            .map(|(i, _)| i)
            .collect();
        if indices.is_empty() {
            return None;
        }
        let start = rng.random_range(0..indices.len());
        for offset in 0..indices.len() {
            let section_idx = indices[(start + offset) % indices.len()];
            let section = &record.sections[section_idx];
            let pool = self.materialize_chunks(record, section_idx, section);
            if !pool.is_empty() {
                let i = rng.random_range(0..pool.len());
                return pool.into_iter().nth(i);
            }
        }
        None
    }

    /// Decorate a chunk using a provided rng (parallel-safe, &self).
    fn decorate_chunk_parallel(
        &self,
        record: &DataRecord,
        chunk: &mut RecordChunk,
        rng: &mut DeterministicRng,
    ) {
        chunk.kvp_meta = record
            .meta_prefix
            .as_ref()
            .map(|s| s.all_metadata())
            .unwrap_or_default();
        if let Some(spec) = record.meta_prefix.as_ref()
            && let Some(prefix) = spec.sample(rng)
        {
            let body_tokens: Vec<&str> = WhitespaceTokenizer.tokenize(&chunk.text);
            let prefix_tokens: Vec<&str> = WhitespaceTokenizer.tokenize(&prefix);
            let total_tokens = prefix_tokens.len() + body_tokens.len();
            let max_window = self.config.chunking.max_window_tokens;
            if max_window > 0 && total_tokens > max_window {
                if prefix_tokens.len() >= max_window {
                    chunk.text = prefix_tokens
                        .into_iter()
                        .take(max_window)
                        .collect::<Vec<_>>()
                        .join(" ");
                    chunk.tokens_estimate = max_window;
                } else {
                    let remaining = max_window - prefix_tokens.len();
                    let trimmed_body: Vec<&str> = body_tokens.into_iter().take(remaining).collect();
                    chunk.text =
                        format!("{}{}{}", prefix, platform_newline(), trimmed_body.join(" "));
                    chunk.tokens_estimate = max_window;
                }
            } else {
                chunk.text = format!("{}{}{}", prefix, platform_newline(), chunk.text);
                chunk.tokens_estimate = total_tokens;
            }
        }
    }

    /// Select an anchor/positive pair for the parallel path (rng-based).
    fn select_anchor_positive_parallel(
        &self,
        record: &DataRecord,
        anchor_selector: &Selector,
        positive_selector: &Selector,
        enforce_window_pair: bool,
        rng: &mut DeterministicRng,
    ) -> Option<(RecordChunk, RecordChunk)> {
        let anchor_chunk = self.select_chunk_parallel(record, anchor_selector, rng)?;
        let mut positive_chunk = self.select_chunk_parallel(record, positive_selector, rng)?;
        if anchor_selector == positive_selector {
            let mut retries = 0usize;
            while !same_selector_pair_is_valid(&anchor_chunk, &positive_chunk, enforce_window_pair)
                && retries < SAME_SELECTOR_PAIR_RETRY_LIMIT
            {
                positive_chunk = self.select_chunk_parallel(record, positive_selector, rng)?;
                retries += 1;
            }
            if !same_selector_pair_is_valid(&anchor_chunk, &positive_chunk, enforce_window_pair) {
                return None;
            }
        }
        // No swap here — finalize_triplet_parallel applies the coin-flip swap.
        Some((anchor_chunk, positive_chunk))
    }

    /// Select (anchor_chunk, positive_chunk, anchor_raw_text) for a recipe — parallel-safe.
    /// Does NOT select a negative; that happens sequentially in Phase 3.
    fn select_anchor_positive_for_recipe(
        &self,
        recipe: &TripletRecipe,
        anchor_record: &DataRecord,
        rng: &mut DeterministicRng,
    ) -> Option<(RecordChunk, RecordChunk, String)> {
        if Self::is_auto_chunk_pair_recipe(recipe) {
            if !self
                .record_has_at_least_two_window_chunks_for_selector(anchor_record, &recipe.anchor)
            {
                return None;
            }
            let mut anchor_chunk =
                self.select_chunk_parallel(anchor_record, &recipe.anchor, rng)?;
            let mut positive_chunk =
                self.select_chunk_parallel(anchor_record, &recipe.anchor, rng)?;
            let mut tries = 0usize;
            while !same_selector_pair_is_valid(&anchor_chunk, &positive_chunk, true) {
                tries += 1;
                if tries >= SAME_SELECTOR_PAIR_RETRY_LIMIT {
                    return None;
                }
                anchor_chunk = self.select_chunk_parallel(anchor_record, &recipe.anchor, rng)?;
                positive_chunk = self.select_chunk_parallel(anchor_record, &recipe.anchor, rng)?;
            }
            let anchor_raw_text = anchor_chunk.text.clone();
            self.decorate_chunk_parallel(anchor_record, &mut anchor_chunk, rng);
            self.decorate_chunk_parallel(anchor_record, &mut positive_chunk, rng);
            return Some((anchor_chunk, positive_chunk, anchor_raw_text));
        }
        let (mut anchor_chunk, mut positive_chunk) = self.select_anchor_positive_parallel(
            anchor_record,
            &recipe.anchor,
            &recipe.positive_selector,
            false,
            rng,
        )?;
        let anchor_raw_text = anchor_chunk.text.clone();
        self.decorate_chunk_parallel(anchor_record, &mut anchor_chunk, rng);
        self.decorate_chunk_parallel(anchor_record, &mut positive_chunk, rng);
        Some((anchor_chunk, positive_chunk, anchor_raw_text))
    }

    fn select_chunk(&mut self, record: &DataRecord, selector: &Selector) -> Option<RecordChunk> {
        match selector {
            Selector::Role(role) => self.select_by_role(record, role),
            Selector::Paragraph(idx) => record.sections.get(*idx).and_then(|section| {
                let pool = self.materialize_chunks(record, *idx, section);
                self.next_chunk_from_pool(&RecordKey::from(record), *idx, pool)
            }),
            Selector::TemporalOffset(offset) => self
                .select_temporal_neighbor(record, *offset)
                .and_then(|neighbor| self.select_by_role(&neighbor, &SectionRole::Context)),
            Selector::Random => {
                if record.sections.is_empty() {
                    return None;
                }
                let idx = self.rng.random_range(0..record.sections.len());
                record.sections.get(idx).and_then(|section| {
                    let pool = self.materialize_chunks(record, idx, section);
                    self.next_chunk_from_pool(&RecordKey::from(record), idx, pool)
                })
            }
        }
    }

    fn select_by_role(&mut self, record: &DataRecord, role: &SectionRole) -> Option<RecordChunk> {
        let indices: Vec<usize> = record
            .sections
            .iter()
            .enumerate()
            .filter(|(_, section)| roles_match(role, &section.role))
            .map(|(idx, _)| idx)
            .collect();
        if indices.is_empty() {
            return None;
        }
        let key = role_cursor_key(&RecordKey::from(record), role);
        let start_offset = self
            .role_cursors
            .get(&key)
            .and_then(|last_idx| indices.iter().position(|idx| idx == last_idx))
            .map(|pos| (pos + 1) % indices.len())
            .unwrap_or_else(|| {
                // On first use for this (record,role), choose a deterministic hashed
                // section offset instead of always starting at the first matching section.
                // This avoids systematic head-bias while preserving reproducibility.
                // Preimage replicates the historical "{id}::{role}" form so
                // sequences are unchanged for existing corpora.
                let record_key = RecordKey::from(record);
                let seed_key = format!("{}::{}", record_key.id, role_label(role));
                (stable_hash_str(self.epoch_seed(), &seed_key) as usize) % indices.len()
            });
        for offset in 0..indices.len() {
            let section_idx = indices[(start_offset + offset) % indices.len()];
            let section = &record.sections[section_idx];
            let pool = self.materialize_chunks(record, section_idx, section);
            if let Some(chunk) =
                self.next_chunk_from_pool(&RecordKey::from(record), section_idx, pool)
            {
                self.role_cursors.insert(key.clone(), section_idx);
                return Some(chunk);
            }
        }
        None
    }

    /// Materialize chunk windows for one section according to `ChunkingStrategy`.
    ///
    /// Window layout algorithm:
    /// 1) Tokenize section text with `split_whitespace`.
    /// 2) Set `span = min(max_window_tokens, total_tokens)`.
    /// 3) If `span == total_tokens`, emit one full-section `ChunkView::Window`.
    /// 4) Otherwise, for each configured overlap:
    ///    - `stride = max(1, span - overlap)`
    ///    - emit windows `[start, min(start + span, total_tokens))` while advancing by `stride`.
    /// 5) Optionally append one `SummaryFallback` chunk when section is longer than `max_window_tokens`.
    ///
    /// This is the source of truth for how chunk windows are "laid" in a record.
    fn materialize_chunks(
        &self,
        record: &DataRecord,
        section_idx: usize,
        section: &RecordSection,
    ) -> Vec<RecordChunk> {
        let mut chunks =
            self.chunker
                .materialize(&self.config.chunking, record, section_idx, section);
        // Propagate the section's precomputed embedding (if any) into every
        // chunk materialized from it. The chunker stays agnostic; Arc makes
        // this a pointer copy, never a vector copy.
        for chunk in &mut chunks {
            chunk.embedding = section.embedding.clone();
            chunk.source = record.source.clone();
        }
        chunks
    }

    fn build_derived_text_recipes(recipes: &[TripletRecipe]) -> Vec<TextRecipe> {
        let mut derived = Vec::new();
        for recipe in recipes {
            let base = recipe.name.as_ref();
            derived.push(TextRecipe {
                name: Cow::Owned(format!("{base}_anchor")),
                selector: recipe.anchor.clone(),
                weight: recipe.weight,
                instruction: None,
            });
            derived.push(TextRecipe {
                name: Cow::Owned(format!("{base}_positive")),
                selector: recipe.positive_selector.clone(),
                weight: recipe.weight,
                instruction: None,
            });
            derived.push(TextRecipe {
                name: Cow::Owned(format!("{base}_negative")),
                selector: recipe.negative_selector.clone(),
                weight: recipe.weight,
                instruction: None,
            });
        }
        derived
    }

    /// Stage A helper: true when record contains an Anchor/Context section whose token
    /// count exceeds `chunking.max_window_tokens`.
    fn record_has_long_anchor_or_context_section(&self, record: &DataRecord) -> bool {
        let window = self.config.chunking.max_window_tokens;
        if window == 0 {
            return false;
        }
        record.sections.iter().any(|s| {
            matches!(s.role, SectionRole::Anchor | SectionRole::Context) && s.token_count > window
        })
    }
    fn sync_records_from_cache(&mut self) -> Result<(), SamplerError> {
        if self.last_sync_version == 0 {
            self.full_sync_records_from_cache()?;
            return Ok(());
        }

        // Force refreshes replace cache contents wholesale; fall back to a
        // full pool resync with exact full-snapshot semantics (evicted =
        // pool minus snapshot, dedup pruned only for records that left).
        // Steady-state advances keep the O(Δ) delta fast path below.
        if self.ingestion.force_refresh_generation() != self.last_force_refresh_generation {
            self.last_force_refresh_generation = self.ingestion.force_refresh_generation();
            self.full_sync_records_from_cache()?;
            return Ok(());
        }

        // Use sync_delta to get only added and evicted records.
        // No full snapshot, no sort, no HashSet diff — O(ΔN) path.
        let (new_version, added, evicted) = self.ingestion.sync_delta(self.last_sync_version);

        if added.is_empty() && evicted.is_empty() {
            self.negative_backend.on_sync_start();
            self.prune_cursor_state();
            self.last_sync_version = new_version;
            return Ok(());
        }

        // --- Process evictions (strict O(Δ)): prune pool state and the
        // per-source index only for IDs that truly left. IDs cleared and
        // immediately re-added in this same sync (e.g. force-refresh
        // returning the same records) keep their cross-batch dedup and
        // split-label state. This preserves the
        // `emitted_text_hashes_blocks_repeats` contract while still freeing
        // state for evicted records.
        let mut truly_evicted = false;
        if !evicted.is_empty() {
            let added_keys: HashSet<RecordKey> = added.iter().map(RecordKey::of_arc).collect();
            for evicted_key in &evicted {
                if added_keys.contains(evicted_key) {
                    continue;
                }
                let Some(record) = self.records.get(evicted_key) else {
                    continue;
                };
                let source = record.source.clone();
                let is_long = self.record_has_long_anchor_or_context_section(record);
                self.records.shift_remove(evicted_key);
                self.emitted_text_hashes.remove(evicted_key);
                self.split_labels.remove(evicted_key);
                self.chunk_index.remove(evicted_key);
                self.remove_id_from_source_index(&source, evicted_key);
                if is_long {
                    self.decrement_long_section_count(&source);
                }
                truly_evicted = true;
            }
        }

        self.negative_backend.on_sync_start();

        // --- Process additions (strict O(Δ log N) via binary search) ---
        // `pool_changed` tracks actual membership change: genuinely new IDs
        // or true evictions. Pure version refreshes of already-pooled IDs
        // leave the pool (and the cached records_by_split map) intact.
        let mut pool_changed = truly_evicted;
        if !added.is_empty() {
            let allowed: HashSet<SplitLabel> = self.allowed_target_splits().into_iter().collect();
            for record in &added {
                // Re-ingested keys (version bump for records already pooled,
                // e.g. a wrapped source re-draining known records) must not
                // create duplicate index entries or double-count long
                // sections. A full rebuild derives each key exactly once from
                // the pool, so skip index/count work here and only refresh
                // the stored Arc.
                let key = RecordKey::from(record);
                if self.records.contains_key(&key) {
                    self.records.insert(key, Arc::clone(record));
                    continue;
                }
                pool_changed = true;
                let label = self.get_or_insert_split_label(&key)?;
                if self.record_has_long_anchor_or_context_section(record) {
                    self.increment_long_section_count(&record.source);
                }
                self.records.insert(key.clone(), Arc::clone(record));
                self.chunk_index.insert(key.clone(), key.clone());
                self.insert_id_into_source_index(record.source.clone(), key, label, &allowed);
            }
        }

        // --- Refresh derived source state (mirrors `rebuild_source_index` tail) ---
        // Drop index entries for sources left with no allowed-split records,
        // then refresh ordering, wrap flags, and cursors exactly as a full
        // rebuild would.
        let empty_sources: Vec<SourceId> = self
            .source_record_indices
            .iter()
            .filter(|(_, ids)| ids.is_empty())
            .map(|(source, _)| source.clone())
            .collect();
        let mut order_changed = !empty_sources.is_empty();
        for source in &empty_sources {
            self.source_record_indices.remove(source);
        }
        // New sources may have appeared via the insertions above.
        if !order_changed {
            for source in self.source_record_indices.keys() {
                if !self.source_order.contains(source) {
                    order_changed = true;
                    break;
                }
            }
        }
        if order_changed {
            self.source_order = self.source_record_indices.keys().cloned().collect();
            self.source_order.sort();
        }
        // Drop deficit counters for sources that left the pool.
        self.source_deficits
            .retain(|s, _| self.source_record_indices.contains_key(s));
        self.refresh_source_wrapped();
        self.source_record_cursors
            .retain(|source, _| self.source_record_indices.contains_key(source));
        if self.source_state_loaded {
            if self.source_order.is_empty() {
                self.source_cycle_idx = 0;
            }
            self.source_state_dirty = self.source_order.len() > 1;
        }
        // Cursor pruning is a no-op unless the valid pool shrank.
        if truly_evicted {
            self.prune_cursor_state();
        }

        if pool_changed {
            self.pool_generation = self.pool_generation.saturating_add(1);
        }
        self.last_sync_version = new_version;
        Ok(())
    }

    /// Full snapshot sync: used only for the initial bootstrap (last_sync_version == 0).
    /// Inserts records in ID-sorted order so subsequent delta syncs can append
    /// new records in sorted order without breaking the IndexMap ordering.
    fn full_sync_records_from_cache(&mut self) -> Result<(), SamplerError> {
        let mut snapshot = self.ingestion.all_records_snapshot();
        snapshot.sort_by_key(RecordKey::of_arc);

        // Determine which records are being evicted BEFORE clearing self.records.
        let old_keys: HashSet<RecordKey> = self.records.keys().cloned().collect();
        let incoming_keys: HashSet<RecordKey> = snapshot.iter().map(RecordKey::of_arc).collect();
        let evicted: HashSet<&RecordKey> = old_keys.difference(&incoming_keys).collect();

        self.records.clear();
        self.long_section_counts.clear();
        for evicted_key in &evicted {
            self.emitted_text_hashes.remove(*evicted_key);
            self.split_labels.remove(*evicted_key);
        }
        self.negative_backend.on_sync_start();
        for record in snapshot {
            let key = RecordKey::from(&record);
            let _ = self.get_or_insert_split_label(&key)?;
            if self.record_has_long_anchor_or_context_section(&record) {
                self.increment_long_section_count(&record.source);
            }
            self.records.insert(key, Arc::clone(&record));
        }
        self.prune_cursor_state();
        self.rebuild_chunk_index();
        self.rebuild_source_index()?;
        self.last_sync_version = self.ingestion.global_version();
        // The pool was fully replaced; invalidate the cached split map.
        self.pool_generation = self.pool_generation.saturating_add(1);
        Ok(())
    }

    fn ingest_internal_for_split(&mut self, target_split: SplitLabel) -> Result<(), SamplerError> {
        if !self.ingestion.has_sources() {
            return Ok(());
        }
        if !self.ingestion_cursors_loaded {
            if let Some(state) = self.split_store.load_sampler_state()? {
                self.ingestion.load_cursors(&state.source_stream_cursors);
                self.ingestion.set_epoch(state.epoch);
                self.ingestion.set_epoch_step(state.epoch_step);
            }
            self.ingestion_cursors_loaded = true;
        }
        if self.ingestion.all_caches_empty() {
            self.ingestion.refresh_all();
        } else {
            self.ingestion.advance(self.config.batch_size);
        }
        let observed = self.ingestion.total_ingest_count();
        if observed == self.last_observed_ingest && !self.records.is_empty() {
            return Ok(());
        }
        self.last_observed_ingest = observed;
        self.sync_records_from_cache()?;
        let max_window_tokens = self.config.chunking.max_window_tokens;
        self.negative_backend.on_records_refreshed(
            &self.records,
            max_window_tokens,
            &|id| self.split_labels.get(id).copied(),
            self.ingestion.last_refreshed_sources(),
        );
        // Epoch tracking and source-state management must run every batch regardless
        // of whether the record pool changed — reconcile advances the sampling cursor.
        self.epoch_tracker.ensure_loaded()?;
        self.refresh_records_by_split_cache()?;
        let cached_records_by_split = self
            .cached_records_by_split
            .as_ref()
            .expect("records_by_split cache populated by refresh");
        self.epoch_tracker
            .reconcile(target_split, cached_records_by_split);
        self.ensure_source_state()?;
        Ok(())
    }

    /// Ensure ingestion cursors (including epoch step) are loaded from the
    /// persisted state so the step counter reflects the saved value before
    /// the wrapper increments it for the current call.
    ///
    /// Only loads ingestion cursors and source epoch.  The sampling state
    /// (RNG, source_record_cursors, …) is restored later by
    /// `ingest_internal_for_split` → `ensure_source_state`, after
    /// `sync_records_from_cache` has populated `source_record_indices`.
    fn ensure_ingestion_cursors_loaded(&mut self) -> Result<(), SamplerError> {
        if !self.ingestion_cursors_loaded {
            if let Some(state) = self.split_store.load_sampler_state()? {
                self.ingestion.load_cursors(&state.source_stream_cursors);
                self.ingestion.set_epoch(state.epoch);
                self.ingestion.set_epoch_step(state.epoch_step);
            }
            self.ingestion_cursors_loaded = true;
        }
        Ok(())
    }

    #[cfg(test)]
    fn ingest_internal(&mut self, split: SplitLabel) -> Result<(), SamplerError> {
        self.ingest_internal_for_split(split)
    }

    fn ingest_internal_with_weights_for_split(
        &mut self,
        target_split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<(), SamplerError> {
        if !self.ingestion.has_sources() {
            return Ok(());
        }
        if !self.ingestion_cursors_loaded {
            if let Some(state) = self.split_store.load_sampler_state()? {
                self.ingestion.load_cursors(&state.source_stream_cursors);
                self.ingestion.set_epoch(state.epoch);
                self.ingestion.set_epoch_step(state.epoch_step);
            }
            self.ingestion_cursors_loaded = true;
        }
        if self.ingestion.all_caches_empty() {
            self.ingestion.refresh_all_with_weights(weights)?;
        } else {
            self.ingestion
                .advance_with_weights(self.config.batch_size, weights)?;
        }
        let observed = self.ingestion.total_ingest_count();
        if observed == self.last_observed_ingest && !self.records.is_empty() {
            return Ok(());
        }
        self.last_observed_ingest = observed;
        self.sync_records_from_cache()?;
        let max_window_tokens = self.config.chunking.max_window_tokens;
        self.negative_backend.on_records_refreshed(
            &self.records,
            max_window_tokens,
            &|id| self.split_labels.get(id).copied(),
            self.ingestion.last_refreshed_sources(),
        );
        self.epoch_tracker.ensure_loaded()?;
        self.refresh_records_by_split_cache()?;
        let cached_records_by_split = self
            .cached_records_by_split
            .as_ref()
            .expect("records_by_split cache populated by refresh");
        self.epoch_tracker
            .reconcile(target_split, cached_records_by_split);
        self.ensure_source_state()?;
        Ok(())
    }

    /// Helper that centralises the weight-fallback logic.
    ///
    /// If `weights` is `Some` and non-uniform or if `weights` is `None`,
    /// delegates to the appropriate ingest method.
    fn ingest_with_weights_fallback(
        &mut self,
        target_split: SplitLabel,
        weights: Option<&HashMap<SourceId, f32>>,
    ) -> Result<(), SamplerError> {
        match weights {
            Some(weights)
                if !weights.is_empty()
                    && !weights
                        .values()
                        .all(|&w| w == *weights.values().next().unwrap()) =>
            {
                self.ingest_internal_with_weights_for_split(target_split, weights)?
            }
            _ => self.ingest_internal_for_split(target_split)?,
        }
        Ok(())
    }

    fn force_ingest_refresh_with_weights_for_split(
        &mut self,
        target_split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<(), SamplerError> {
        if !self.ingestion.has_sources() {
            return Ok(());
        }
        if !self.ingestion_cursors_loaded {
            if let Some(state) = self.split_store.load_sampler_state()? {
                self.ingestion.load_cursors(&state.source_stream_cursors);
                self.ingestion.set_epoch(state.epoch);
                self.ingestion.set_epoch_step(state.epoch_step);
            }
            self.ingestion_cursors_loaded = true;
        }
        self.ingestion.force_refresh_all_with_weights(weights)?;
        self.last_observed_ingest = self.ingestion.total_ingest_count();
        self.sync_records_from_cache()?;
        let max_window_tokens = self.config.chunking.max_window_tokens;
        self.negative_backend.on_records_refreshed(
            &self.records,
            max_window_tokens,
            &|id| self.split_labels.get(id).copied(),
            self.ingestion.last_refreshed_sources(),
        );
        self.epoch_tracker.ensure_loaded()?;
        self.refresh_records_by_split_cache()?;
        let cached_records_by_split = self
            .cached_records_by_split
            .as_ref()
            .expect("records_by_split cache populated by refresh");
        self.epoch_tracker
            .reconcile(target_split, cached_records_by_split);
        self.ensure_source_state()?;
        Ok(())
    }

    /// Select one triplet candidate for a specific source and split.
    ///
    /// This helper is the source-level recipe execution path used by both pair and
    /// triplet batch builders. It always starts from `triplet_recipes_for_source`,
    /// which already includes any auto-injected long-section recipe when eligible.
    ///
    /// Returns:
    /// - candidate triplet (with recipe) if one was sampled,
    /// - number of recipe attempts consumed (for round-robin bookkeeping).
    ///
    /// Takes the batch `seen` sets so negative-text dedup commits atomically.
    #[allow(clippy::too_many_arguments)]
    fn sample_source_triplet_candidate(
        &mut self,
        source: &str,
        target_split: SplitLabel,
        recipe_orders: &mut HashMap<RecipeKey, Vec<usize>>,
        recipe_positions: &mut HashMap<RecipeKey, usize>,
        rng: &mut DeterministicRng,
        seen: &mut HashSet<(RecordKey, RecordKey, RecordKey)>,
        seen_negatives: &mut IntSet<u64>,
    ) -> (Option<(TripletRecipe, SampleTriplet)>, usize) {
        // Stage A (source-level injection): resolve effective recipe pool,
        // including auto-injected long-section recipe when source is eligible.
        let (recipes, _auto_injected) = self.resolve_source_triplet_plan(source);
        if recipes.is_empty() {
            return (None, 0);
        }
        if !recipe_orders.contains_key(source) {
            let recipe_weights: Vec<f32> = recipes.iter().map(|r| r.weight).collect();
            let order =
                self.recipe_order_weighted_cycled(&recipe_weights, self.triplet_recipe_rr_idx, rng);
            recipe_orders.insert(source.to_string(), order);
        }
        let order = recipe_orders
            .get(source)
            .expect("recipe order missing for source");
        let pos = recipe_positions.entry(source.to_string()).or_insert(0);
        let Some(anchor) = self.choose_anchor_record(Some(source), target_split) else {
            return (None, 0);
        };

        let mut attempts = 0usize;
        for offset in 0..order.len() {
            let idx = order[(*pos + offset) % order.len()];
            attempts = attempts.saturating_add(1);
            let recipe = recipes[idx].clone();
            // Stage B/C happen inside the unique-negative builder, which draws
            // anchor/positive once then retries only the negative (per-batch
            // negative-text dedup, best-effort pre-pad). Commit is atomic
            // inside the helper.
            if let Some(sample) =
                self.make_triplet_with_unique_negative(&recipe, &anchor, rng, seen, seen_negatives)
            {
                *pos = (*pos + offset + 1) % order.len();
                return (Some((recipe, sample)), attempts);
            }
        }

        (None, attempts)
    }

    fn next_pair_batch_inner_with_weights(
        &mut self,
        target_split: SplitLabel,
        weights: Option<&HashMap<SourceId, f32>>,
    ) -> Result<SampleBatch, SamplerError> {
        self.ingest_with_weights_fallback(target_split, weights)?;
        self.ensure_split_has_records(target_split)?;
        let sources = self.source_order.clone();
        if sources.is_empty() {
            if self.triplet_recipes.is_empty() {
                return Err(SamplerError::Configuration(
                    "no triplet recipes available".into(),
                ));
            }
            let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
            let recipe_weights: Vec<f32> = self.triplet_recipes.iter().map(|r| r.weight).collect();
            let recipe_order = self.recipe_order_weighted_cycled(
                &recipe_weights,
                self.triplet_recipe_rr_idx,
                &mut rng,
            );
            let mut pairs = Vec::new();
            let mut seen = HashSet::new();
            // Per-batch negative-text dedup (Option A, best-effort pre-pad).
            // `pad_with_reuse` below may still duplicate when the unique pool
            // is smaller than `batch_size`.
            let mut seen_negatives: IntSet<u64> = IntSet::default();
            let mut last_recipe_name = None;
            let mut recipe_pos = 0usize;
            let mut recipe_steps = 0usize;
            let attempts = self.config.batch_size * 4 * recipe_order.len().max(1);
            for _ in 0..attempts {
                if pairs.len() >= self.config.batch_size {
                    break;
                }
                let Some(anchor) = self.choose_anchor_record(None, target_split) else {
                    break;
                };
                let mut triplet = None;
                for offset in 0..recipe_order.len() {
                    let idx = recipe_order[(recipe_pos + offset) % recipe_order.len()];
                    recipe_steps = recipe_steps.saturating_add(1);
                    let recipe = self.triplet_recipes[idx].clone();
                    last_recipe_name = Some(recipe.name.clone());
                    if let Some(sample) = self.make_triplet_with_unique_negative(
                        &recipe,
                        &anchor,
                        &mut rng,
                        &mut seen,
                        &mut seen_negatives,
                    ) {
                        triplet = Some((recipe, sample));
                        recipe_pos = (recipe_pos + offset + 1) % recipe_order.len();
                        break;
                    }
                }
                if let Some((recipe, triplet)) = triplet {
                    let SampleTriplet {
                        recipe: triplet_recipe_name,
                        anchor,
                        positive,
                        negative,
                        weight,
                        instruction,
                    } = triplet;

                    // Look up the anchor record's label to override structural defaults.
                    // For SRD pair-mode data, the label is stored in DataRecord.
                    let record_label = self
                        .records
                        .get(&RecordKey::of_chunk(&anchor))
                        .and_then(|r| r.label.clone());

                    if pairs.len() < self.config.batch_size {
                        pairs.push(SamplePair {
                            recipe: triplet_recipe_name.clone(),
                            anchor: anchor.clone(),
                            positive: positive.clone(),
                            weight,
                            instruction: instruction.clone(),
                            label: record_label.clone().unwrap_or(PairLabel::Positive),
                            reason: None,
                        });
                    }
                    if pairs.len() < self.config.batch_size {
                        pairs.push(SamplePair {
                            recipe: triplet_recipe_name,
                            anchor,
                            positive: negative,
                            weight,
                            instruction,
                            label: record_label.unwrap_or(PairLabel::Negative),
                            reason: Some(strategy_reason(&recipe.negative_strategy).to_string()),
                        });
                    }
                }
            }
            if recipe_steps > 0 {
                self.triplet_recipe_rr_idx =
                    self.triplet_recipe_rr_idx.saturating_add(recipe_steps);
            }
            self.rng = rng;
            pad_with_reuse(&mut pairs, self.config.batch_size);
            if pairs.len() == self.config.batch_size {
                return Ok(SampleBatch { pairs });
            }
            return Err(SamplerError::Exhausted(
                last_recipe_name
                    .unwrap_or(Cow::Borrowed(RECIPE_LABEL_TRIPLETS))
                    .to_string(),
            ));
        }

        let mut pairs = Vec::new();
        let mut seen = HashSet::new();
        // Per-batch negative-text dedup (Option A, best-effort pre-pad).
        let mut seen_negatives: IntSet<u64> = IntSet::default();
        let mut source_steps = 0usize;
        let mut cycle = (self.source_cycle_idx / sources.len()) as u64;
        let mut source_idx = self.source_cycle_idx % sources.len();
        let mut cycle_sources = self.shuffled_source_cycle(cycle);
        // Explicit non-uniform weights engage deficit round-robin source
        // choice; uniform/empty maps keep legacy uniform cycling bit-identical.
        let weighted: Option<&HashMap<SourceId, f32>> = match weights {
            Some(w) if Self::weights_are_selective(w) => Some(w),
            _ => None,
        };
        let mut recipe_orders: HashMap<RecipeKey, Vec<usize>> = HashMap::new();
        let mut recipe_positions: HashMap<RecipeKey, usize> = HashMap::new();
        let mut recipe_steps = 0usize;
        let max_recipe_len = sources
            .iter()
            .map(|source| self.triplet_recipe_count_for_source(source))
            .max()
            .unwrap_or(1)
            .max(1);
        let attempts = self.config.batch_size * 4 * sources.len() * max_recipe_len;
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        for _ in 0..attempts {
            if pairs.len() >= self.config.batch_size {
                break;
            }
            let weighted_source;
            let source = if let Some(w) = weighted {
                match self.select_weighted_source(&sources, w) {
                    Some(s) => {
                        weighted_source = s;
                        weighted_source.as_str()
                    }
                    None => break,
                }
            } else {
                cycle_sources[source_idx].as_str()
            };
            let (triplet, attempts_used) = self.sample_source_triplet_candidate(
                source,
                target_split,
                &mut recipe_orders,
                &mut recipe_positions,
                &mut rng,
                &mut seen,
                &mut seen_negatives,
            );
            recipe_steps = recipe_steps.saturating_add(attempts_used);
            if let Some((recipe, triplet)) = triplet {
                let SampleTriplet {
                    recipe: triplet_recipe_name,
                    anchor,
                    positive,
                    negative,
                    weight,
                    instruction,
                } = triplet;

                // Look up the anchor record's label to override structural defaults.
                let record_label = self
                    .records
                    .get(&RecordKey::of_chunk(&anchor))
                    .and_then(|r| r.label.clone());

                if pairs.len() < self.config.batch_size {
                    pairs.push(SamplePair {
                        recipe: triplet_recipe_name.clone(),
                        anchor: anchor.clone(),
                        positive: positive.clone(),
                        weight,
                        instruction: instruction.clone(),
                        label: record_label.clone().unwrap_or(PairLabel::Positive),
                        reason: None,
                    });
                }
                if pairs.len() < self.config.batch_size {
                    pairs.push(SamplePair {
                        recipe: triplet_recipe_name,
                        anchor,
                        positive: negative,
                        weight,
                        instruction,
                        label: record_label.unwrap_or(PairLabel::Negative),
                        reason: Some(strategy_reason(&recipe.negative_strategy).to_string()),
                    });
                }
            }
            source_idx += 1;
            source_steps += 1;
            if source_idx >= cycle_sources.len() {
                source_idx = 0;
                cycle = cycle.saturating_add(1);
                cycle_sources = self.shuffled_source_cycle(cycle);
            }
        }
        if recipe_steps > 0 {
            self.triplet_recipe_rr_idx = self.triplet_recipe_rr_idx.saturating_add(recipe_steps);
        }
        self.rng = rng;
        pad_with_reuse(&mut pairs, self.config.batch_size);
        if pairs.len() == self.config.batch_size {
            self.source_cycle_idx = self.source_cycle_idx.saturating_add(source_steps);
            self.source_state_dirty = sources.len() > 1;
            return Ok(SampleBatch { pairs });
        }
        Err(SamplerError::Exhausted(RECIPE_LABEL_TRIPLETS.into()))
    }

    /// Check whether `sample` passes both per-batch and per-window text dedup.
    /// The per-batch set is never cleared mid-batch (survives epoch advances),
    /// while `emitted_text_hashes` is cleared on epoch advance to allow fresh
    /// permutations.  Both must pass for the sample to be emitted.
    fn try_emit_text_sample(
        &mut self,
        sample: &TextSample,
        batch_texts: &mut HashSet<String>,
    ) -> bool {
        batch_texts.insert(sample.chunk.text.clone())
            && self
                .emitted_text_hashes
                .entry(RecordKey::of_chunk(&sample.chunk))
                .or_default()
                .insert(stable_hash_str(0, &sample.chunk.text))
    }

    fn next_text_batch_inner_with_weights(
        &mut self,
        target_split: SplitLabel,
        weights: Option<&HashMap<SourceId, f32>>,
    ) -> Result<TextBatch, SamplerError> {
        self.ingest_with_weights_fallback(target_split, weights)?;
        self.ensure_split_has_records(target_split)?;
        let sources = self.source_order.clone();
        if sources.is_empty() {
            if self.text_recipes.is_empty() {
                return Err(SamplerError::Configuration(
                    "no text recipes configured".into(),
                ));
            }
            let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
            let recipe_weights: Vec<f32> = self.text_recipes.iter().map(|r| r.weight).collect();
            let recipe_order = self.text_recipe_order_weighted_cycled(
                &recipe_weights,
                self.text_recipe_rr_idx,
                &mut rng,
            );
            let mut samples = Vec::new();
            let mut batch_texts = HashSet::new();
            let mut last_recipe_name = None;
            let mut recipe_pos = 0usize;
            let mut recipe_steps = 0usize;
            let attempts = self.config.batch_size * 4 * recipe_order.len().max(1);
            for _ in 0..attempts {
                if samples.len() >= self.config.batch_size {
                    break;
                }
                let recipe_idx = recipe_order[recipe_pos];
                recipe_pos = (recipe_pos + 1) % recipe_order.len();
                recipe_steps = recipe_steps.saturating_add(1);
                let recipe = self.text_recipes[recipe_idx].clone();
                last_recipe_name = Some(recipe.name.clone());
                if let Some(sample) =
                    self.make_text_sample_for_split(&recipe, None, target_split, &mut rng)
                    && self.try_emit_text_sample(&sample, &mut batch_texts)
                {
                    samples.push(sample);
                }
            }
            if recipe_steps > 0 {
                self.text_recipe_rr_idx = self.text_recipe_rr_idx.saturating_add(recipe_steps);
            }
            self.rng = rng;
            pad_with_reuse(&mut samples, self.config.batch_size);
            if samples.len() == self.config.batch_size {
                return Ok(TextBatch { samples });
            }
            return Err(SamplerError::Exhausted(
                last_recipe_name
                    .unwrap_or(Cow::Borrowed(RECIPE_LABEL_TEXT))
                    .to_string(),
            ));
        }

        let mut samples = Vec::new();
        let mut batch_texts = HashSet::new();
        let mut source_steps = 0usize;
        let mut cycle = (self.source_cycle_idx / sources.len()) as u64;
        let mut idx = self.source_cycle_idx % sources.len();
        let mut cycle_sources = self.shuffled_source_cycle(cycle);
        let mut recipe_orders: HashMap<RecipeKey, Vec<usize>> = HashMap::new();
        let mut recipe_positions: HashMap<RecipeKey, usize> = HashMap::new();
        let mut recipe_steps = 0usize;
        let max_recipe_len = sources
            .iter()
            .map(|source| self.text_recipes_for_source(source).len())
            .max()
            .unwrap_or(1)
            .max(1);
        let attempts = self.config.batch_size * 4 * sources.len() * max_recipe_len;
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        // Explicit non-uniform weights engage deficit round-robin source
        // choice; uniform/empty maps keep legacy uniform cycling bit-identical.
        let weighted: Option<&HashMap<SourceId, f32>> = match weights {
            Some(w) if Self::weights_are_selective(w) => Some(w),
            _ => None,
        };
        for _ in 0..attempts {
            if samples.len() >= self.config.batch_size {
                break;
            }
            let weighted_source;
            let source = if let Some(w) = weighted {
                match self.select_weighted_source(&sources, w) {
                    Some(s) => {
                        weighted_source = s;
                        weighted_source.as_str()
                    }
                    None => break,
                }
            } else {
                cycle_sources[idx].as_str()
            };
            let recipes = self.text_recipes_for_source(source).to_vec();
            if recipes.is_empty() {
                idx += 1;
                source_steps += 1;
                if idx >= cycle_sources.len() {
                    idx = 0;
                    cycle = cycle.saturating_add(1);
                    cycle_sources = self.shuffled_source_cycle(cycle);
                }
                continue;
            }
            if !recipe_orders.contains_key(source) {
                let recipe_weights: Vec<f32> = recipes.iter().map(|r| r.weight).collect();
                let order = self.text_recipe_order_weighted_cycled(
                    &recipe_weights,
                    self.text_recipe_rr_idx,
                    &mut rng,
                );
                recipe_orders.insert(source.to_string(), order);
            }
            let order = recipe_orders
                .get(source)
                .expect("recipe order missing for source");
            let pos = recipe_positions.entry(source.to_string()).or_insert(0);
            let mut sample: Option<(TextRecipe, TextSample)> = None;
            for offset in 0..order.len() {
                let recipe_idx = order[(*pos + offset) % order.len()];
                let recipe = recipes[recipe_idx].clone();
                if let Some(item) =
                    self.make_text_sample_for_split(&recipe, Some(source), target_split, &mut rng)
                    && self.try_emit_text_sample(&item, &mut batch_texts)
                {
                    recipe_steps = recipe_steps.saturating_add(offset + 1);
                    *pos = (*pos + offset + 1) % order.len();
                    sample = Some((recipe, item));
                    break;
                }
            }
            if sample.is_none() {
                recipe_steps = recipe_steps.saturating_add(order.len());
            }
            if let Some((_recipe, sample)) = sample {
                samples.push(sample);
            }
            idx += 1;
            source_steps += 1;
            if idx >= cycle_sources.len() {
                idx = 0;
                cycle = cycle.saturating_add(1);
                cycle_sources = self.shuffled_source_cycle(cycle);
            }
        }
        if samples.len() != self.config.batch_size {
            pad_with_reuse(&mut samples, self.config.batch_size);
        }
        if samples.len() != self.config.batch_size {
            self.rng = rng;
            return Err(SamplerError::Exhausted(RECIPE_LABEL_TEXT.into()));
        }
        self.rng = rng;
        self.source_cycle_idx = self.source_cycle_idx.saturating_add(source_steps);
        self.source_state_dirty = sources.len() > 1;
        if recipe_steps > 0 {
            self.text_recipe_rr_idx = self.text_recipe_rr_idx.saturating_add(recipe_steps);
        }
        Ok(TextBatch { samples })
    }

    fn next_triplet_batch_inner_with_weights(
        &mut self,
        target_split: SplitLabel,
        weights: Option<&HashMap<SourceId, f32>>,
    ) -> Result<TripletBatch, SamplerError> {
        self.ingest_with_weights_fallback(target_split, weights)?;
        self.ensure_split_has_records(target_split)?;
        let sources = self.source_order.clone();
        if sources.is_empty() {
            if self.triplet_recipes.is_empty() {
                return Err(SamplerError::Configuration(
                    "no triplet recipes configured".into(),
                ));
            }
            let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
            let recipe_weights: Vec<f32> = self.triplet_recipes.iter().map(|r| r.weight).collect();
            let recipe_order = self.recipe_order_weighted_cycled(
                &recipe_weights,
                self.triplet_recipe_rr_idx,
                &mut rng,
            );
            let mut triplets = Vec::new();
            let mut seen = HashSet::new();
            // Per-batch negative-text dedup (Option A, best-effort pre-pad).
            let mut seen_negatives: IntSet<u64> = IntSet::default();
            let mut last_recipe_name = None;
            let mut recipe_pos = 0usize;
            let mut recipe_steps = 0usize;
            let attempts = self.config.batch_size * 4 * recipe_order.len().max(1);
            for _ in 0..attempts {
                if triplets.len() >= self.config.batch_size {
                    break;
                }
                let Some(anchor) = self.choose_anchor_record(None, target_split) else {
                    break;
                };
                let mut triplet = None;
                for offset in 0..recipe_order.len() {
                    let idx = recipe_order[(recipe_pos + offset) % recipe_order.len()];
                    recipe_steps = recipe_steps.saturating_add(1);
                    let recipe = self.triplet_recipes[idx].clone();
                    last_recipe_name = Some(recipe.name.clone());
                    if let Some(sample) = self.make_triplet_with_unique_negative(
                        &recipe,
                        &anchor,
                        &mut rng,
                        &mut seen,
                        &mut seen_negatives,
                    ) {
                        triplet = Some(sample);
                        recipe_pos = (recipe_pos + offset + 1) % recipe_order.len();
                        break;
                    }
                }
                if let Some(triplet) = triplet {
                    triplets.push(triplet);
                }
            }
            if recipe_steps > 0 {
                self.triplet_recipe_rr_idx =
                    self.triplet_recipe_rr_idx.saturating_add(recipe_steps);
            }
            self.rng = rng;
            pad_with_reuse(&mut triplets, self.config.batch_size);
            if triplets.len() == self.config.batch_size {
                return Ok(TripletBatch { triplets });
            }
            return Err(SamplerError::Exhausted(
                last_recipe_name
                    .unwrap_or(Cow::Borrowed(RECIPE_LABEL_TRIPLETS))
                    .to_string(),
            ));
        }

        // ── Declarations for multi-source parallel path ────────────────────────
        let mut triplets: Vec<SampleTriplet> = Vec::new();
        let mut seen: HashSet<(RecordKey, RecordKey, RecordKey)> = HashSet::new();
        // Per-batch negative-text dedup (Option A, best-effort pre-pad).
        let mut seen_negatives: IntSet<u64> = IntSet::default();
        let mut source_steps = 0usize;
        let mut cycle = (self.source_cycle_idx / sources.len()) as u64;
        let mut source_idx = self.source_cycle_idx % sources.len();
        let mut cycle_sources = self.shuffled_source_cycle(cycle);
        // Explicit non-uniform weights engage deficit round-robin source
        // choice; uniform/empty maps keep legacy uniform cycling bit-identical.
        let weighted: Option<&HashMap<SourceId, f32>> = match weights {
            Some(w) if Self::weights_are_selective(w) => Some(w),
            _ => None,
        };
        let mut recipe_steps = 0usize;
        let max_recipe_len = sources
            .iter()
            .map(|source| self.triplet_recipe_count_for_source(source.as_str()))
            .max()
            .unwrap_or(1)
            .max(1);
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));

        // ── Phase 1: sequential anchor pre-assignment ──────────────────────────
        // Advance cursors and pick anchors deterministically. Capture a fork seed
        // for each slot so that Phase 2 can reconstruct independent rngs in
        // deterministic order.
        struct SlotPlan {
            anchor: Arc<DataRecord>,
            recipes: Vec<TripletRecipe>,
            fork_seed: u64,
        }
        let target_slots = self.config.batch_size * 4 * max_recipe_len;
        let mut slot_plans: Vec<SlotPlan> = Vec::with_capacity(target_slots);

        for _ in 0..target_slots {
            if slot_plans.len() >= target_slots {
                break;
            }
            let weighted_source;
            let source = if let Some(w) = weighted {
                match self.select_weighted_source(&sources, w) {
                    Some(s) => {
                        weighted_source = s;
                        weighted_source.as_str()
                    }
                    None => break,
                }
            } else {
                cycle_sources[source_idx].as_str()
            };
            let (recipes, _) = self.resolve_source_triplet_plan(source);
            if !recipes.is_empty() {
                let fork_seed = rng.next_u64();
                if let Some(anchor) = self.choose_anchor_record(Some(source), target_split) {
                    slot_plans.push(SlotPlan {
                        anchor,
                        recipes,
                        fork_seed,
                    });
                    recipe_steps = recipe_steps.saturating_add(1);
                }
            }
            source_idx += 1;
            source_steps += 1;
            if source_idx >= cycle_sources.len() {
                source_idx = 0;
                cycle = cycle.saturating_add(1);
                cycle_sources = self.shuffled_source_cycle(cycle);
            }
        }
        self.rng = rng;

        // ── Phase 2: parallel anchor+positive selection ────────────────────────
        // Only chunk selection runs in parallel; negative selection and the
        // coin-flip swap remain sequential in Phase 3 so that the BM25 cursor
        // rotation (shared RwLock state) is deterministic across runs.
        struct SlotCandidate {
            recipe: TripletRecipe,
            anchor: Arc<DataRecord>,
            anchor_chunk: RecordChunk,
            positive_chunk: RecordChunk,
            anchor_raw_text: String,
        }
        let mut raw_candidates: Vec<(usize, Option<SlotCandidate>)> = slot_plans
            .par_iter()
            .enumerate()
            .map(|(slot_idx, plan)| {
                let mut fork_rng = DeterministicRng::new(plan.fork_seed);
                // Use weighted shuffle so recipes with higher weight are tried more
                // often and zero-weight recipes are excluded entirely.
                let weights: Vec<f32> = plan.recipes.iter().map(|r| r.weight).collect();
                let order = weighted_recipe_order(&weights, &mut fork_rng);
                let mut candidate = None;
                for &idx in &order {
                    let recipe = &plan.recipes[idx];
                    if let Some((ac, pc, raw)) =
                        self.select_anchor_positive_for_recipe(recipe, &plan.anchor, &mut fork_rng)
                    {
                        candidate = Some(SlotCandidate {
                            recipe: recipe.clone(),
                            anchor: Arc::clone(&plan.anchor),
                            anchor_chunk: ac,
                            positive_chunk: pc,
                            anchor_raw_text: raw,
                        });
                        break;
                    }
                }
                (slot_idx, candidate)
            })
            .collect();

        // ── Phase 3: sequential finalization (negative selection + coin-flip) ───
        // Sort by slot_idx to restore deterministic ordering, then finalize each
        // candidate sequentially so that BM25 cursor rotation advances in a fixed
        // order regardless of rayon's thread scheduling.
        raw_candidates.sort_unstable_by_key(|(i, _)| *i);
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        for (_, candidate) in raw_candidates {
            if triplets.len() >= self.config.batch_size {
                break;
            }
            let Some(sc) = candidate else { continue };
            // Hoisted 50 % coin-flip swap (single RNG draw) + guard (A).
            // Unrecoverable a==p pairs skip before any negative work.
            let (final_anchor, final_positive) = if rng.next_u64() & ANCHOR_POSITIVE_SWAP_MASK == 0
            {
                (sc.positive_chunk, sc.anchor_chunk)
            } else {
                (sc.anchor_chunk, sc.positive_chunk)
            };
            if !sc.recipe.allow_same_anchor_positive && final_anchor.text == final_positive.text {
                continue;
            }
            // Negative-only retries: references only, no anchor/positive clones.
            // (B) checks inside loop are all negative-resolvable.
            let mut accepted: Option<(RecordChunk, f32, String)> = None;
            for _ in 0..NEG_DEDUP_RETRIES {
                let (negative_record, fallback_used) = match self.select_negative_record(
                    &sc.anchor,
                    &sc.recipe.negative_strategy,
                    Some(&sc.anchor_raw_text),
                    &mut rng,
                ) {
                    Some(pair) => pair,
                    None => break,
                };
                let mut negative_chunk = match self.select_chunk_parallel(
                    &negative_record,
                    &sc.recipe.negative_selector,
                    &mut rng,
                ) {
                    Some(c) => c,
                    None => continue,
                };
                self.decorate_chunk_parallel(&negative_record, &mut negative_chunk, &mut rng);
                if negative_chunk.text == final_positive.text
                    || negative_chunk.text == final_anchor.text
                {
                    continue;
                }
                let key = (
                    RecordKey::of_chunk(&final_anchor),
                    RecordKey::of_chunk(&final_positive),
                    RecordKey::of_chunk(&negative_chunk),
                );
                let neg_h = negative_text_hash(&negative_chunk);
                // Atomic check-then-commit: no partial `seen` mutation on reject.
                if seen.contains(&key) || seen_negatives.contains(&neg_h) {
                    continue;
                }
                let chunk_weight = self.triplet_chunk_weight(
                    &sc.recipe,
                    &final_anchor,
                    &final_positive,
                    &negative_chunk,
                );
                let weight = sc.recipe.weight * chunk_weight;
                let recipe_name = if fallback_used {
                    format!("{}_fallback_same_split", sc.recipe.name)
                } else {
                    sc.recipe.name.to_string()
                };
                seen.insert(key);
                seen_negatives.insert(neg_h);
                accepted = Some((negative_chunk, weight, recipe_name));
                break;
            }
            let Some((negative_chunk, weight, recipe_name)) = accepted else {
                continue;
            };
            let triplet = SampleTriplet {
                recipe: recipe_name,
                anchor: final_anchor,
                positive: final_positive,
                negative: negative_chunk,
                weight,
                instruction: sc.recipe.instruction.as_ref().map(|s| s.to_string()),
            };
            if triplets.len() < self.config.batch_size {
                triplets.push(triplet);
            }
        }
        self.rng = rng;

        if recipe_steps > 0 {
            self.triplet_recipe_rr_idx = self.triplet_recipe_rr_idx.saturating_add(recipe_steps);
        }
        pad_with_reuse(&mut triplets, self.config.batch_size);
        if triplets.len() == self.config.batch_size {
            self.source_cycle_idx = self.source_cycle_idx.saturating_add(source_steps);
            self.source_state_dirty = sources.len() > 1;
            let batch = TripletBatch { triplets };
            return Ok(batch);
        }
        Err(SamplerError::Exhausted(RECIPE_LABEL_TRIPLETS.into()))
    }

    // ── test-only helpers ─────────────────────────────────────────────────────

    /// Return a mutable reference to the concrete [`backends::Bm25Backend`], panicking
    /// when the feature is disabled (which won't happen — call sites are
    /// equally gated on `#[cfg(feature = "bm25-mining")]`).
    #[cfg(test)]
    #[cfg(feature = "bm25-mining")]
    fn bm25_backend_mut(&mut self) -> &mut backends::Bm25Backend {
        self.negative_backend
            .as_any_mut()
            .downcast_mut::<backends::Bm25Backend>()
            .expect("bm25_backend_mut: negative_backend is Bm25Backend when bm25-mining feature is active")
    }

    /// Test shim: `recipe_order_weighted_shuffled` using the inner rng.
    #[cfg(test)]
    fn recipe_order_weighted_shuffled_seeded(&mut self, weights: &[f32]) -> Vec<usize> {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        let result = self.recipe_order_weighted_shuffled(weights, &mut rng);
        self.rng = rng;
        result
    }

    /// Test shim: `recipe_order_weighted_cycled` using the inner rng.
    #[cfg(test)]
    fn recipe_order_weighted_cycled_seeded(
        &mut self,
        weights: &[f32],
        rr_idx: usize,
    ) -> Vec<usize> {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        let result = self.recipe_order_weighted_cycled(weights, rr_idx, &mut rng);
        self.rng = rng;
        result
    }

    /// Test shim: `text_recipe_order_weighted_shuffled` using the inner rng.
    #[cfg(test)]
    fn text_recipe_order_weighted_shuffled_seeded(&mut self, weights: &[f32]) -> Vec<usize> {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        let result = self.text_recipe_order_weighted_shuffled(weights, &mut rng);
        self.rng = rng;
        result
    }

    /// Test shim: `text_recipe_order_weighted_cycled` using the inner rng.
    #[cfg(test)]
    fn text_recipe_order_weighted_cycled_seeded(
        &mut self,
        weights: &[f32],
        rr_idx: usize,
    ) -> Vec<usize> {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        let result = self.text_recipe_order_weighted_cycled(weights, rr_idx, &mut rng);
        self.rng = rng;
        result
    }

    /// Test shim: `select_negative_record` using the inner rng.
    #[cfg(test)]
    fn select_negative_record_seeded(
        &mut self,
        anchor_record: &DataRecord,
        strategy: &NegativeStrategy,
        anchor_query_text: Option<&str>,
    ) -> Option<(Arc<DataRecord>, bool)> {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        let result =
            self.select_negative_record(anchor_record, strategy, anchor_query_text, &mut rng);
        self.rng = rng;
        result
    }

    /// Test shim: `make_triplet_with_unique_negative` using the inner rng and
    /// fresh per-batch dedup sets (single-candidate semantics).
    #[cfg(test)]
    fn make_triplet_with_unique_negative_seeded(
        &mut self,
        recipe: &TripletRecipe,
        anchor: &DataRecord,
    ) -> Option<SampleTriplet> {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        let mut seen = HashSet::new();
        let mut seen_negatives: IntSet<u64> = IntSet::default();
        let result = self.make_triplet_with_unique_negative(
            recipe,
            anchor,
            &mut rng,
            &mut seen,
            &mut seen_negatives,
        );
        self.rng = rng;
        result
    }

    /// Test shim: `decorate_chunk` using the inner rng.
    #[cfg(test)]
    fn decorate_chunk_seeded(&mut self, record: &DataRecord, chunk: &mut RecordChunk) {
        let mut rng = std::mem::replace(&mut self.rng, DeterministicRng::new(0));
        self.decorate_chunk(record, chunk, &mut rng);
        self.rng = rng;
    }

    // ── extended-metrics helpers ───────────────────────────────────────────────

    /// Return cumulative `(fallback_count, selection_count)` from the BM25
    /// backend since it was created.  Only available when both `bm25-mining`
    /// and `extended-metrics` features are active.
    #[cfg(all(feature = "bm25-mining", feature = "extended-metrics"))]
    fn bm25_fallback_stats(&self) -> (u64, u64) {
        self.negative_backend.bm25_fallback_stats()
    }

    /// Look up `anchor`'s split label and return BM25-ranked candidate IDs.
    ///
    /// Mirrors the old `bm25_ranked_candidates` method that tests called on
    /// `TripletSamplerInner` before BM25 state moved into `Bm25Backend`.
    #[cfg(test)]
    #[cfg(feature = "bm25-mining")]
    fn bm25_ranked_candidates(&mut self, anchor: &crate::data::DataRecord) -> Vec<RecordKey> {
        let split = self
            .split_store
            .label_for(&RecordKey::from(anchor))
            .unwrap_or(SplitLabel::Train);
        self.negative_backend
            .as_any_mut()
            .downcast_mut::<backends::Bm25Backend>()
            .expect("bm25_ranked_candidates: Bm25Backend")
            .ranked_candidates_pub(anchor, split)
    }
}

/// Build a shuffled selection order from a slice of recipe weights.
///
/// Each recipe with `weight > 0.0` receives a number of slots proportional to its weight
/// relative to the smallest positive weight, capped at
/// [`RECIPE_ORDER_MAX_WEIGHT_MULTIPLIER`].  The resulting expanded slot list is then
/// shuffled, giving each recipe approximately the requested frequency in round-robin
/// cycling.
///
/// This is a free function (no `self`) so it can be called from rayon parallel closures as
/// well as from within the `TripletSamplerInner` instance methods.
fn weighted_recipe_order(weights: &[f32], rng: &mut DeterministicRng) -> Vec<usize> {
    let nonzero: Vec<(usize, f32)> = weights
        .iter()
        .enumerate()
        .filter(|(_, w)| **w > 0.0)
        .map(|(i, &w)| (i, w))
        .collect();
    if nonzero.is_empty() {
        return Vec::new();
    }
    let w_min = nonzero
        .iter()
        .map(|(_, w)| *w)
        .fold(f32::INFINITY, f32::min);
    let mut order: Vec<usize> = Vec::new();
    for (recipe_idx, w) in &nonzero {
        let tickets = ((w / w_min).round() as usize).clamp(1, RECIPE_ORDER_MAX_WEIGHT_MULTIPLIER);
        for _ in 0..tickets {
            order.push(*recipe_idx);
        }
    }
    order.shuffle(rng);
    order
}

fn same_selector_pair_is_valid(
    anchor_chunk: &RecordChunk,
    positive_chunk: &RecordChunk,
    enforce_window_pair: bool,
) -> bool {
    if triplet_chunk_key(anchor_chunk) == triplet_chunk_key(positive_chunk) {
        return false;
    }
    if !enforce_window_pair {
        return true;
    }
    matches!(
        (&anchor_chunk.view, &positive_chunk.view),
        (ChunkView::Window { .. }, ChunkView::Window { .. })
    )
}

impl<S: SplitStore + EpochStateStore + SamplerStateStore + 'static> TripletSampler<S> {
    /// Create a sampler from config and a split-state backend.
    pub fn new(config: SamplerConfig, split_store: Arc<S>) -> Self {
        let inner = TripletSamplerInner::new(config, split_store);
        Self {
            inner: Mutex::new(inner),
        }
    }

    /// Create a sampler from config with a custom chunking implementation.
    pub fn new_with_chunker(
        config: SamplerConfig,
        split_store: Arc<S>,
        chunker: Arc<dyn ChunkingAlgorithm>,
    ) -> Self {
        let inner = TripletSamplerInner::new_with_chunker(config, split_store, chunker);
        Self {
            inner: Mutex::new(inner),
        }
    }

    /// Return an unweighted pair batch for `split`.
    ///
    /// Samples all sources uniformly. Prefer
    /// [`Self::next_pair_batch_with_weights_for_split`] with an explicit
    /// per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call next_pair_batch_with_weights_for_split(split, &weights) with an explicit per-source weight map to honor a data mixture"
    )]
    pub fn next_pair_batch_for_split(
        &self,
        split: SplitLabel,
    ) -> Result<SampleBatch, SamplerError> {
        self.next_pair_batch_with_weights_for_split(split, &HashMap::new())
    }

    /// Return an unweighted text batch for `split`.
    ///
    /// Samples all sources uniformly. Prefer
    /// [`Self::next_text_batch_with_weights_for_split`] with an explicit
    /// per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call next_text_batch_with_weights_for_split(split, &weights) with an explicit per-source weight map to honor a data mixture"
    )]
    pub fn next_text_batch_for_split(&self, split: SplitLabel) -> Result<TextBatch, SamplerError> {
        self.next_text_batch_with_weights_for_split(split, &HashMap::new())
    }

    /// Return an unweighted triplet batch for `split`.
    ///
    /// Samples all sources uniformly. Prefer
    /// [`Self::next_triplet_batch_with_weights_for_split`] with an explicit
    /// per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call next_triplet_batch_with_weights_for_split(split, &weights) with an explicit per-source weight map to honor a data mixture"
    )]
    pub fn next_triplet_batch_for_split(
        &self,
        split: SplitLabel,
    ) -> Result<TripletBatch, SamplerError> {
        self.next_triplet_batch_with_weights_for_split(split, &HashMap::new())
    }

    /// Shared retry loop for all three batch types: load state, bump step,
    /// retry on exhaustion up to EXHAUSTION_RETRY_LIMIT times, and return
    /// the batch or propagate the final error.
    fn next_batch_with_retry<B>(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
        inner_fn: impl Fn(
            &mut TripletSamplerInner<S>,
            SplitLabel,
            Option<&HashMap<SourceId, f32>>,
        ) -> Result<B, SamplerError>,
    ) -> Result<B, SamplerError> {
        let mut inner = self.inner.lock().unwrap();
        inner.ensure_split_allowed(split)?;
        // Validate weight keys up front, on every path: a typo'd source id
        // must fail loudly, never silently sample as if unweighted.
        if !weights.is_empty() {
            inner.ingestion.validate_weights(weights)?;
        }
        // Load persisted state BEFORE bumping step so the restored step
        // counter is reflected before we increment it for this call.
        inner.ensure_ingestion_cursors_loaded()?;
        // Bump epoch step once per public call, regardless of success/exhaustion.
        inner.ingestion.increment_epoch_step();
        for attempt in 0..=EXHAUSTION_RETRY_LIMIT {
            match inner_fn(&mut inner, split, Some(weights)) {
                Ok(batch) => return Ok(batch),
                Err(SamplerError::Exhausted(_)) if attempt < EXHAUSTION_RETRY_LIMIT => {
                    inner.force_ingest_refresh_with_weights_for_split(split, weights)?;
                }
                Err(err) => return Err(err),
            }
        }
        // The for-loop always returns via the match arms above.
        unreachable!()
    }

    /// Return a weighted pair batch for `split` using per-source weights.
    ///
    /// Source choice is deficit round-robin over the map: deterministic and
    /// exact over each weight cycle (0.75/0.25 yields exactly 3:1 anchors).
    /// Unlisted sources get quantum 1.0; non-finite/negative quanta count as
    /// 0. Empty or all-equal maps keep legacy uniform cycling bit-identically.
    /// Weights also steer ingestion; see `ingest_with_weights_fallback`.
    pub fn next_pair_batch_with_weights_for_split(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<SampleBatch, SamplerError> {
        self.next_batch_with_retry(split, weights, |inner, split, weights| {
            inner.next_pair_batch_inner_with_weights(split, weights)
        })
    }

    /// Return a weighted text batch for `split` using per-source weights.
    ///
    /// Source choice is deficit round-robin over the map: deterministic and
    /// exact over each weight cycle. Unlisted sources get quantum 1.0;
    /// non-finite/negative quanta count as 0. Empty or all-equal maps keep
    /// legacy uniform cycling bit-identically.
    pub fn next_text_batch_with_weights_for_split(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<TextBatch, SamplerError> {
        self.next_batch_with_retry(split, weights, |inner, split, weights| {
            inner.next_text_batch_inner_with_weights(split, weights)
        })
    }

    /// Return a weighted triplet batch for `split` using per-source weights.
    ///
    /// Source choice is deficit round-robin over the map: deterministic and
    /// exact over each weight cycle. Unlisted sources get quantum 1.0;
    /// non-finite/negative quanta count as 0. Empty or all-equal maps keep
    /// legacy uniform cycling bit-identically.
    pub fn next_triplet_batch_with_weights_for_split(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<TripletBatch, SamplerError> {
        self.next_batch_with_retry(split, weights, |inner, split, weights| {
            inner.next_triplet_batch_inner_with_weights(split, weights)
        })
    }

    /// Spawn a background prefetcher for triplet batches.
    ///
    /// Samples all sources uniformly. Prefer
    /// [`Self::prefetch_triplet_batches_with_weights`] with an explicit
    /// per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call prefetch_triplet_batches_with_weights(split, capacity, weights) with an explicit per-source weight map to honor a data mixture"
    )]
    pub fn prefetch_triplet_batches(
        self: Arc<Self>,
        split: SplitLabel,
        capacity: usize,
    ) -> BatchPrefetcher<TripletBatch> {
        BatchPrefetcher::new(capacity, move || {
            self.next_triplet_batch_with_weights_for_split(split, &HashMap::new())
        })
    }

    /// Spawn a background prefetcher for weighted triplet batches.
    pub fn prefetch_triplet_batches_with_weights(
        self: Arc<Self>,
        split: SplitLabel,
        capacity: usize,
        weights: HashMap<SourceId, f32>,
    ) -> BatchPrefetcher<TripletBatch> {
        BatchPrefetcher::new(capacity, move || {
            self.next_triplet_batch_with_weights_for_split(split, &weights)
        })
    }

    /// Spawn a background prefetcher for pair batches.
    ///
    /// Samples all sources uniformly. Prefer
    /// [`Self::prefetch_pair_batches_with_weights`] with an explicit
    /// per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call prefetch_pair_batches_with_weights(split, capacity, weights) with an explicit per-source weight map to honor a data mixture"
    )]
    pub fn prefetch_pair_batches(
        self: Arc<Self>,
        split: SplitLabel,
        capacity: usize,
    ) -> BatchPrefetcher<SampleBatch> {
        BatchPrefetcher::new(capacity, move || {
            self.next_pair_batch_with_weights_for_split(split, &HashMap::new())
        })
    }

    /// Spawn a background prefetcher for weighted pair batches.
    pub fn prefetch_pair_batches_with_weights(
        self: Arc<Self>,
        split: SplitLabel,
        capacity: usize,
        weights: HashMap<SourceId, f32>,
    ) -> BatchPrefetcher<SampleBatch> {
        BatchPrefetcher::new(capacity, move || {
            self.next_pair_batch_with_weights_for_split(split, &weights)
        })
    }

    /// Spawn a background prefetcher for text batches.
    ///
    /// Samples all sources uniformly. Prefer
    /// [`Self::prefetch_text_batches_with_weights`] with an explicit
    /// per-source weight map to honor a data mixture.
    #[deprecated(
        note = "samples all sources uniformly; call prefetch_text_batches_with_weights(split, capacity, weights) with an explicit per-source weight map to honor a data mixture"
    )]
    pub fn prefetch_text_batches(
        self: Arc<Self>,
        split: SplitLabel,
        capacity: usize,
    ) -> BatchPrefetcher<TextBatch> {
        BatchPrefetcher::new(capacity, move || {
            self.next_text_batch_with_weights_for_split(split, &HashMap::new())
        })
    }

    /// Spawn a background prefetcher for weighted text batches.
    pub fn prefetch_text_batches_with_weights(
        self: Arc<Self>,
        split: SplitLabel,
        capacity: usize,
        weights: HashMap<SourceId, f32>,
    ) -> BatchPrefetcher<TextBatch> {
        BatchPrefetcher::new(capacity, move || {
            self.next_text_batch_with_weights_for_split(split, &weights)
        })
    }

    /// Return the currently active text recipes.
    pub fn text_recipes(&self) -> Vec<TextRecipe> {
        let inner = self.inner.lock().unwrap();
        inner.text_recipes().to_vec()
    }

    /// Register a data source for ingestion and sampling.
    ///
    /// Returns an error if the source's `id()` matches the reserved `__*__`
    /// pattern used for internal synthetic/metadata source identifiers.
    pub fn register_source(
        &self,
        source: Box<dyn DataSource + 'static>,
    ) -> Result<(), SamplerError> {
        let mut inner = self.inner.lock().unwrap();
        inner.register_source(source)
    }

    /// Force sampler epoch to `epoch` (advanced deterministic replay control).
    pub fn set_epoch(&self, epoch: u64) -> Result<(), SamplerError> {
        let mut inner = self.inner.lock().unwrap();
        inner.set_epoch(epoch)
    }

    /// Persist sampler and split runtime state for restart-resume.
    ///
    /// When `save_to` is `Some(path)`, current persisted runtime state is also
    /// mirrored to `path` when supported by the split-store backend.
    pub fn save_sampler_state(&self, save_to: Option<&Path>) -> Result<(), SamplerError> {
        let mut inner = self.inner.lock().unwrap();
        inner.save_sampler_state(save_to)
    }

    /// Return cumulative `(fallback_count, selection_count)` from the BM25
    /// backend since it was created.  Only available when both `bm25-mining`
    /// and `extended-metrics` features are active.
    ///
    /// `selection_count` counts calls where the pool was non-empty.
    /// `fallback_count` counts the subset where BM25 produced no candidates
    /// intersecting the pool and random selection was used instead.
    #[cfg(all(feature = "bm25-mining", feature = "extended-metrics"))]
    pub fn bm25_fallback_stats(&self) -> (u64, u64) {
        let inner = self.inner.lock().unwrap();
        inner.bm25_fallback_stats()
    }
}

impl<S: SplitStore + EpochStateStore + SamplerStateStore + 'static> Sampler for TripletSampler<S> {
    fn next_pair_batch(&self, split: SplitLabel) -> Result<SampleBatch, SamplerError> {
        self.next_pair_batch_with_weights_for_split(split, &HashMap::new())
    }

    fn next_pair_batch_with_weights(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<SampleBatch, SamplerError> {
        self.next_pair_batch_with_weights_for_split(split, weights)
    }

    fn next_text_batch(&self, split: SplitLabel) -> Result<TextBatch, SamplerError> {
        self.next_text_batch_with_weights_for_split(split, &HashMap::new())
    }

    fn next_text_batch_with_weights(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<TextBatch, SamplerError> {
        self.next_text_batch_with_weights_for_split(split, weights)
    }

    fn next_triplet_batch(&self, split: SplitLabel) -> Result<TripletBatch, SamplerError> {
        self.next_triplet_batch_with_weights_for_split(split, &HashMap::new())
    }

    fn next_triplet_batch_with_weights(
        &self,
        split: SplitLabel,
        weights: &HashMap<SourceId, f32>,
    ) -> Result<TripletBatch, SamplerError> {
        self.next_triplet_batch_with_weights_for_split(split, weights)
    }

    fn save_sampler_state(&self, save_to: Option<&std::path::Path>) -> Result<(), SamplerError> {
        TripletSampler::save_sampler_state(self, save_to)
    }
}

fn roles_match(target: &SectionRole, candidate: &SectionRole) -> bool {
    target == candidate
}

fn role_cursor_key(record_key: &RecordKey, role: &SectionRole) -> (RecordKey, String) {
    (record_key.clone(), role_label(role))
}

fn role_label(role: &SectionRole) -> String {
    match role {
        SectionRole::Anchor => ROLE_LABEL_ANCHOR.into(),
        SectionRole::Context => ROLE_LABEL_CONTEXT.into(),
    }
}

fn taxonomy_value(record: &DataRecord, field: MetadataKey) -> Option<&str> {
    record.taxonomy.iter().find_map(|entry| field.strip(entry))
}

fn strategy_reason(strategy: &NegativeStrategy) -> &'static str {
    match strategy {
        NegativeStrategy::WrongPublicationDate => NEG_REASON_WRONG_DATE,
        NegativeStrategy::WrongArticle => NEG_REASON_WRONG_ARTICLE,
        NegativeStrategy::QuestionAnswerMismatch => NEG_REASON_WRONG_QA,
        NegativeStrategy::SameRecord => NEG_REASON_SAME_RECORD,
    }
}

/// Dedup key for text batch generation: (record_id, text).
///
/// Unlike `triplet_chunk_key` which compares structural identity
/// (record_id, section_idx, window_index), this key compares actual
/// text content. This prevents text-columns mode sources (where Anchor
/// and Context sections carry identical text) from producing duplicate
/// samples from the same record.
#[cfg(test)]
fn text_dedup_key(chunk: &RecordChunk) -> (String, String) {
    (chunk.record_id.clone(), chunk.text.clone())
}

fn triplet_chunk_key(chunk: &RecordChunk) -> String {
    match &chunk.view {
        ChunkView::Window { index, .. } => {
            format!("{}|{}|w|{}", chunk.record_id, chunk.section_idx, index)
        }
        ChunkView::SummaryFallback { strategy, .. } => {
            format!("{}|{}|s|{}", chunk.record_id, chunk.section_idx, strategy)
        }
    }
}

/// Bounded negative-redraw attempts per anchor-positive pair.
///
/// When a candidate negative collides with an already-accepted negative in
/// the current batch (or fails the intra-triplet text guard), only the
/// negative is redrawn — the anchor/positive pair is kept. After this many
/// failures the slot is skipped; the outer `attempts` budget then moves on.
const NEG_DEDUP_RETRIES: usize = 4;

/// Pure `u64` text hash for per-batch negative dedup (Option A: negative-vs-negative).
///
/// InfoNCE-style objectives compare text content, so the key must be text-only:
/// no `RecordKey` coupling (same text from different records/windows must still
/// collide) and no `String` allocation in the hot loop.
#[inline]
fn negative_text_hash(chunk: &RecordChunk) -> u64 {
    stable_hash_str(0, &chunk.text)
}

fn pad_with_reuse<T: Clone>(items: &mut Vec<T>, target: usize) {
    if items.is_empty() || items.len() >= target {
        return;
    }
    let seed = items.clone();
    let base_len = seed.len();
    for idx in 0..(target - items.len()) {
        items.push(seed[idx % base_len].clone());
    }
}

#[cfg(test)]
mod tests;