weirflow 0.1.0

//! Backend-resident fixed-point graph state.
//!
//! Prepared graph buffers remove host repacking. Resident graph buffers go one
//! step further: invariant CSR inputs live behind backend resource handles so
//! repeated fixed-point dispatches do not re-upload graph structure.

#![allow(clippy::too_many_arguments)]
mod analysis_methods;
#[cfg(test)]
pub mod tests;

use crate::fixed_point_graph::FixedPointAnalysisKind;
use crate::fixed_point_graph::FixedPointForwardGraph;
use std::collections::HashSet;

pub use crate::fixed_point_resident_batch::{
    FixedPointResidentBatch, FixedPointResidentBatchStats,
};
pub use crate::fixed_point_resident_cache::FixedPointResidentGraphCache;
pub use crate::fixed_point_resident_frontier::FixedPointResidentFrontierScratch;
pub use crate::fixed_point_resident_plan::{
    FixedPointBorrowedResidentPlan, FixedPointResidentPlan,
};

/// Backend-resident invariant graph inputs for fixed-point analyses.
#[derive(Clone, Debug, PartialEq)]
pub struct FixedPointResidentGraph {
    kind: FixedPointAnalysisKind,
    node_count: u32,
    edge_count: u32,
    layout_hash: u64,
    resources: Vec<vyre::backend::Resource>,
}

impl FixedPointResidentGraph {
    /// Upload invariant graph buffers once into backend-resident resources.
    pub fn upload(
        backend: &dyn vyre::VyreBackend,
        graph: &FixedPointForwardGraph,
    ) -> Result<Self, vyre::BackendError> {
        let buffers = [
            graph.pg_nodes_bytes.as_slice(),
            graph.edge_offsets_bytes.as_slice(),
            graph.edge_targets_bytes.as_slice(),
            graph.edge_kind_mask_bytes.as_slice(),
        ];
        let mut resources = crate::staging_reserve::reserved_vec(
            buffers.len() + 1,
            "fixed-point resident graph resource",
        )
        .map_err(vyre::BackendError::new)?;
        for bytes in buffers {
            let resource = match backend.allocate_resident(bytes.len()) {
                Ok(resource) => resource,
                Err(error) => {
                    let _ = free_uploaded_prefix(backend, resources);
                    return Err(error);
                }
            };
            resources.push(resource);
        }
        let uploads = [
            (&resources[0], buffers[0]),
            (&resources[1], buffers[1]),
            (&resources[2], buffers[2]),
            (&resources[3], buffers[3]),
        ];
        if let Err(error) = backend.upload_resident_many(&uploads) {
            let _ = free_uploaded_prefix(backend, resources);
            return Err(error);
        }
        resources.push(resources[0].clone());
        Ok(Self {
            kind: graph.kind(),
            node_count: graph.node_count(),
            edge_count: graph.edge_count(),
            layout_hash: graph.stable_layout_hash(),
            resources,
        })
    }

    /// Number of graph nodes in the resident dispatch domain.
    #[must_use]
    pub fn node_count(&self) -> u32 {
        self.node_count
    }

    /// Analysis family this resident graph was uploaded for.
    #[must_use]
    pub fn kind(&self) -> FixedPointAnalysisKind {
        self.kind
    }

    /// Verify this resident graph layout is consumed by the analysis family
    /// that uploaded it.
    pub fn require_kind(
        &self,
        expected: FixedPointAnalysisKind,
        consumer: &str,
    ) -> Result<(), String> {
        if self.kind == expected {
            return Ok(());
        }
        Err(resident_kind_mismatch(consumer, self.kind, expected))
    }

    /// Number of graph edges in the resident dispatch domain.
    #[must_use]
    pub fn edge_count(&self) -> u32 {
        self.edge_count
    }

    /// Stable hash of the normalized graph layout uploaded into resident buffers.
    #[must_use]
    pub fn stable_layout_hash(&self) -> u64 {
        self.layout_hash
    }

