weir/fixed_point_scratch/

buffers.rs

use super::FrontierDensityTelemetry;

/// Cached CUDA-oriented changed-flag Program for one fixed-point graph shape.
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct FixedPointForwardChangedProgramCache {
    pub(crate) kind: crate::fixed_point_graph::FixedPointAnalysisKind,
    pub(crate) node_count: u32,
    pub(crate) edge_count: u32,
    pub(crate) program: vyre::ir::Program,
}

/// Caller-owned scratch for Weir fixed-point GPU wrappers.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct FixedPointScratch {
    pub(crate) frontier_words: Vec<u32>,
    pub(crate) frontier_bytes: Vec<u8>,
    pub(crate) previous_frontier_bytes: Vec<u8>,
    pub(crate) changed_bytes: Vec<u8>,
    pub(crate) static_input_bytes: Vec<Vec<u8>>,
    pub(crate) outputs: Vec<Vec<u8>>,
    pub(crate) next: Vec<u32>,
    pub(crate) resources: Vec<vyre::backend::Resource>,
    pub(crate) alternate_resources: Vec<vyre::backend::Resource>,
    pub(crate) changed_resources: Vec<vyre::backend::Resource>,
    pub(crate) clear_changed_program: Option<vyre::ir::Program>,
    pub(crate) forward_changed_program: Option<FixedPointForwardChangedProgramCache>,
    pub(crate) edge_kind_masks: Vec<u32>,
    pub(crate) reverse_indegree: Vec<usize>,
    pub(crate) reverse_cursor: Vec<usize>,
    pub(crate) reverse_offsets: Vec<u32>,
    pub(crate) reverse_targets: Vec<u32>,
    pub(crate) reverse_masks: Vec<u32>,
    pub(crate) frontier_density: FrontierDensityTelemetry,
    pub(crate) dispatch_config: vyre::DispatchConfig,
    #[cfg(feature = "gpu-telemetry")]
    pub(crate) telemetry_popcount_input_bytes: Vec<u8>,
    #[cfg(feature = "gpu-telemetry")]
    pub(crate) telemetry_popcount_output_bytes: Vec<u8>,
    #[cfg(feature = "gpu-telemetry")]
    pub(crate) telemetry_sum_bytes: Vec<u8>,
    #[cfg(feature = "gpu-telemetry")]
    pub(crate) telemetry_xor_words: Vec<u32>,
    #[cfg(feature = "gpu-telemetry")]
    pub(crate) telemetry_popcount_program: Option<(u32, vyre::ir::Program)>,
    #[cfg(feature = "gpu-telemetry")]
    pub(crate) telemetry_reduce_program: Option<(u32, vyre::ir::Program)>,
}

