weirflow 0.1.0

//! Scale-aware execution planning for Weir fixed-point analyses.
//!
//! This module owns scheduling policy only. It does not dispatch kernels, build
//! graphs, mutate scratch, or own backend resources.

use crate::dense_domain::{
    plan_sparse_dense_domain, SparseDenseDomainMode, SparseDenseDomainObservation,
    SparseDenseDomainPolicy,
};
use crate::fixed_point_scratch::{FrontierDensityTelemetry, FrontierExecutionMode};

/// Shape and memory facts used to choose a fixed-point execution family.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct FixedPointExecutionPlan {
    /// Dispatch-domain nodes.
    pub node_count: u32,
    /// Dispatch-domain CSR edges.
    pub edge_count: u32,
    /// Stable normalized graph layout hash when available.
    pub layout_hash: u64,
    /// Retained invariant graph bytes for the prepared plan.
    pub retained_graph_bytes: usize,
    /// Frontier bytes transferred per host-visible iteration.
    pub frontier_bytes_per_iteration: usize,
    /// Average outgoing degree in parts per million.
    pub average_degree_ppm: u64,
    /// Latest frontier density telemetry used by the decision.
    pub frontier_density: FrontierDensityTelemetry,
    /// Selected execution family for this graph and observed frontier.
    pub execution_mode: FrontierExecutionMode,
}

impl FixedPointExecutionPlan {
    /// Conservative bytes moved by one host-visible iteration.
    pub fn estimated_iteration_io_bytes(self) -> Result<usize, String> {
        self.frontier_bytes_per_iteration
            .checked_mul(2)
            .ok_or_else(|| {
                "weir fixed-point iteration IO byte estimate overflowed usize. Fix: shard the graph before planning host-visible fixed-point execution."
                    .to_string()
            })
    }

    /// Whether repeated runs should favor resident graph resources.
    #[must_use]
    pub fn should_use_resident_graph(self) -> bool {
        self.retained_graph_bytes >= 4096 || self.edge_count >= 1024 || self.node_count >= 1024
    }
}

/// Build a scale-aware plan for an already prepared fixed-point graph.
pub fn plan_prepared_graph(
    graph: &crate::fixed_point_graph::FixedPointForwardGraph,
    frontier_density: FrontierDensityTelemetry,
) -> Result<FixedPointExecutionPlan, String> {
    plan_from_parts(
        graph.node_count(),
        graph.edge_count(),
        graph.stable_layout_hash(),
        graph.retained_bytes(),
        frontier_density,
    )
}

/// Build a scale-aware plan from graph shape and retained-byte facts.
pub fn plan_from_parts(
    node_count: u32,
    edge_count: u32,
    layout_hash: u64,
    retained_graph_bytes: usize,
    frontier_density: FrontierDensityTelemetry,
) -> Result<FixedPointExecutionPlan, String> {
    let frontier_bytes_per_iteration = frontier_bytes_for_node_count(node_count)?;
    let average_degree_ppm = if node_count == 0 {
        0
    } else {
        (u64::from(edge_count) * 1_000_000) / u64::from(node_count)
    };
    let execution_mode = choose_execution_mode(node_count, average_degree_ppm, frontier_density);
    Ok(FixedPointExecutionPlan {
        node_count,
        edge_count,
        layout_hash,
        retained_graph_bytes,
        frontier_bytes_per_iteration,
        average_degree_ppm,
        frontier_density,
        execution_mode,
    })
}

fn choose_execution_mode(
    node_count: u32,
    average_degree_ppm: u64,
    frontier_density: FrontierDensityTelemetry,
) -> FrontierExecutionMode {
    let plan = plan_sparse_dense_domain(
        SparseDenseDomainPolicy::fixed_point_execution(),
        SparseDenseDomainObservation {
            domain_bits: u64::from(node_count),
            samples: frontier_density.samples,
            iterations: frontier_density.iterations,
            active_bits_total: frontier_density.active_bits_total,
            delta_bits_total: frontier_density.delta_bits_total,
            last_active_bits: frontier_density.last_active_bits,
            peak_active_bits: frontier_density.peak_active_bits,
            average_degree_ppm,
        },
    );
    match plan.mode {
        SparseDenseDomainMode::Empty => FrontierExecutionMode::Empty,
        SparseDenseDomainMode::Sparse => FrontierExecutionMode::Sparse,
        SparseDenseDomainMode::Hybrid => FrontierExecutionMode::Hybrid,
        SparseDenseDomainMode::Dense => FrontierExecutionMode::Dense,
    }
}

const fn bitset_words(bits: u32) -> u64 {
    (bits as u64).div_ceil(32)
}