    /// Verify that a prepared graph plan matches this resident graph layout.
    pub fn require_same_layout(
        &self,
        graph: &FixedPointForwardGraph,
        consumer: &str,
    ) -> Result<(), String> {
        if self.kind != graph.kind() {
            return Err(resident_prepared_kind_mismatch(
                consumer,
                self.kind,
                graph.kind(),
            ));
        }
        if self.layout_hash != graph.stable_layout_hash() {
            let mut scratch = crate::error_format::ErrorFormatScratch::default();
            return Err(crate::error_format::write_fixed_point_layout_mismatch(
                &mut scratch,
                consumer,
                self.layout_hash,
                graph.stable_layout_hash(),
            ));
        }
        Ok(())
    }

    /// Number of resident invariant graph resources.
    #[must_use]
    pub fn resident_resource_count(&self) -> usize {
        self.resources.len()
    }

    /// Rebuild borrowed-frontier dispatch resources into caller-owned scratch.
    pub fn resources_with_frontier_into(
        &self,
        frontier_bytes: &[u8],
        resources: &mut Vec<vyre::backend::Resource>,
    ) -> Result<(), String> {
        let invariant_len = stage_frontier_resources(&self.resources, resources)?;
        rewrite_borrowed_resource(&mut resources[invariant_len], frontier_bytes)?;
        rewrite_borrowed_resource(&mut resources[invariant_len + 1], frontier_bytes)?;
        Ok(())
    }

    /// Rebuild resident-frontier dispatch resources into caller-owned scratch.
    pub fn resources_with_resident_frontier_into(
        &self,
        frontier_in: &vyre::backend::Resource,
        frontier_out: &vyre::backend::Resource,
        resources: &mut Vec<vyre::backend::Resource>,
    ) -> Result<(), String> {
        let invariant_len = self.resources.len();
        let total_len = invariant_len.checked_add(2).ok_or_else(|| {
            "fixed-point resident resource list length overflowed usize. Fix: shard resident resources before dispatch."
                .to_string()
        })?;
        crate::staging_reserve::reserve_vec(
            resources,
            total_len,
            "fixed-point resident resource scratch",
        )?;
        if resources.len() < total_len {
            resources.resize_with(total_len, vyre::backend::Resource::default);
        } else {
            resources.truncate(total_len);
        }
        for (slot, resource) in resources[..invariant_len].iter_mut().zip(&self.resources) {
            match resource {
                vyre::backend::Resource::Resident(id) => {
                    if !matches!(slot, vyre::backend::Resource::Resident(existing) if *existing == *id)
                    {
                        *slot = vyre::backend::Resource::Resident(*id);
                    }
                }
                vyre::backend::Resource::Borrowed(_) => {
                    if slot != resource {
                        *slot = resource.clone();
                    }
                }
            }
        }
        resources[invariant_len] = frontier_in.clone();
        resources[invariant_len + 1] = frontier_out.clone();
        Ok(())
    }

    /// Build dispatch resources with an in-place resident frontier and a
    /// resident changed flag.
    pub fn resources_with_resident_frontier_changed_into(
        &self,
        frontier: &vyre::backend::Resource,
        changed: &vyre::backend::Resource,
        resources: &mut Vec<vyre::backend::Resource>,
    ) -> Result<(), String> {
        let invariant_len = stage_frontier_resources(&self.resources, resources)?;
        if resources[invariant_len] != *frontier {
            resources[invariant_len] = frontier.clone();
        }
        if resources[invariant_len + 1] != *changed {
            resources[invariant_len + 1] = changed.clone();
        }
        Ok(())
    }