impl FixedPointScratch {
    /// Allocate reusable fixed-point scratch with enough retained capacity for
    /// a known hot loop shape.
    ///
    /// `frontier_words` sizes both the current and next frontier word buffers.
    /// The byte frontier buffer is pre-sized to the exact little-endian byte
    /// representation capacity used by dispatch wrappers. `output_slots`
    /// preallocates reusable backend output vectors, and `resource_slots`
    /// preallocates the dispatch resource list plus retained static input
    /// byte slots for invariant graph buffers.
    pub fn with_capacities(
        frontier_words: usize,
        output_slots: usize,
        resource_slots: usize,
    ) -> Result<Self, String> {
        let frontier_bytes = frontier_words
            .checked_mul(std::mem::size_of::<u32>())
            .ok_or_else(|| {
                "weir fixed-point scratch frontier byte capacity overflowed usize. Fix: lower frontier_words before allocating reusable scratch."
                    .to_string()
            })?;
        let mut scratch = Self {
            frontier_words: Vec::new(),
            frontier_bytes: Vec::new(),
            previous_frontier_bytes: Vec::new(),
            changed_bytes: Vec::new(),
            static_input_bytes: Vec::new(),
            outputs: Vec::new(),
            next: Vec::new(),
            resources: Vec::new(),
            alternate_resources: Vec::new(),
            changed_resources: Vec::new(),
            clear_changed_program: None,
            forward_changed_program: None,
            edge_kind_masks: Vec::new(),
            reverse_indegree: Vec::new(),
            reverse_cursor: Vec::new(),
            reverse_offsets: Vec::new(),
            reverse_targets: Vec::new(),
            reverse_masks: Vec::new(),
            frontier_density: FrontierDensityTelemetry::default(),
            dispatch_config: vyre::DispatchConfig::default(),
            #[cfg(feature = "gpu-telemetry")]
            telemetry_popcount_input_bytes: Vec::new(),
            #[cfg(feature = "gpu-telemetry")]
            telemetry_popcount_output_bytes: Vec::new(),
            #[cfg(feature = "gpu-telemetry")]
            telemetry_sum_bytes: Vec::new(),
            #[cfg(feature = "gpu-telemetry")]
            telemetry_xor_words: Vec::new(),
            #[cfg(feature = "gpu-telemetry")]
            telemetry_popcount_program: None,
            #[cfg(feature = "gpu-telemetry")]
            telemetry_reduce_program: None,
        };
        crate::staging_reserve::reserve_vec(
            &mut scratch.frontier_words,
            frontier_words,
            "fixed-point frontier word",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.frontier_bytes,
            frontier_bytes,
            "fixed-point frontier byte",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.previous_frontier_bytes,
            frontier_bytes,
            "fixed-point previous frontier byte",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.changed_bytes,
            std::mem::size_of::<u32>(),
            "fixed-point changed flag byte",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.static_input_bytes,
            resource_slots,
            "fixed-point static input byte slot",
        )?;
        crate::staging_reserve::resize_vec_slots(
            &mut scratch.outputs,
            output_slots,
            "fixed-point output slot",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.next,
            frontier_words,
            "fixed-point next frontier word",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.resources,
            resource_slots,
            "fixed-point resident resource",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.alternate_resources,
            resource_slots,
            "fixed-point alternate resident resource",
        )?;
        crate::staging_reserve::reserve_vec(
            &mut scratch.changed_resources,
            resource_slots,
            "fixed-point changed resident resource",
        )?;
        #[cfg(feature = "gpu-telemetry")]
        {
            crate::staging_reserve::reserve_vec(
                &mut scratch.telemetry_popcount_input_bytes,
                frontier_bytes,
                "telemetry popcount input byte",
            )?;
            crate::staging_reserve::reserve_vec(
                &mut scratch.telemetry_popcount_output_bytes,
                frontier_bytes,
                "telemetry popcount output byte",
            )?;
            crate::staging_reserve::reserve_vec(
                &mut scratch.telemetry_sum_bytes,
                std::mem::size_of::<u32>(),
                "telemetry reduce sum byte",
            )?;
            crate::staging_reserve::reserve_vec(
                &mut scratch.telemetry_xor_words,
                frontier_words,
                "telemetry xor word",
            )?;
        }
        Ok(scratch)
    }

    /// Drop logical contents while preserving allocation capacity.
    pub fn clear(&mut self) {
        self.frontier_words.clear();
        self.frontier_bytes.clear();
        self.previous_frontier_bytes.clear();
        self.changed_bytes.clear();
        crate::output_scratch::clear_outputs_preserving_slots(&mut self.static_input_bytes);
        crate::output_scratch::clear_outputs_preserving_slots(&mut self.outputs);
        self.next.clear();
        self.resources.clear();
        self.alternate_resources.clear();
        self.changed_resources.clear();
        self.edge_kind_masks.clear();
        self.reverse_indegree.clear();
        self.reverse_cursor.clear();
        self.reverse_offsets.clear();
        self.reverse_targets.clear();
        self.reverse_masks.clear();
        self.frontier_density = FrontierDensityTelemetry::default();
        #[cfg(feature = "gpu-telemetry")]
        {
            self.telemetry_popcount_input_bytes.clear();
            self.telemetry_popcount_output_bytes.clear();
            self.telemetry_sum_bytes.clear();
            self.telemetry_xor_words.clear();
            self.telemetry_popcount_program = None;
            self.telemetry_reduce_program = None;
        }
    }

    /// Prepare retained static input byte slots for invariant dispatch inputs.
    pub(crate) fn prepare_static_input_byte_slots(&mut self, slots: usize) -> Result<(), String> {
        crate::staging_reserve::resize_vec_slots(
            &mut self.static_input_bytes,
            slots,
            "fixed-point static input byte slot",
        )?;
        crate::output_scratch::clear_outputs_preserving_slots(&mut self.static_input_bytes);
        Ok(())
    }