fn frontier_bytes_for_node_count(node_count: u32) -> Result<usize, String> {
    let words = usize::try_from(bitset_words(node_count)).map_err(|source| {
        format!(
            "weir fixed-point plan frontier word count cannot fit usize: {source}. Fix: shard the graph before planning."
        )
    })?;
    words.checked_mul(std::mem::size_of::<u32>()).ok_or_else(|| {
        "weir fixed-point plan frontier byte count overflowed usize. Fix: shard the graph before planning."
            .to_string()
    })
}

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

    fn density(
        active_total: u64,
        delta_total: u64,
        samples: u64,
        iterations: u64,
    ) -> FrontierDensityTelemetry {
        FrontierDensityTelemetry {
            domain_bits: 4096,
            samples,
            iterations,
            active_bits_total: active_total,
            delta_bits_total: delta_total,
            last_active_bits: active_total / samples.max(1),
            last_delta_bits: delta_total / iterations.max(1),
            peak_active_bits: active_total / samples.max(1),
            peak_delta_bits: delta_total / iterations.max(1),
            truncated_frontier_samples: 0,
            domain_overflow_events: 0,
            arithmetic_overflow_events: 0,
        }
    }

    #[test]
    fn execution_plan_prefers_sparse_for_large_low_degree_low_delta_frontiers() {
        let plan = plan_from_parts(4096, 4096, 7, 32768, density(8, 4, 2, 1))
            .expect("large low-degree execution plan must fit host byte accounting");

        assert_eq!(plan.frontier_bytes_per_iteration, 512);
        assert_eq!(plan.average_degree_ppm, 1_000_000);
        assert_eq!(plan.execution_mode, FrontierExecutionMode::Sparse);
        assert!(plan.should_use_resident_graph());
    }

    #[test]
    fn execution_plan_keeps_small_graphs_dense_to_avoid_scheduler_churn() {
        let plan = plan_from_parts(128, 16, 9, 2048, density(2, 1, 2, 1))
            .expect("small graph execution plan must fit host byte accounting");

        assert_eq!(plan.frontier_bytes_per_iteration, 16);
        assert_eq!(plan.execution_mode, FrontierExecutionMode::Dense);
        assert!(!plan.should_use_resident_graph());
    }

    #[test]
    fn execution_plan_uses_hybrid_for_mid_density_mid_degree_graphs() {
        let plan = plan_from_parts(4096, 16384, 11, 65536, density(1024, 256, 2, 1))
            .expect("mid-density execution plan must fit host byte accounting");

        assert_eq!(plan.average_degree_ppm, 4_000_000);
        assert_eq!(plan.execution_mode, FrontierExecutionMode::Hybrid);
    }

    #[test]
    fn execution_plan_uses_exact_bitset_word_ceiling_at_u32_limit() {
        let plan = plan_from_parts(u32::MAX, u32::MAX, 13, 0, density(1, 1, 1, 1))
            .expect("u32-limit execution plan must fit host byte accounting");

        assert_eq!(plan.frontier_bytes_per_iteration, 536_870_912);
        assert_eq!(
            plan.estimated_iteration_io_bytes()
                .expect("u32 graph frontier IO estimate must fit usize"),
            1_073_741_824
        );
        assert_eq!(
            plan.average_degree_ppm,
            (u64::from(u32::MAX) * 1_000_000) / u64::from(u32::MAX)
        );
    }

    #[test]
    fn execution_plan_rejects_manual_iteration_io_overflow_without_panic() {
        let plan = FixedPointExecutionPlan {
            node_count: 1,
            edge_count: 0,
            layout_hash: 0,
            retained_graph_bytes: 0,
            frontier_bytes_per_iteration: usize::MAX,
            average_degree_ppm: 0,
            frontier_density: density(0, 0, 1, 1),
            execution_mode: FrontierExecutionMode::Dense,
        };

        let error = plan
            .estimated_iteration_io_bytes()
            .expect_err("manually constructed overflowing execution plans must return errors");

        assert!(
            error.contains("iteration IO byte estimate overflowed usize")
                && error.contains("Fix: shard the graph"),
            "overflow diagnostic must identify the byte estimate and operator action"
        );
    }

    #[test]
    fn execution_plan_source_has_no_planner_panic_paths() {
        let source = include_str!("fixed_point_execution_plan.rs");

        for forbidden in [
            concat!("panic", "!("),
            concat!(".unwrap_or_else", "(|source|"),
            concat!(".", "unwrap()"),
        ] {
            assert!(
                !source.contains(forbidden),
                "Fix: fixed-point execution planning must return errors instead of panicking or unwrapping on release-path shape arithmetic: {forbidden}"
            );
        }
    }
}