    /// Free every resident resource owned by this graph.
    pub fn free(self, backend: &dyn vyre::VyreBackend) -> Result<(), vyre::BackendError> {
        let mut first_error = None;
        let mut freed = HashSet::new();
        freed.try_reserve(self.resources.len()).map_err(|source| {
            vyre::BackendError::new(format!(
                "fixed-point resident graph free could not reserve {} released handle slot(s): {source}. Fix: shard resident graph resources before cleanup.",
                self.resources.len()
            ))
        })?;
        for resource in self.resources {
            if let vyre::backend::Resource::Resident(id) = resource {
                if !freed.insert(id) {
                    continue;
                }
                if let Err(error) = backend.free_resident(vyre::backend::Resource::Resident(id)) {
                    first_error.get_or_insert(error);
                }
            } else if let Err(error) = backend.free_resident(resource) {
                first_error.get_or_insert(error);
            }
        }
        if let Some(error) = first_error {
            Err(error)
        } else {
            Ok(())
        }
    }
}

fn stage_frontier_resources(
    invariant_resources: &[vyre::backend::Resource],
    resources: &mut Vec<vyre::backend::Resource>,
) -> Result<usize, String> {
    let invariant_len = invariant_resources.len();
    let total_len = invariant_len.checked_add(2).ok_or_else(|| {
        "fixed-point resident resource list length overflowed usize. Fix: shard resident resources before dispatch."
            .to_string()
    })?;
    crate::staging_reserve::reserve_vec(
        resources,
        total_len,
        "fixed-point resident resource scratch",
    )?;
    if resources.len() < total_len {
        resources.resize_with(total_len, vyre::backend::Resource::default);
    } else {
        resources.truncate(total_len);
    }
    for (slot, resource) in resources[..invariant_len]
        .iter_mut()
        .zip(invariant_resources)
    {
        match resource {
            vyre::backend::Resource::Resident(id) => {
                if !matches!(slot, vyre::backend::Resource::Resident(existing) if *existing == *id)
                {
                    *slot = vyre::backend::Resource::Resident(*id);
                }
            }
            vyre::backend::Resource::Borrowed(_) => {
                if slot != resource {
                    *slot = resource.clone();
                }
            }
        }
    }
    Ok(invariant_len)
}

fn rewrite_borrowed_resource(
    resource: &mut vyre::backend::Resource,
    bytes: &[u8],
) -> Result<(), String> {
    match resource {
        vyre::backend::Resource::Borrowed(buffer) => {
            if buffer.len() == bytes.len() {
                buffer.copy_from_slice(bytes);
            } else {
                buffer.clear();
                crate::staging_reserve::reserve_vec(
                    buffer,
                    bytes.len(),
                    "fixed-point borrowed frontier bytes",
                )?;
                buffer.extend_from_slice(bytes);
            }
        }
        vyre::backend::Resource::Resident(_) => {
            let mut buffer = crate::staging_reserve::reserved_vec(
                bytes.len(),
                "fixed-point borrowed frontier bytes",
            )?;
            buffer.extend_from_slice(bytes);
            *resource = vyre::backend::Resource::Borrowed(buffer);
        }
    }
    Ok(())
}

/// Static labels used by the shared resident fixed-point ping-pong loop.
#[derive(Clone, Copy)]
pub(crate) struct FixedPointResidentLoopLabels {
    pub(crate) analysis: &'static str,
    pub(crate) output: &'static str,
    pub(crate) tail: &'static str,
}

/// Static labels used by the shared resident fixed-point closure driver.
#[derive(Clone, Copy)]
pub(crate) struct FixedPointResidentClosureLabels {
    pub(crate) kind: FixedPointAnalysisKind,
    pub(crate) analysis: &'static str,
    pub(crate) capacity: &'static str,
    pub(crate) output: &'static str,
    pub(crate) tail: &'static str,
}

impl FixedPointResidentClosureLabels {
    fn loop_labels(self) -> FixedPointResidentLoopLabels {
        FixedPointResidentLoopLabels {
            analysis: self.analysis,
            output: self.output,
            tail: self.tail,
        }
    }
}