    /// Number of backend output slots currently retained by this scratch.
    #[must_use]
    pub fn output_slot_count(&self) -> usize {
        self.outputs.len()
    }

    /// Current retained capacity for backend output slots.
    #[must_use]
    pub fn output_slot_capacity(&self) -> usize {
        self.outputs.capacity()
    }

    /// Current retained capacity for static dispatch input byte slots.
    #[must_use]
    pub fn static_input_slot_capacity(&self) -> usize {
        self.static_input_bytes.capacity()
    }

    /// Current frontier byte-buffer capacity retained by this scratch.
    #[must_use]
    pub fn frontier_byte_capacity(&self) -> usize {
        self.frontier_bytes.capacity()
    }

    /// Current frontier word-buffer capacity retained by this scratch.
    #[must_use]
    pub fn frontier_word_capacity(&self) -> usize {
        self.frontier_words.capacity()
    }

    /// Current decoded-next-word capacity retained by this scratch.
    #[must_use]
    pub fn next_word_capacity(&self) -> usize {
        self.next.capacity()
    }

    /// Current dispatch resource-list capacity retained by this scratch.
    #[must_use]
    pub fn resource_capacity(&self) -> usize {
        self.resources.capacity()
    }

    /// Current alternate dispatch resource-list capacity retained by this scratch.
    #[must_use]
    pub fn alternate_resource_capacity(&self) -> usize {
        self.alternate_resources.capacity()
    }

    /// Current filtered-edge-kind mask capacity retained by this scratch.
    #[must_use]
    pub fn edge_kind_mask_capacity(&self) -> usize {
        self.edge_kind_masks.capacity()
    }

    /// Current reverse-CSR offset capacity retained by this scratch.
    #[must_use]
    pub fn reverse_offset_capacity(&self) -> usize {
        self.reverse_offsets.capacity()
    }

    /// Current reverse-CSR target capacity retained by this scratch.
    #[must_use]
    pub fn reverse_target_capacity(&self) -> usize {
        self.reverse_targets.capacity()
    }

    /// Latest fixed-point frontier density counters for this scratch session.
    #[must_use]
    pub fn frontier_density(&self) -> FrontierDensityTelemetry {
        self.frontier_density
    }

    pub(crate) fn prepare_frontier_words(
        &mut self,
        words: usize,
        seed_bits: &[u32],
    ) -> Result<(), String> {
        crate::dispatch_decode::try_write_zero_words(
            &mut self.frontier_words,
            words,
            "fixed-point frontier word staging",
        )?;
        for (dst, src) in self
            .frontier_words
            .iter_mut()
            .zip(seed_bits.iter().copied())
        {
            *dst = src;
        }
        Ok(())
    }

    pub(crate) fn begin_frontier_density(&mut self, domain_bits: u32) {
        self.frontier_density = FrontierDensityTelemetry::begin(domain_bits);
        self.record_current_frontier_sample();
    }

    pub(crate) fn record_decoded_frontier_transition(&mut self) {
        let domain_bits = domain_bits_to_usize(
            self.frontier_density.domain_bits,
            &mut self.frontier_density,
        );
        let active = count_set_domain_bits(&self.next, domain_bits, &mut self.frontier_density);
        let delta = count_changed_domain_bits(
            &self.frontier_words,
            &self.next,
            domain_bits,
            &mut self.frontier_density,
        );
        self.frontier_density.record_transition(active, delta);
    }

    pub(crate) fn record_compacted_frontier_window(&mut self, iterations: u32) {
        let domain_bits = domain_bits_to_usize(
            self.frontier_density.domain_bits,
            &mut self.frontier_density,
        );
        let active = count_set_domain_bits(&self.next, domain_bits, &mut self.frontier_density);
        let delta = count_changed_domain_bits(
            &self.frontier_words,
            &self.next,
            domain_bits,
            &mut self.frontier_density,
        );
        self.frontier_density
            .record_window(iterations, active, delta);
    }

    fn record_current_frontier_sample(&mut self) {
        let domain_bits = domain_bits_to_usize(
            self.frontier_density.domain_bits,
            &mut self.frontier_density,
        );
        let active = count_set_domain_bits(
            &self.frontier_words,
            domain_bits,
            &mut self.frontier_density,
        );
        self.frontier_density.record_sample(active);
    }

    pub(crate) fn prepare_next_words(&mut self, words: usize) -> Result<(), String> {
        self.next.clear();
        reserve_total(&mut self.next, words, "fixed-point next frontier word")
    }

    pub(crate) fn prepare_grid_dispatch_config(
        &mut self,
        base: &vyre::DispatchConfig,
        grid_override: [u32; 3],
    ) {
        if !dispatch_config_matches_base_except_grid(&self.dispatch_config, base) {
            self.dispatch_config.profile.clone_from(&base.profile);
            self.dispatch_config.ulp_budget = base.ulp_budget;
            self.dispatch_config.timeout = base.timeout;
            self.dispatch_config.label.clone_from(&base.label);
            self.dispatch_config.max_output_bytes = base.max_output_bytes;
            self.dispatch_config.workgroup_override = base.workgroup_override;
            self.dispatch_config.fixpoint_iterations = base.fixpoint_iterations;
            self.dispatch_config.speculation = base.speculation;
            self.dispatch_config.persistent_thread = base.persistent_thread;
            self.dispatch_config.cooperative = base.cooperative;
        }
        self.dispatch_config.grid_override = Some(grid_override);
    }

    pub(crate) fn prepare_edge_kind_masks(
        &mut self,
        edge_kind_mask: &[u32],
        allowed_mask: u32,
    ) -> Result<(), String> {
        reserve_total(
            &mut self.edge_kind_masks,
            edge_kind_mask.len(),
            "fixed-point filtered edge-kind mask",
        )?;
        if self.edge_kind_masks.len() == edge_kind_mask.len() {
            for (slot, kind) in self.edge_kind_masks.iter_mut().zip(edge_kind_mask) {
                *slot = *kind & allowed_mask;
            }
        } else {
            self.edge_kind_masks.clear();
            self.edge_kind_masks
                .extend(edge_kind_mask.iter().map(|kind| kind & allowed_mask));
        }
        Ok(())
    }