/// Build a fixed-point resident result buffer with one shared reservation
/// policy across all resident closure families.
pub(crate) fn reserved_resident_closure_result(
    stage: &str,
    graph: &FixedPointResidentGraph,
) -> Result<Vec<u32>, String> {
    crate::staging_reserve::reserved_fixed_point_result(stage, graph.node_count())
}

/// Run a resident graph closure where the mutable frontier is passed as
/// borrowed dispatch bytes on each iteration.
pub(crate) fn run_borrowed_resident_closure_into(
    pipeline: &dyn vyre::CompiledPipeline,
    graph: &FixedPointResidentGraph,
    seed_bits: &[u32],
    max_iterations: u32,
    config: &vyre::DispatchConfig,
    scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
    result: &mut Vec<u32>,
    labels: FixedPointResidentClosureLabels,
    kind_consumer: &str,
) -> Result<(), String> {
    let words = prepare_resident_closure_frontier(
        graph,
        seed_bits,
        max_iterations,
        scratch,
        labels,
        kind_consumer,
    )?;
    scratch.begin_frontier_density(graph.node_count());
    scratch.prepare_grid_dispatch_config(config, [graph.node_count().max(1), 1, 1]);
    scratch.prepare_next_words(words)?;
    for _ in 0..max_iterations {
        crate::dispatch_decode::try_pack_u32_into(&scratch.frontier_words, &mut scratch.frontier_bytes, "weir u32 byte staging")?;
        graph.resources_with_frontier_into(&scratch.frontier_bytes, &mut scratch.resources)?;
        pipeline
            .dispatch_persistent_handles_into(
                &scratch.resources,
                &scratch.dispatch_config,
                &mut scratch.outputs,
            )
            .map_err(|error| error.to_string())?;
        crate::dispatch_decode::unpack_only_exact_u32_into(
            &scratch.outputs,
            labels.analysis,
            labels.output,
            words,
            &mut scratch.next,
        )?;
        crate::dispatch_decode::require_bitset_tail_clear(
            labels.tail,
            &scratch.next,
            graph.node_count(),
        )?;
        scratch.record_decoded_frontier_transition();
        if crate::fixed_point_scratch::copy_if_converged(
            &scratch.next,
            &scratch.frontier_words,
            result,
        )? {
            return Ok(());
        }
        std::mem::swap(&mut scratch.frontier_words, &mut scratch.next);
    }
    Err(format!(
        "{} did not converge within {max_iterations} iterations",
        labels.analysis
    ))
}

/// Run a resident graph closure using temporary backend-resident frontier
/// resources for this invocation.
pub(crate) fn run_ephemeral_resident_closure_into(
    backend: &dyn vyre::VyreBackend,
    pipeline: &dyn vyre::CompiledPipeline,
    graph: &FixedPointResidentGraph,
    seed_bits: &[u32],
    max_iterations: u32,
    config: &vyre::DispatchConfig,
    scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
    result: &mut Vec<u32>,
    labels: FixedPointResidentClosureLabels,
    kind_consumer: &str,
) -> Result<(), String> {
    let words = prepare_resident_closure_frontier(
        graph,
        seed_bits,
        max_iterations,
        scratch,
        labels,
        kind_consumer,
    )?;
    let frontier_byte_len = resident_frontier_byte_len(labels, words)?;
    with_ephemeral_resident_frontier_pair(
        backend,
        frontier_byte_len,
        |frontier_in, frontier_out| {
            run_resident_frontier_pair_loop(
                backend,
                pipeline,
                graph,
                frontier_in,
                frontier_out,
                max_iterations,
                config,
                scratch,
                result,
                labels.loop_labels(),
            )
        },
    )
}

/// Run a resident graph closure using caller-retained backend frontier
/// resources for repeated invocations.
pub(crate) fn run_reusable_resident_closure_into(
    backend: &dyn vyre::VyreBackend,
    pipeline: &dyn vyre::CompiledPipeline,
    graph: &FixedPointResidentGraph,
    seed_bits: &[u32],
    max_iterations: u32,
    config: &vyre::DispatchConfig,
    scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
    resident_frontier: &mut crate::fixed_point_resident::FixedPointResidentFrontierScratch,
    result: &mut Vec<u32>,
    labels: FixedPointResidentClosureLabels,
    kind_consumer: &str,
) -> Result<(), String> {
    let words = prepare_resident_closure_frontier(
        graph,
        seed_bits,
        max_iterations,
        scratch,
        labels,
        kind_consumer,
    )?;
    let frontier_byte_len = resident_frontier_byte_len(labels, words)?;
    let (frontier_in, frontier_out) =
        resident_frontier.ensure_pair_refs(backend, frontier_byte_len)?;
    run_resident_frontier_pair_loop(
        backend,
        pipeline,
        graph,
        frontier_in,
        frontier_out,
        max_iterations,
        config,
        scratch,
        result,
        labels.loop_labels(),
    )
}

fn prepare_resident_closure_frontier(
    graph: &FixedPointResidentGraph,
    seed_bits: &[u32],
    max_iterations: u32,
    scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
    labels: FixedPointResidentClosureLabels,
    kind_consumer: &str,
) -> Result<usize, String> {
    graph.require_kind(labels.kind, kind_consumer)?;
    crate::dispatch_decode::require_positive_iterations(labels.analysis, max_iterations)?;
    let words = crate::dispatch_decode::bitset_word_capacity(labels.capacity, graph.node_count())?;
    crate::dispatch_decode::require_bitset_words(labels.analysis, seed_bits, words)?;
    crate::dispatch_decode::require_bitset_tail_clear(
        labels.analysis,
        seed_bits,
        graph.node_count(),
    )?;
    scratch.prepare_frontier_words(words, seed_bits)?;
    Ok(words)
}

fn resident_frontier_byte_len(
    labels: FixedPointResidentClosureLabels,
    words: usize,
) -> Result<usize, String> {
    words.checked_mul(4).ok_or_else(|| {
        format!(
            "{} resident frontier byte length overflow. Fix: shard the graph before GPU dispatch.",
            labels.analysis
        )
    })
}

/// Allocate a temporary resident frontier pair, run a closure, then free both
/// resources while preserving the original run error over cleanup errors.
pub(crate) fn with_ephemeral_resident_frontier_pair<F>(
    backend: &dyn vyre::VyreBackend,
    byte_len: usize,
    run: F,
) -> Result<(), String>
where
    F: FnOnce(&vyre::backend::Resource, &vyre::backend::Resource) -> Result<(), String>,
{
    let frontier_in = backend
        .allocate_resident(byte_len)
        .map_err(|error| error.to_string())?;
    let frontier_out = match backend.allocate_resident(byte_len) {
        Ok(resource) => resource,
        Err(error) => {
            let _ = backend.free_resident(frontier_in);
            return Err(error.to_string());
        }
    };
    let run_result = run(&frontier_in, &frontier_out);
    let free_in = backend
        .free_resident(frontier_in)
        .map_err(|error| error.to_string());
    let free_out = backend
        .free_resident(frontier_out)
        .map_err(|error| error.to_string());
    match (run_result, free_in, free_out) {
        (Err(error), _, _) => Err(error),
        (Ok(_), Err(error), _) | (Ok(_), _, Err(error)) => Err(error),
        (Ok(()), Ok(()), Ok(())) => Ok(()),
    }
}