    pub(crate) fn prepare_reverse_csr_scratch(
        &mut self,
        node_count: usize,
        edge_count: usize,
    ) -> Result<(), String> {
        self.reverse_indegree.clear();
        reserve_total(
            &mut self.reverse_indegree,
            node_count,
            "fixed-point reverse CSR indegree",
        )?;
        self.reverse_indegree
            .extend(std::iter::repeat_n(0, node_count));
        self.reverse_cursor.clear();
        reserve_total(
            &mut self.reverse_cursor,
            node_count,
            "fixed-point reverse CSR cursor",
        )?;
        self.reverse_cursor
            .extend(std::iter::repeat_n(0, node_count));
        let offset_words = node_count.checked_add(1).ok_or_else(|| {
            "fixed-point reverse CSR scratch node_count overflows sentinel offset. Fix: shard the graph before reverse CSR preparation."
                .to_string()
        })?;
        crate::dispatch_decode::try_write_zero_words(
            &mut self.reverse_offsets,
            offset_words,
            "fixed-point reverse CSR offset",
        )?;
        self.reverse_targets.clear();
        reserve_total(
            &mut self.reverse_targets,
            edge_count.max(1),
            "fixed-point reverse CSR target",
        )?;
        self.reverse_masks.clear();
        reserve_total(
            &mut self.reverse_masks,
            edge_count.max(1),
            "fixed-point reverse CSR mask",
        )?;
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn replace_outputs_preserving_slots(&mut self, incoming: Vec<Vec<u8>>) {
        crate::output_scratch::replace_outputs_preserving_slots(&mut self.outputs, incoming);
    }
}

#[inline]
fn reserve_total<T>(values: &mut Vec<T>, total: usize, field: &'static str) -> Result<(), String> {
    crate::staging_reserve::reserve_vec(values, total, field)
}

fn dispatch_config_matches_base_except_grid(
    cached: &vyre::DispatchConfig,
    base: &vyre::DispatchConfig,
) -> bool {
    cached.profile == base.profile
        && cached.ulp_budget == base.ulp_budget
        && cached.timeout == base.timeout
        && cached.label == base.label
        && cached.max_output_bytes == base.max_output_bytes
        && cached.workgroup_override == base.workgroup_override
        && cached.fixpoint_iterations == base.fixpoint_iterations
        && cached.speculation == base.speculation
        && cached.persistent_thread == base.persistent_thread
        && cached.cooperative == base.cooperative
}

#[inline]
pub fn count_set_domain_bits(
    words: &[u32],
    domain_bits: usize,
    telemetry: &mut FrontierDensityTelemetry,
) -> u64 {
    let full_words = domain_bits / 32;
    let tail_bits = domain_bits % 32;
    let required_words = full_words + usize::from(tail_bits != 0);
    if words.len() < required_words {
        telemetry.record_truncated_frontier_sample();
    }
    let available_full_words = full_words.min(words.len());
    let mut count = 0_u64;
    for word in words.iter().take(available_full_words) {
        count = bounded_telemetry_add(count, u64::from(word.count_ones()), telemetry);
    }
    if tail_bits != 0 && words.len() > full_words {
        let mask = (1u32 << tail_bits) - 1;
        count = bounded_telemetry_add(
            count,
            u64::from((words[full_words] & mask).count_ones()),
            telemetry,
        );
    }
    count
}

#[inline]
pub fn count_changed_domain_bits(
    current: &[u32],
    next: &[u32],
    domain_bits: usize,
    telemetry: &mut FrontierDensityTelemetry,
) -> u64 {
    let full_words = domain_bits / 32;
    let tail_bits = domain_bits % 32;
    let required_words = full_words + usize::from(tail_bits != 0);
    if current.len() < required_words || next.len() < required_words {
        telemetry.record_truncated_frontier_sample();
    }
    let available_full_words = full_words.min(current.len()).min(next.len());
    let mut count = 0_u64;
    for index in 0..available_full_words {
        count = bounded_telemetry_add(
            count,
            u64::from((current[index] ^ next[index]).count_ones()),
            telemetry,
        );
    }
    if tail_bits != 0 && current.len() > full_words && next.len() > full_words {
        let mask = (1u32 << tail_bits) - 1;
        count = bounded_telemetry_add(
            count,
            u64::from(((current[full_words] ^ next[full_words]) & mask).count_ones()),
            telemetry,
        );
    }
    count
}

/// GPU-accelerated `count_set_domain_bits` using `bitset_popcount` + `reduce_sum`.
///
/// Composes two vyre primitives on the backend so that host work is reduced
/// to a single scalar download. Falls back to the scalar host path for very
/// small frontiers (fewer than 64 words) where dispatch overhead dominates.
///
/// # Errors
///
/// Returns an actionable error string when backend dispatch fails or the
/// returned scalar cannot be decoded.
#[cfg(feature = "gpu-telemetry")]
pub fn count_set_domain_bits_gpu(
    backend: &dyn vyre::VyreBackend,
    config: &vyre::DispatchConfig,
    words: &[u32],
    domain_bits: usize,
    telemetry: &mut FrontierDensityTelemetry,
    scratch: &mut FixedPointScratch,
) -> Result<u64, String> {
    let full_words = domain_bits / 32;
    let tail_bits = domain_bits % 32;
    let required_words = full_words + usize::from(tail_bits != 0);
    if words.len() < required_words {
        telemetry.record_truncated_frontier_sample();
    }

    let available_words = required_words.min(words.len());
    if available_words == 0 {
        return Ok(0);
    }

    // Fallback to host scalar path for very small frontiers where
    // kernel launch overhead dominates.
    if available_words < 64 {
        return Ok(count_set_domain_bits(words, domain_bits, telemetry));
    }

    let words_u32 = u32::try_from(available_words).map_err(|_| {
        "telemetry word count exceeds u32 dispatch metadata. Fix: shard the frontier before GPU telemetry dispatch."
            .to_string()
    })?;

    // Stage input bytes, masking the tail word so the GPU only counts
    // valid domain bits.
    crate::dispatch_decode::try_write_zero_words(
        &mut scratch.telemetry_xor_words,
        available_words,
        "telemetry set bits staging",
    )?;
    for (dst, src) in scratch
        .telemetry_xor_words
        .iter_mut()
        .zip(words.iter().copied())
    {
        *dst = src;
    }
    if tail_bits != 0 && available_words > full_words {
        let mask = (1u32 << tail_bits) - 1;
        scratch.telemetry_xor_words[full_words] &= mask;
    }
    crate::dispatch_decode::try_pack_u32_into(
        &scratch.telemetry_xor_words[..available_words],
        &mut scratch.telemetry_popcount_input_bytes,
        "telemetry set bits input",
    )?;

    let popcount_outputs = dispatch_popcount_then_reduce(backend, config, words_u32, scratch)?;

    let sum_bytes =
        crate::dispatch_decode::only_output(&popcount_outputs, "telemetry reduce_sum", "sum")?;
    let sum_u32 = crate::dispatch_decode::unpack_exact_u32_scalar(sum_bytes, "telemetry sum")?;
    let sum_u64 = u64::from(sum_u32);
    telemetry.record_sample(sum_u64);
    Ok(sum_u64)
}

/// GPU-accelerated `count_changed_domain_bits` using `bitset_popcount` + `reduce_sum`.
///
/// XORs the two frontier buffers on the host (cheap: ~word_count XORs),
/// then dispatches the popcount+reduce composition on the GPU.  The tail
/// word is masked before dispatch so only valid domain bits are counted.
///
/// # Errors
///
/// Returns an actionable error string when backend dispatch fails or the
/// returned scalar cannot be decoded.
#[cfg(feature = "gpu-telemetry")]
pub fn count_changed_domain_bits_gpu(
    backend: &dyn vyre::VyreBackend,
    config: &vyre::DispatchConfig,
    current: &[u32],
    next: &[u32],
    domain_bits: usize,
    telemetry: &mut FrontierDensityTelemetry,
    scratch: &mut FixedPointScratch,
) -> Result<u64, String> {
    let full_words = domain_bits / 32;
    let tail_bits = domain_bits % 32;
    let required_words = full_words + usize::from(tail_bits != 0);
    if current.len() < required_words || next.len() < required_words {
        telemetry.record_truncated_frontier_sample();
    }

    let available_words = required_words.min(current.len()).min(next.len());
    if available_words == 0 {
        return Ok(0);
    }

    // Fallback to host scalar path for very small frontiers.
    if available_words < 64 {
        return Ok(count_changed_domain_bits(
            current,
            next,
            domain_bits,
            telemetry,
        ));
    }

    let words_u32 = u32::try_from(available_words).map_err(|_| {
        "telemetry word count exceeds u32 dispatch metadata. Fix: shard the frontier before GPU telemetry dispatch."
            .to_string()
    })?;

    // XOR on host into reusable scratch, masking the tail word.
    crate::dispatch_decode::try_write_zero_words(
        &mut scratch.telemetry_xor_words,
        available_words,
        "telemetry changed bits staging",
    )?;
    for index in 0..available_words {
        scratch.telemetry_xor_words[index] = current[index] ^ next[index];
    }
    if tail_bits != 0 && available_words > full_words {
        let mask = (1u32 << tail_bits) - 1;
        scratch.telemetry_xor_words[full_words] &= mask;
    }
    crate::dispatch_decode::try_pack_u32_into(
        &scratch.telemetry_xor_words[..available_words],
        &mut scratch.telemetry_popcount_input_bytes,
        "telemetry changed bits input",
    )?;

    let popcount_outputs = dispatch_popcount_then_reduce(backend, config, words_u32, scratch)?;

    let sum_bytes =
        crate::dispatch_decode::only_output(&popcount_outputs, "telemetry reduce_sum", "sum")?;
    let sum_u32 = crate::dispatch_decode::unpack_exact_u32_scalar(sum_bytes, "telemetry sum")?;
    let sum_u64 = u64::from(sum_u32);
    telemetry.record_sample(sum_u64);
    Ok(sum_u64)
}

#[cfg(feature = "gpu-telemetry")]
fn dispatch_popcount_then_reduce(
    backend: &dyn vyre::VyreBackend,
    config: &vyre::DispatchConfig,
    words: u32,
    scratch: &mut FixedPointScratch,
) -> Result<Vec<Vec<u8>>, String> {
    let popcount_byte_len = words
        .checked_mul(4)
        .ok_or_else(|| {
            "telemetry popcount byte length overflows u32. Fix: shard the frontier before GPU telemetry dispatch."
                .to_string()
        })? as usize;

    // Build or reuse cached popcount program.
    let popcount_program = match &scratch.telemetry_popcount_program {
        Some((cached_words, _)) if *cached_words == words => None,
        _ => Some(vyre_primitives::bitset::popcount::bitset_popcount(
            "input",
            "count_words",
            words,
        )),
    };
    if let Some(program) = popcount_program {
        scratch.telemetry_popcount_program = Some((words, program));
    }
    let popcount_program = scratch
        .telemetry_popcount_program
        .as_ref()
        .map(|(_, p)| p)
        .ok_or_else(|| {
            "telemetry popcount program missing after cache update. Fix: report this as an internal logic error."
                .to_string()
        })?;

    // Zero-popcount output staging so the backend sees a clean buffer.
    crate::dispatch_decode::try_write_zero_bytes(
        &mut scratch.telemetry_popcount_output_bytes,
        popcount_byte_len,
        "telemetry popcount output zero staging",
    )?;

    let popcount_inputs: Vec<Vec<u8>> = vec![
        scratch.telemetry_popcount_input_bytes[..popcount_byte_len].to_vec(),
        scratch.telemetry_popcount_output_bytes[..popcount_byte_len].to_vec(),
    ];
    let popcount_outputs = backend
        .dispatch(popcount_program, &popcount_inputs, config)
        .map_err(|error| {
            format!(
                "telemetry bitset_popcount dispatch failed: {error}. Fix: verify backend supports the bitset::popcount op and input buffers are sized correctly."
            )
        })?;

    // Extract per-word popcounts from the popcount output.
    let popcount_result = crate::dispatch_decode::only_output(
        &popcount_outputs,
        "telemetry bitset_popcount",
        "count_words",
    )?;

    // Build or reuse cached reduce program.
    let reduce_program = match &scratch.telemetry_reduce_program {
        Some((cached_count, _)) if *cached_count == words => None,
        _ => Some(vyre_primitives::reduce::sum::reduce_sum(
            "values", "out", words,
        )),
    };
    if let Some(program) = reduce_program {
        scratch.telemetry_reduce_program = Some((words, program));
    }
    let reduce_program = scratch
        .telemetry_reduce_program
        .as_ref()
        .map(|(_, p)| p)
        .ok_or_else(|| {
            "telemetry reduce program missing after cache update. Fix: report this as an internal logic error."
                .to_string()
        })?;

    // Zero reduce output staging.
    crate::dispatch_decode::try_write_zero_bytes(
        &mut scratch.telemetry_sum_bytes,
        std::mem::size_of::<u32>(),
        "telemetry reduce sum output zero staging",
    )?;

    let reduce_inputs: Vec<Vec<u8>> = vec![
        popcount_result.to_vec(),
        scratch.telemetry_sum_bytes[..std::mem::size_of::<u32>()].to_vec(),
    ];
    backend
        .dispatch(reduce_program, &reduce_inputs, config)
        .map_err(|error| {
            format!(
                "telemetry reduce_sum dispatch failed: {error}. Fix: verify backend supports the reduce::sum op and input buffers are sized correctly."
            )
        })
}

#[inline]
fn domain_bits_to_usize(domain_bits: u64, telemetry: &mut FrontierDensityTelemetry) -> usize {
    match usize::try_from(domain_bits) {
        Ok(bits) => bits,
        Err(_) => {
            // Domain overflow is a malformed-input cold path.
            domain_bits_overflow(telemetry)
        }
    }
}

#[cold]
fn domain_bits_overflow(telemetry: &mut FrontierDensityTelemetry) -> usize {
    telemetry.record_domain_overflow();
    usize::MAX
}

#[inline]
fn bounded_telemetry_add(left: u64, right: u64, telemetry: &mut FrontierDensityTelemetry) -> u64 {
    match left.checked_add(right) {
        Some(sum) => sum,
        None => {
            // Arithmetic overflow in telemetry counters is a cold saturation path.
            telemetry_overflow(telemetry)
        }
    }
}

#[cold]
fn telemetry_overflow(telemetry: &mut FrontierDensityTelemetry) -> u64 {
    telemetry.record_arithmetic_overflow();
    u64::MAX
}

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