/// Run the common resident fixed-point ping-pong frontier loop.
pub(crate) fn run_resident_frontier_pair_loop(
    backend: &dyn vyre::VyreBackend,
    pipeline: &dyn vyre::CompiledPipeline,
    graph: &FixedPointResidentGraph,
    frontier_in: &vyre::backend::Resource,
    frontier_out: &vyre::backend::Resource,
    max_iterations: u32,
    config: &vyre::DispatchConfig,
    scratch: &mut crate::fixed_point_scratch::FixedPointScratch,
    result: &mut Vec<u32>,
    labels: FixedPointResidentLoopLabels,
) -> Result<(), String> {
    let words = scratch.frontier_words.len();
    scratch.begin_frontier_density(graph.node_count());
    scratch.prepare_grid_dispatch_config(config, [graph.node_count().max(1), 1, 1]);
    scratch.prepare_next_words(words)?;
    crate::dispatch_decode::try_pack_u32_into(&scratch.frontier_words, &mut scratch.frontier_bytes, "weir u32 byte staging")?;
    backend
        .upload_resident_many(&[
            (frontier_in, scratch.frontier_bytes.as_slice()),
            (frontier_out, scratch.frontier_bytes.as_slice()),
        ])
        .map_err(|error| error.to_string())?;
    let mut current_frontier = frontier_in;
    let mut next_frontier = frontier_out;
    for _ in 0..max_iterations {
        graph.resources_with_resident_frontier_into(
            current_frontier,
            next_frontier,
            &mut scratch.resources,
        )?;
        pipeline
            .dispatch_persistent_handles_into(
                &scratch.resources,
                &scratch.dispatch_config,
                &mut scratch.outputs,
            )
            .map_err(|error| error.to_string())?;
        crate::dispatch_decode::unpack_only_exact_u32_into(
            &scratch.outputs,
            labels.analysis,
            labels.output,
            words,
            &mut scratch.next,
        )?;
        crate::dispatch_decode::require_bitset_tail_clear(
            labels.tail,
            &scratch.next,
            graph.node_count(),
        )?;
        scratch.record_decoded_frontier_transition();
        let supports_gpu_equal = backend
            .supported_ops()
            .contains(vyre_primitives::bitset::equal::OP_ID);
        let converged = if supports_gpu_equal {
            crate::fixed_point_scratch::converged_via_gpu_backend(
                &scratch.outputs[0],
                &scratch.frontier_bytes,
                words,
                backend,
            )
            .unwrap_or(false)
        } else {
            crate::fixed_point_scratch::copy_if_converged(
                &scratch.next,
                &scratch.frontier_words,
                result,
            )?
        };
        if converged {
            if !supports_gpu_equal {
                // copy_if_converged already copied into result
            } else {
                crate::fixed_point_scratch::copy_words_into(&scratch.next, result)?;
            }
            return Ok(());
        }
        std::mem::swap(&mut scratch.frontier_words, &mut scratch.next);
        crate::dispatch_decode::try_pack_u32_into(&scratch.frontier_words, &mut scratch.frontier_bytes, "weir u32 byte staging")?;
        std::mem::swap(&mut current_frontier, &mut next_frontier);
    }
    Err(format!(
        "{} did not converge within {max_iterations} iterations",
        labels.analysis
    ))
}

fn free_uploaded_prefix(
    backend: &dyn vyre::VyreBackend,
    resources: Vec<vyre::backend::Resource>,
) -> Result<(), vyre::BackendError> {
    let mut freed = HashSet::new();
    freed.try_reserve(resources.len()).map_err(|source| {
        vyre::BackendError::new(format!(
            "fixed-point resident upload rollback could not reserve {} released handle slot(s): {source}. Fix: shard resident graph upload before cleanup.",
            resources.len()
        ))
    })?;
    for resource in resources {
        if let vyre::backend::Resource::Resident(id) = resource {
            if !freed.insert(id) {
                continue;
            }
            let _ = backend.free_resident(vyre::backend::Resource::Resident(id));
        } else {
            let _ = backend.free_resident(resource);
        }
    }
    Ok(())
}

#[cold]
fn resident_kind_mismatch(
    consumer: &str,
    got: FixedPointAnalysisKind,
    expected: FixedPointAnalysisKind,
) -> String {
    let mut scratch = crate::error_format::ErrorFormatScratch::default();
    let _ = std::fmt::Write::write_fmt(
        &mut scratch.buf,
        format_args!(
            "{consumer} received a {got:?} resident fixed-point graph, expected {expected:?}. Fix: upload the graph with the matching Weir analysis constructor before resident dispatch."
        ),
    );
    scratch.finish()
}