    #[test]
    fn frontier_telemetry_counts_available_words_instead_of_panicking() {
        let mut telemetry = FrontierDensityTelemetry::begin(96);
        assert_eq!(count_set_domain_bits(&[u32::MAX], 96, &mut telemetry), 32);
        assert_eq!(
            count_changed_domain_bits(&[u32::MAX], &[0], 96, &mut telemetry),
            32
        );
        assert_eq!(telemetry.truncated_frontier_samples, 2);
    }

    #[test]
    fn fixed_point_scratch_buffers_source_has_no_release_path_panic_telemetry_or_resize_growth() {
        let source = include_str!("buffers.rs");
        let production = source
            .split("#[cfg(test)]\nmod tests")
            .next()
            .expect("fixed-point scratch buffer production source must precede tests");
        assert!(
            !production.contains(concat!("panic", "!("))
                && !production.contains(".unwrap_or_else(")
                && !production.contains(".resize(")
                && !production.contains("eprintln!"),
            "Fix: fixed-point scratch buffers must reserve/extend explicitly and frontier telemetry must not abort or write stderr on malformed decode lengths."
        );
        assert!(
            production.contains("bounded_telemetry_add")
                && production.contains("record_truncated_frontier_sample")
                && production.contains("extend(std::iter::repeat_n(0, node_count))"),
            "Fix: fixed-point scratch buffers must keep non-aborting structured telemetry and resize-free reverse CSR staging."
        );
    }

    #[cfg(feature = "gpu-telemetry")]
    #[test]
    fn gpu_telemetry_matches_cpu_reference() {
        let backend = vyre_driver_reference::CpuRefBackend;
        let config = vyre::DispatchConfig::default();

        for &bits in &[96u32, 1_000, 10_000] {
            let words = ((bits + 31) / 32) as usize;
            let data: Vec<u32> = (0..words).map(|i| i as u32).collect();
            let next: Vec<u32> = (0..words).map(|i| !(i as u32)).collect();
            let mut scratch = FixedPointScratch::with_capacities(words, 2, 2)
                .expect("test scratch must allocate");

            let cpu_set = count_set_domain_bits(
                &data,
                bits as usize,
                &mut FrontierDensityTelemetry::default(),
            );
            let gpu_set = count_set_domain_bits_gpu(
                &backend,
                &config,
                &data,
                bits as usize,
                &mut FrontierDensityTelemetry::default(),
                &mut scratch,
            )
            .expect("gpu set bits must succeed");
            assert_eq!(cpu_set, gpu_set, "count_set_domain_bits mismatch at {bits}");

            let cpu_chg = count_changed_domain_bits(
                &data,
                &next,
                bits as usize,
                &mut FrontierDensityTelemetry::default(),
            );
            let gpu_chg = count_changed_domain_bits_gpu(
                &backend,
                &config,
                &data,
                &next,
                bits as usize,
                &mut FrontierDensityTelemetry::default(),
                &mut scratch,
            )
            .expect("gpu changed bits must succeed");
            assert_eq!(
                cpu_chg, gpu_chg,
                "count_changed_domain_bits mismatch at {bits}"
            );
        }
    }

    #[cfg(feature = "gpu-telemetry")]
    #[test]
    fn gpu_telemetry_fallback_for_small_frontiers() {
        let backend = vyre_driver_reference::CpuRefBackend;
        let config = vyre::DispatchConfig::default();
        let data: Vec<u32> = vec![0b1010_1010; 31]; // 31 words = 992 bits (< 64 words threshold)
        let next: Vec<u32> = vec![0b0101_0101; 31];
        let mut scratch =
            FixedPointScratch::with_capacities(31, 2, 2).expect("test scratch must allocate");

        let result = count_set_domain_bits_gpu(
            &backend,
            &config,
            &data,
            992,
            &mut FrontierDensityTelemetry::default(),
            &mut scratch,
        )
        .expect("small frontier gpu telemetry must succeed");
        assert_eq!(result, 4 * 31);

        let result = count_changed_domain_bits_gpu(
            &backend,
            &config,
            &data,
            &next,
            992,
            &mut FrontierDensityTelemetry::default(),
            &mut scratch,
        )
        .expect("small frontier gpu changed telemetry must succeed");
        assert_eq!(result, 8 * 31);
    }
}