#[cold]
fn resident_prepared_kind_mismatch(
    consumer: &str,
    resident: FixedPointAnalysisKind,
    prepared: FixedPointAnalysisKind,
) -> String {
    let mut scratch = crate::error_format::ErrorFormatScratch::default();
    let _ = std::fmt::Write::write_fmt(
        &mut scratch.buf,
        format_args!(
            "{consumer} received a {resident:?} resident fixed-point graph but a {prepared:?} prepared graph. Fix: build the resident graph from the same prepared analysis plan before sequence-window dispatch."
        ),
    );
    scratch.finish()
}

#[cfg(test)]
mod resident_closure_driver_tests {
    use super::{
        prepare_resident_closure_frontier, resident_frontier_byte_len,
        FixedPointResidentClosureLabels, FixedPointResidentGraph,
    };
    use crate::fixed_point_graph::FixedPointAnalysisKind;

    const LABELS: FixedPointResidentClosureLabels = FixedPointResidentClosureLabels {
        kind: FixedPointAnalysisKind::Reaching,
        analysis: "test_closure",
        capacity: "test_closure resident",
        output: "frontier",
        tail: "test_closure output",
    };

    fn graph(kind: FixedPointAnalysisKind, node_count: u32) -> FixedPointResidentGraph {
        FixedPointResidentGraph {
            kind,
            node_count,
            edge_count: 0,
            layout_hash: 0,
            resources: Vec::new(),
        }
    }

    #[test]
    fn resident_closure_driver_prepares_seed_once_for_all_wrappers() {
        let graph = graph(FixedPointAnalysisKind::Reaching, 33);
        let mut scratch = crate::fixed_point_scratch::FixedPointScratch::default();

        let words = prepare_resident_closure_frontier(
            &graph,
            &[0x8000_0001, 1],
            7,
            &mut scratch,
            LABELS,
            "test_closure_kind",
        )
        .expect("Fix: shared resident driver must accept a canonical two-word frontier");

        assert_eq!(words, 2);
        assert_eq!(scratch.frontier_words, [0x8000_0001, 1]);
    }

    #[test]
    fn resident_closure_driver_rejects_wrong_graph_family_before_dispatch() {
        let graph = graph(FixedPointAnalysisKind::Live, 1);
        let mut scratch = crate::fixed_point_scratch::FixedPointScratch::default();

        let error = prepare_resident_closure_frontier(
            &graph,
            &[1],
            1,
            &mut scratch,
            LABELS,
            "test_closure_kind",
        )
        .expect_err("Fix: shared resident driver must reject mismatched resident graph families");

        assert!(error.contains("test_closure_kind"));
        assert!(scratch.frontier_words.is_empty());
    }

    #[test]
    fn resident_closure_driver_rejects_dirty_tail_bits() {
        let graph = graph(FixedPointAnalysisKind::Reaching, 33);
        let mut scratch = crate::fixed_point_scratch::FixedPointScratch::default();

        let error = prepare_resident_closure_frontier(
            &graph,
            &[0, 0b10],
            1,
            &mut scratch,
            LABELS,
            "test_closure_kind",
        )
        .expect_err("Fix: shared resident driver must reject frontier bits outside node_count");

        assert!(error.contains("test_closure"));
        assert!(scratch.frontier_words.is_empty());
    }

    #[test]
    fn resident_closure_driver_uses_checked_frontier_byte_lengths() {
        assert_eq!(resident_frontier_byte_len(LABELS, 0).unwrap(), 0);
        assert_eq!(resident_frontier_byte_len(LABELS, 3).unwrap(), 12);

        let error = resident_frontier_byte_len(LABELS, usize::MAX)
            .expect_err("Fix: resident frontier byte length arithmetic must be checked");
        assert!(error.contains("test_closure"));
        assert!(error.contains("overflow"));
    }
}