vyre-driver-wgpu 0.6.3

//! Batched megakernel dispatch built on a persistent device work queue.

use super::batch::{
    persistent_storage_binding_usage, queue_state_word, FileBatch, HitRecord, FILE_METADATA_WORDS,
    HIT_RECORD_WORDS, QUEUE_STATE_WORDS,
};
use super::dispatch_plan::{BatchDispatchPlan, BatchDispatchPlanCache, BatchDispatchPlanLookup};
use super::segmentation::SEGMENT_WORDS;
use super::pipeline_cache::{BatchPipelineCache, BatchPipelineShape};
use crate::buffer::GpuBufferHandle;
use crate::{pipeline::WgpuPipeline, WgpuBackend};
use std::sync::Arc;
use std::time::{Duration, Instant};
use vyre_driver::{CompiledPipeline, DispatchConfig, VyreBackend};
use vyre_foundation::ir::{BufferAccess, BufferDecl, DataType, Expr, Node, Program};
use vyre_runtime::megakernel::advanced::hierarchical_atomics::record_hit_to_ring_hierarchical;
use vyre_runtime::megakernel::ir_util::atomic_load_relaxed;
use vyre_runtime::megakernel::rule_catalog::{
    accepted_rule_fingerprints_and_rejections_into, pack_rule_catalog_into, BatchRuleProgram,
    BatchRuleRejection, RuleCatalogPackingScratch, RULE_META_WORDS,
};
use vyre_runtime::megakernel::scaling::{
    MegakernelLaunchPolicy, MegakernelLaunchRecommendation, MegakernelLaunchRequest,
};
use vyre_runtime::megakernel::MegakernelDispatchTopology;
use vyre_runtime::PipelineError;

/// Schema version for WGPU scan batch segmentation evidence.
pub const WGPU_SCAN_BATCH_SEGMENTATION_SCHEMA_VERSION: u32 = 1;

/// Input counters for WGPU scan batch segmentation evidence.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct WgpuScanBatchSegmentationRequest {
    /// Logical scan chunks in the batch.
    pub chunk_count: u32,
    /// Maximum chunks recorded into one command encoder.
    pub max_chunks_per_command_encoder: u32,
    /// Bind groups reused across command encoders.
    pub bind_group_reuse_count: u32,
    /// Bind groups created for command encoders.
    pub bind_group_create_count: u32,
    /// Host-to-device copy commands recorded for the batch.
    pub upload_copy_count: u32,
    /// Device-to-host or device-to-staging copy commands recorded for the batch.
    pub readback_copy_count: u32,
    /// CPU oracle or backend-independent match digest.
    pub expected_match_digest: u64,
    /// WGPU segmented batch match digest.
    pub actual_match_digest: u64,
}

impl WgpuScanBatchSegmentationRequest {
    /// Construct WGPU scan batch segmentation counters.
    #[must_use]
    pub const fn new(
        chunk_count: u32,
        max_chunks_per_command_encoder: u32,
        bind_group_reuse_count: u32,
        bind_group_create_count: u32,
        upload_copy_count: u32,
        readback_copy_count: u32,
        expected_match_digest: u64,
        actual_match_digest: u64,
    ) -> Self {
        Self {
            chunk_count,
            max_chunks_per_command_encoder,
            bind_group_reuse_count,
            bind_group_create_count,
            upload_copy_count,
            readback_copy_count,
            expected_match_digest,
            actual_match_digest,
        }
    }
}

/// Evidence emitted for one WGPU segmented scan batch.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WgpuScanBatchSegmentationEvidence {
    /// Evidence schema version.
    pub schema_version: u32,
    /// Logical scan chunks in the batch.
    pub chunk_count: u32,
    /// Segment count after applying the command encoder chunk limit.
    pub segment_count: u32,
    /// Command encoders required by the segmentation plan.
    pub command_encoder_count: u32,
    /// Bind groups reused across command encoders.
    pub bind_group_reuse_count: u32,
    /// Bind groups created for command encoders.
    pub bind_group_create_count: u32,
    /// Bind group reuse ratio in basis points.
    pub bind_group_reuse_bps: u16,
    /// Host-to-device copy commands recorded for the batch.
    pub upload_copy_count: u32,
    /// Device-to-host or device-to-staging copy commands recorded for the batch.
    pub readback_copy_count: u32,
    /// Total copy commands recorded for the batch.
    pub copy_count: u32,
    /// Stable match digest when WGPU output matches the oracle.
    pub match_digest: u64,
    /// True when expected and actual match digests are identical.
    pub match_parity: bool,
    /// True when command encoder, bind group, and copy counts are present.
    pub all_command_counts_recorded: bool,
}

impl WgpuScanBatchSegmentationEvidence {
    /// Return true when evidence has the schema, command counts, and match
    /// parity required by release benchmark claims.
    #[must_use]
    pub const fn is_complete(self) -> bool {
        self.schema_version == WGPU_SCAN_BATCH_SEGMENTATION_SCHEMA_VERSION
            && self.chunk_count != 0
            && self.segment_count != 0
            && self.command_encoder_count == self.segment_count
            && self.copy_count == self.upload_copy_count + self.readback_copy_count
            && self.match_digest != 0
            && self.match_parity
            && self.all_command_counts_recorded
    }
}

/// WGPU scan batch segmentation evidence error.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum WgpuScanBatchSegmentationError {
    /// The batch contains no chunks.
    EmptyBatch,
    /// The command encoder segmentation limit is zero.
    ZeroChunksPerCommandEncoder,
    /// Bind group counts do not account for every command encoder.
    BindGroupCountMismatch {
        /// Command encoders produced by segmentation.
        command_encoder_count: u32,
        /// Bind groups reused across command encoders.
        bind_group_reuse_count: u32,
        /// Bind groups created for command encoders.
        bind_group_create_count: u32,
    },
    /// Copy count overflowed the evidence ABI.
    CopyCountOverflow,
    /// Match digest is absent.
    ZeroMatchDigest,
    /// WGPU output digest diverged from the oracle.
    MatchDigestMismatch {
        /// CPU oracle or backend-independent match digest.
        expected_match_digest: u64,
        /// WGPU segmented batch match digest.
        actual_match_digest: u64,
    },
}

impl std::fmt::Display for WgpuScanBatchSegmentationError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::EmptyBatch => formatter.write_str(
                "WGPU scan batch has zero chunks. Fix: publish at least one scan chunk before recording segmentation evidence.",
            ),
            Self::ZeroChunksPerCommandEncoder => formatter.write_str(
                "WGPU scan batch has zero chunks per command encoder. Fix: configure a positive segmentation limit.",
            ),
            Self::BindGroupCountMismatch {
                command_encoder_count,
                bind_group_reuse_count,
                bind_group_create_count,
            } => write!(
                formatter,
                "WGPU scan batch bind group counts reuse={bind_group_reuse_count} create={bind_group_create_count} do not account for {command_encoder_count} command encoder(s). Fix: record one reused or created bind group per segment."
            ),
            Self::CopyCountOverflow => formatter.write_str(
                "WGPU scan batch copy count overflowed u32. Fix: shard the scan batch before recording evidence.",
            ),
            Self::ZeroMatchDigest => formatter.write_str(
                "WGPU scan batch match digest is zero. Fix: compute the match digest before accepting segmentation evidence.",
            ),
            Self::MatchDigestMismatch {
                expected_match_digest,
                actual_match_digest,
            } => write!(
                formatter,
                "WGPU scan batch match digest mismatch expected={expected_match_digest:#x} actual={actual_match_digest:#x}. Fix: reject the segmented batch or repair command/copy segmentation before reporting portable scan parity."
            ),
        }
    }
}

impl std::error::Error for WgpuScanBatchSegmentationError {}

/// Build WGPU scan batch segmentation evidence from recorded command counters.
///
/// # Errors
///
/// Returns [`WgpuScanBatchSegmentationError`] when the batch is empty, the
/// command counts are incomplete, copy counts overflow, or match parity fails.
pub fn wgpu_scan_batch_segmentation_evidence(
    request: WgpuScanBatchSegmentationRequest,
) -> Result<WgpuScanBatchSegmentationEvidence, WgpuScanBatchSegmentationError> {
    if request.chunk_count == 0 {
        return Err(WgpuScanBatchSegmentationError::EmptyBatch);
    }
    if request.max_chunks_per_command_encoder == 0 {
        return Err(WgpuScanBatchSegmentationError::ZeroChunksPerCommandEncoder);
    }
    if request.expected_match_digest == 0 || request.actual_match_digest == 0 {
        return Err(WgpuScanBatchSegmentationError::ZeroMatchDigest);
    }
    if request.expected_match_digest != request.actual_match_digest {
        return Err(WgpuScanBatchSegmentationError::MatchDigestMismatch {
            expected_match_digest: request.expected_match_digest,
            actual_match_digest: request.actual_match_digest,
        });
    }

    let command_encoder_count = div_ceil_u32(
        request.chunk_count,
        request.max_chunks_per_command_encoder,
    );
    let bind_group_count = request
        .bind_group_reuse_count
        .checked_add(request.bind_group_create_count)
        .ok_or(WgpuScanBatchSegmentationError::BindGroupCountMismatch {
            command_encoder_count,
            bind_group_reuse_count: request.bind_group_reuse_count,
            bind_group_create_count: request.bind_group_create_count,
        })?;
    if bind_group_count != command_encoder_count {
        return Err(WgpuScanBatchSegmentationError::BindGroupCountMismatch {
            command_encoder_count,
            bind_group_reuse_count: request.bind_group_reuse_count,
            bind_group_create_count: request.bind_group_create_count,
        });
    }

    let copy_count = request
        .upload_copy_count
        .checked_add(request.readback_copy_count)
        .ok_or(WgpuScanBatchSegmentationError::CopyCountOverflow)?;
    let bind_group_reuse_bps =
        ((u64::from(request.bind_group_reuse_count) * 10_000) / u64::from(command_encoder_count))
            as u16;

    Ok(WgpuScanBatchSegmentationEvidence {
        schema_version: WGPU_SCAN_BATCH_SEGMENTATION_SCHEMA_VERSION,
        chunk_count: request.chunk_count,
        segment_count: command_encoder_count,
        command_encoder_count,
        bind_group_reuse_count: request.bind_group_reuse_count,
        bind_group_create_count: request.bind_group_create_count,
        bind_group_reuse_bps,
        upload_copy_count: request.upload_copy_count,
        readback_copy_count: request.readback_copy_count,
        copy_count,
        match_digest: request.expected_match_digest,
        match_parity: true,
        all_command_counts_recorded: true,
    })
}

const fn div_ceil_u32(numerator: u32, denominator: u32) -> u32 {
    ((numerator as u64 + denominator as u64 - 1) / denominator as u64) as u32
}

/// Sparse hit-ring writer selected for the batched megakernel.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum BatchHitWriter {
    /// Select hierarchical subgroup atomics when the backend advertises them,
    /// otherwise use the scalar writer.
    Auto,
    /// One global atomic per hit. Universally supported but slower under high
    /// hit density.
    Scalar,
    /// One global atomic per subgroup. Requires subgroup operations and fails
    /// loudly if the backend cannot compile subgroup intrinsics.
    HierarchicalSubgroup,
}

// NOTE: the `scan_batch_segmentation_tests` test module was relocated to the END
// of this file. An inline test module here previously split the production source
// that the source-shape tests inspect (they take everything before the first test
// module), truncating it before the launch/dispatch lines they assert on. Keeping
// all test modules at the end keeps that production-source view intact. (This note
// deliberately avoids the literal test-config attribute so it does not re-trigger
// that truncation.)

impl BatchHitWriter {
    /// Resolve this selection against backend subgroup capability.
    ///
    /// # Errors
    ///
    /// Returns [`PipelineError::Backend`] when subgroup atomics are explicitly
    /// requested on a backend that does not report subgroup support.
    pub fn resolve_for_backend(self, subgroup_supported: bool) -> Result<Self, PipelineError> {
        match (self, subgroup_supported) {
            (Self::Auto, true) => Ok(Self::HierarchicalSubgroup),
            (Self::Auto, false) => Ok(Self::Scalar),
            (Self::HierarchicalSubgroup, false) => Err(PipelineError::Backend(
                "BatchHitWriter::HierarchicalSubgroup requires backend subgroup ops, but this backend reports supports_subgroup_ops=false. Fix: use BatchHitWriter::Auto/Scalar or run on a subgroup-capable adapter."
                    .to_string(),
            )),
            (mode, _) => Ok(mode),
        }
    }
}

/// Immutable pipeline + launch geometry for batched megakernel scans.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BatchDispatchConfig {
    /// Worker lanes per workgroup.
    pub workgroup_size_x: u32,
    /// Number of workgroups to launch for each batch.
    pub worker_groups: u32,
    /// Maximum sparse hits retained in the output ring.
    pub hit_capacity: u32,
    /// Per-dispatch timeout budget.
    pub timeout: Duration,
    /// Optional graph-node count hint for topology selection.
    pub graph_node_count: u32,
    /// Optional graph-edge count hint for topology selection.
    pub graph_edge_count: u32,
    /// Optional active-frontier density in basis points.
    pub frontier_density_bps: u16,
    /// Optional memory-pressure estimate in basis points.
    pub memory_pressure_bps: u16,
    /// Additional device-resident bytes already committed for this dispatch family.
    ///
    /// The dispatcher adds its fixed queue-state resident footprint when building
    /// the shared launch-policy request.
    pub resident_device_bytes: u64,
    /// Hard device-memory budget for policy planning. Zero means unbounded.
    pub device_memory_budget_bytes: u64,
    /// Hot opcode count observed by the caller or runtime telemetry.
    pub hot_opcode_count: u32,
    /// Hot window count observed by the caller or runtime telemetry.
    pub hot_window_count: u32,
    /// Requeued continuation count observed by the caller or runtime telemetry.
    pub requeue_count: u64,
    /// Maximum priority age observed by the caller or runtime telemetry.
    pub max_priority_age: u32,
}

impl Default for BatchDispatchConfig {
    fn default() -> Self {
        Self {
            workgroup_size_x: 64,
            // `0` is a sentinel meaning "compute from adapter occupancy at
            // dispatcher construction time".  Explicit non-zero values are
            // preserved so callers who set `worker_groups` by hand are not
            // overridden.
            worker_groups: 0,
            hit_capacity: 65_536,
            timeout: Duration::from_secs(30),
            graph_node_count: 0,
            graph_edge_count: 0,
            frontier_density_bps: 0,
            memory_pressure_bps: 0,
            resident_device_bytes: 0,
            device_memory_budget_bytes: 0,
            hot_opcode_count: 0,
            hot_window_count: 0,
            requeue_count: 0,
            max_priority_age: 0,
        }
    }
}

impl BatchDispatchConfig {
    /// Attach graph-topology hints used by the shared megakernel policy.
    #[must_use]
    pub const fn with_graph_hints(
        mut self,
        graph_node_count: u32,
        graph_edge_count: u32,
        frontier_density_bps: u16,
        memory_pressure_bps: u16,
    ) -> Self {
        self.graph_node_count = graph_node_count;
        self.graph_edge_count = graph_edge_count;
        self.frontier_density_bps = if frontier_density_bps > 10_000 {
            10_000
        } else {
            frontier_density_bps
        };
        self.memory_pressure_bps = if memory_pressure_bps > 10_000 {
            10_000
        } else {
            memory_pressure_bps
        };
        self
    }

    /// Attach hard device-memory budget hints used by the shared launch policy.
    #[must_use]
    pub const fn with_device_memory_budget(
        mut self,
        resident_device_bytes: u64,
        device_memory_budget_bytes: u64,
    ) -> Self {
        self.resident_device_bytes = resident_device_bytes;
        self.device_memory_budget_bytes = device_memory_budget_bytes;
        self
    }

    /// Attach execution hotness hints used by interpreter/JIT routing.
    #[must_use]
    pub const fn with_execution_hints(
        mut self,
        hot_opcode_count: u32,
        hot_window_count: u32,
        requeue_count: u64,
        max_priority_age: u32,
    ) -> Self {
        self.hot_opcode_count = hot_opcode_count;
        self.hot_window_count = hot_window_count;
        self.requeue_count = requeue_count;
        self.max_priority_age = max_priority_age;
        self
    }

    /// Return the shared launch-policy recommendation for this batch shape.
    ///
    /// # Errors
    ///
    /// Returns [`PipelineError::Backend`] when adapter limits are malformed.
    pub fn launch_recommendation(
        &self,
        limits: &wgpu::Limits,
        queue_len: u32,
    ) -> Result<MegakernelLaunchRecommendation, PipelineError> {
        let resident_device_bytes = self
            .resident_device_bytes
            .checked_add(batch_fixed_resident_overhead_bytes())
            .ok_or_else(|| {
                PipelineError::Backend(
                    "megakernel resident byte estimate overflowed u64. Fix: shard resident state before launch recommendation."
                        .to_string(),
                )
            })?;
        MegakernelLaunchPolicy::standard()
            .recommend(MegakernelLaunchRequest {
                queue_len,
                requested_worker_groups: self.worker_groups,
                max_workgroup_size_x: self.workgroup_size_x,
                max_compute_workgroups_per_dimension: limits.max_compute_workgroups_per_dimension,
                max_compute_invocations_per_workgroup: limits.max_compute_invocations_per_workgroup,
                requested_hit_capacity: self.hit_capacity,
                expected_hits_per_item: 1,
                hot_opcode_count: self.hot_opcode_count,
                hot_window_count: self.hot_window_count,
                requeue_count: self.requeue_count,
                max_priority_age: self.max_priority_age,
                graph_node_count: if self.graph_node_count == 0 {
                    queue_len
                } else {
                    self.graph_node_count
                },
                graph_edge_count: self.graph_edge_count,
                frontier_density_bps: self.frontier_density_bps,
                memory_pressure_bps: self.memory_pressure_bps,
                resident_device_bytes,
                device_memory_budget_bytes: self.device_memory_budget_bytes,
            })
            .map_err(|source| PipelineError::Backend(source.to_string()))
    }
}

fn batch_fixed_resident_overhead_bytes() -> u64 {
    dispatcher_usize_to_u64(QUEUE_STATE_WORDS, "queue-state word count")
        .saturating_mul(dispatcher_usize_to_u64(
            std::mem::size_of::<u32>(),
            "u32 byte width",
        ))
}

fn dispatcher_usize_to_u64<T>(value: T, label: &'static str) -> u64
where
    T: TryInto<u64> + Copy + std::fmt::Display,
    T::Error: std::fmt::Display,
{
    let _ = label;
    value.try_into().unwrap_or(u64::MAX)
}

fn dispatcher_abi_u32<T>(value: T, label: &'static str) -> u32
where
    T: TryInto<u32> + Copy + std::fmt::Display,
    T::Error: std::fmt::Display,
{
    let _ = label;
    value.try_into().unwrap_or(u32::MAX)
}

/// Observability returned from one batched dispatch.
#[derive(Debug, Clone)]
pub struct BatchDispatchReport {
    /// Sparse hit count written by the device (clamped to `hit_capacity`; the
    /// number of `hits` actually decodable).
    pub hit_count: u32,
    /// Matches the device produced BEYOND `hit_capacity` and therefore DROPPED
    /// from the hit ring (raw atomic head minus capacity). `> 0` means this
    /// dispatch's hit set is INCOMPLETE — a recall-critical overflow the caller
    /// MUST surface and recover (re-scan with a larger ring or on the host),
    /// never treat as a complete result. Zero on a healthy dispatch.
    pub dropped_hits: u32,
    /// Hits compacted out of the sparse ring.
    pub hits: Vec<HitRecord>,
    /// Work items processed by the queue.
    pub items_processed: u32,
    /// Wall-clock GPU execution time.
    pub wall_time: Duration,
    /// Rules that were isolated from the batch because their catalog entry was
    /// malformed. The rest of the batch still ran.
    pub rejected_rules: Vec<BatchRuleRejection>,
    /// Production telemetry for performance gates and dispatch tuning.
    pub telemetry: BatchDispatchTelemetry,
}

/// Megakernel dispatch counters returned when the caller owns hit storage.
#[derive(Debug, Clone)]
pub struct BatchDispatchSummary {
    /// Sparse hit count written by the device (clamped to `hit_capacity`; the
    /// number of `HitRecord`s decoded into the caller's storage).
    pub hit_count: u32,
    /// Matches the device produced BEYOND `hit_capacity` and therefore DROPPED
    /// from the hit ring (raw atomic head minus capacity). `> 0` means this
    /// dispatch's hit set is INCOMPLETE — a recall-critical overflow the caller
    /// MUST surface and recover, never treat as a complete result. Zero on a
    /// healthy dispatch.
    pub dropped_hits: u32,
    /// Work items processed by the queue.
    pub items_processed: u32,
    /// Wall-clock GPU execution time.
    pub wall_time: Duration,
    /// Rules that were isolated from the batch because their catalog entry was
    /// malformed. The rest of the batch still ran.
    pub rejected_rules: Vec<BatchRuleRejection>,
    /// Production telemetry for performance gates and dispatch tuning.
    pub telemetry: BatchDispatchTelemetry,
}

/// Megakernel dispatch counters used by scale/performance gates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BatchDispatchTelemetry {
    /// Bytes uploaded by this dispatch for rule-catalog refreshes.
    pub bytes_uploaded: u64,
    /// Bytes read back from queue-state and sparse hit output buffers.
    pub bytes_read_back: u64,
    /// Total host/device transfer bytes directly attributable to this dispatch.
    pub bytes_moved: u64,
    /// Resident allocations performed for refreshed rule-catalog buffers.
    pub resident_allocations: u32,
    /// Kernel launches submitted for the megakernel dispatch.
    pub kernel_launches: u32,
    /// Host-visible synchronization/readback wait points.
    pub sync_points: u32,
    /// Approximate lane occupancy in basis points, capped at 10000.
    pub occupancy_proxy_bps: u16,
    /// Active frontier density passed into the launch policy.
    pub frontier_density_bps: u16,
    /// Queue-state readback volume.
    pub queue_state_readback_bytes: u64,
    /// Sparse hit-ring readback volume.
    pub hit_readback_bytes: u64,
    /// Estimated peak device bytes required by the selected launch plan.
    pub estimated_peak_device_bytes: u64,
    /// Hard device-memory budget applied to this dispatch. Zero means unbounded.
    pub device_memory_budget_bytes: u64,
    /// Scale-aware topology selected by the launch policy.
    pub topology: MegakernelDispatchTopology,
    /// Whether this dispatch reused a cached fixed-batch launch plan.
    pub dispatch_plan_cache_hit: bool,
    /// Number of fixed-batch launch plans resident in the dispatcher cache.
    pub dispatch_plan_cache_entries: u16,
}

impl Default for BatchDispatchTelemetry {
    fn default() -> Self {
        Self {
            bytes_uploaded: 0,
            bytes_read_back: 0,
            bytes_moved: 0,
            resident_allocations: 0,
            kernel_launches: 0,
            sync_points: 0,
            occupancy_proxy_bps: 0,
            frontier_density_bps: 0,
            queue_state_readback_bytes: 0,
            hit_readback_bytes: 0,
            estimated_peak_device_bytes: 0,
            device_memory_budget_bytes: 0,
            topology: MegakernelDispatchTopology::SparseFrontier,
            dispatch_plan_cache_hit: false,
            dispatch_plan_cache_entries: 0,
        }
    }
}

struct RuleBufferUpdate {
    rejected_rules: Vec<BatchRuleRejection>,
    uploaded_bytes: u64,
    resident_allocations: u32,
}

const BATCH_PIPELINE_CACHE_CAP: usize = 32;

/// One compiled batched megakernel pipeline plus cached rule buffers.
pub struct BatchDispatcher {
    backend: WgpuBackend,
    config: BatchDispatchConfig,
    hit_writer: BatchHitWriter,
    pipeline: Arc<WgpuPipeline>,
    pipeline_cache: BatchPipelineCache,
    launch: MegakernelLaunchRecommendation,
    dispatch_plan_cache: BatchDispatchPlanCache,
    active_rule_fingerprints: Vec<[u8; 32]>,
    fingerprint_scratch: Vec<[u8; 32]>,
    fingerprint_occupied_scratch: Vec<bool>,
    fingerprint_addressed_scratch: Vec<bool>,
    rejection_scratch: Vec<BatchRuleRejection>,
    packing_scratch: RuleCatalogPackingScratch,
    rule_meta: Option<GpuBufferHandle>,
    transitions: Option<GpuBufferHandle>,
    accept: Option<GpuBufferHandle>,
    /// Shared byte→class maps (256 entries per unique DFA) backing the
    /// compressed transition tables. Uploaded alongside the other rule buffers.
    class_maps: Option<GpuBufferHandle>,
    queue_state_bytes: Vec<u8>,
    hit_bytes: Vec<u8>,
}

impl std::fmt::Debug for BatchDispatcher {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("BatchDispatcher")
            .field("config", &self.config)
            .field("hit_writer", &self.hit_writer)
            .field("pipeline_id", &self.pipeline.id())
            .field("launch", &self.launch)
            .field("rule_count", &self.active_rule_fingerprints.len())
            .finish()
    }
}

impl BatchDispatcher {
    /// Compile the batched megakernel program on a live wgpu backend.
    ///
    /// Defaults to the [`BatchHitWriter::Scalar`] hit writer. This is a
    /// CORRECTNESS requirement, not a performance default: the batch kernel's
    /// per-work-item scan (`dfa_byte_scanner`) loops `scan_start..emit_end`, so
    /// lanes in one subgroup execute DIFFERENT iteration counts (segments/files
    /// differ in length) and exit the loop at different points — divergent control
    /// flow.
    /// The hierarchical-subgroup writer aggregates hits with `subgroupBallot`/
    /// `subgroupAdd`/`subgroupShuffle` and elects a leader lane; under divergence
    /// the elected leader can already have exited, so its reserved ring slot is
    /// never broadcast and hits found by still-running lanes are dropped. That
    /// surfaced as a real, data-dependent recall loss in the keyhog GPU≡CPU
    /// parity gate (6 of 46 detector firings silently missed, every miss a match
    /// found after its subgroup's leader lane finished a shorter file). The
    /// scalar writer does one independent `atomicAdd` per hit and is correct
    /// under ANY divergence; for sparse credential matches the per-byte DFA step
    /// dominates and the extra atomics are negligible. Callers with a genuinely
    /// uniform-iteration kernel may opt into a subgroup writer via
    /// [`Self::new_with_hit_writer`].
    ///
    /// # Errors
    ///
    /// Returns [`PipelineError::Backend`] when pipeline compilation fails.
    pub fn new(backend: WgpuBackend, config: BatchDispatchConfig) -> Result<Self, PipelineError> {
        Self::new_with_hit_writer(backend, config, BatchHitWriter::Scalar)
    }

    /// Compile with an explicit sparse-hit publication algorithm.
    ///
    /// # Errors
    ///
    /// Returns [`PipelineError::Backend`] when hierarchical subgroup atomics are
    /// requested on a backend that reports no subgroup support, or when
    /// pipeline compilation fails.
    pub fn new_with_hit_writer(
        backend: WgpuBackend,
        mut config: BatchDispatchConfig,
        requested_hit_writer: BatchHitWriter,
    ) -> Result<Self, PipelineError> {
        if config.workgroup_size_x == 0 {
            return Err(PipelineError::QueueFull {
                queue: "submission",
                fix: "BatchDispatchConfig requires non-zero workgroup_size_x",
            });
        }
        let seed_queue_len = config
            .worker_groups
            .max(1)
            .checked_mul(config.workgroup_size_x)
            .ok_or_else(|| PipelineError::QueueFull {
                queue: "submission",
                fix: "megakernel seed queue length overflowed u32; reduce worker_groups or workgroup_size_x",
            })?;
        let launch = config.launch_recommendation(backend.device_limits(), seed_queue_len)?;
        if config.worker_groups == 0 {
            config.worker_groups = launch.worker_groups;
        }
        if config.hit_capacity == 0 {
            config.hit_capacity = launch.hit_capacity;
        }
        // The batch kernel's per-work-item scan (`dfa_byte_scanner`) loops
        // `scan_start..emit_end`, so subgroup lanes diverge as shorter
        // segments/files finish first. The hierarchical-subgroup writer aggregates hits with
        // subgroup ballot/add/shuffle and REQUIRES uniform control flow (see the
        // `hierarchical_atomics` module contract); under this divergence it
        // strands the elected leader's reserved ring slot once that lane exits,
        // silently dropping hits found by still-running lanes (a real recall loss
        // in the keyhog GPU≡CPU parity gate). So the hierarchical writer is never
        // sound for this dispatcher: `Auto` (which would resolve to Hierarchical
        // on a subgroup backend) DOWNGRADES to the correct scalar writer, and an
        // EXPLICIT hierarchical request is a caller error that fails loudly rather
        // than silently losing recall.
        let resolved = requested_hit_writer.resolve_for_backend(backend.supports_subgroup_ops())?;
        let hit_writer = match resolved {
            BatchHitWriter::HierarchicalSubgroup => {
                if matches!(requested_hit_writer, BatchHitWriter::Auto) {
                    BatchHitWriter::Scalar
                } else {
                    return Err(PipelineError::Backend(
                        "BatchHitWriter::HierarchicalSubgroup is unsound for the batched megakernel: \
                         its per-work-item DFA scan loops scan_start..emit_end, so subgroup lanes diverge \
                         as shorter segments/files finish, and subgroup hit-aggregation requires uniform \
                         control flow — under divergence the leader lane exits before broadcasting \
                         its reserved ring slot and hits are silently dropped (detector-firing recall \
                         loss). Fix: use BatchHitWriter::Scalar (the default) or BatchHitWriter::Auto."
                            .to_string(),
                    ));
                }
            }
            other => other,
        };
        let program = build_batch_program(
            config.workgroup_size_x,
            config.worker_groups,
            config.hit_capacity,
            hit_writer,
        );
        let pipeline = backend.compile_persistent(&program, &DispatchConfig::default())?;
        let pipeline_workgroup_size_x = config.workgroup_size_x;
        let pipeline_hit_capacity = config.hit_capacity;
        let mut pipeline_cache = BatchPipelineCache::with_cap(BATCH_PIPELINE_CACHE_CAP);
        pipeline_cache.seed(
            BatchPipelineShape {
                workgroup_size_x: pipeline_workgroup_size_x,
                worker_groups: launch.worker_groups,
                hit_capacity: pipeline_hit_capacity,
            },
            pipeline.clone(),
        );
        Ok(Self {
            backend,
            config,
            hit_writer,
            pipeline: pipeline.clone(),
            pipeline_cache,
            launch,
            dispatch_plan_cache: BatchDispatchPlanCache::default(),
            active_rule_fingerprints: Vec::new(),
            fingerprint_scratch: Vec::new(),
            fingerprint_occupied_scratch: Vec::new(),
            fingerprint_addressed_scratch: Vec::new(),
            rejection_scratch: Vec::new(),
            packing_scratch: RuleCatalogPackingScratch::default(),
            rule_meta: None,
            transitions: None,
            accept: None,
            class_maps: None,
            queue_state_bytes: Vec::with_capacity(QUEUE_STATE_WORDS * std::mem::size_of::<u32>()),
            hit_bytes: Vec::new(),
        })
    }

    /// Dispatch one `FileBatch` against many compiled DFA rules in one launch.
    ///
    /// # Errors
    ///
    /// Returns [`PipelineError::Backend`] on pipeline, upload, or readback
    /// failures.
    pub fn dispatch(
        &mut self,
        batch: &FileBatch,
        rules: &[BatchRuleProgram],
    ) -> Result<BatchDispatchReport, PipelineError> {
        let hit_capacity = usize::try_from(batch.hit_capacity()).map_err(|source| {
            PipelineError::Backend(format!(
                "batch hit capacity cannot fit usize: {source}. Fix: reduce hit_capacity or shard the batch."
            ))
        })?;
        let mut hits = Vec::with_capacity(hit_capacity);
        let summary = self.dispatch_into(batch, rules, &mut hits)?;
        Ok(BatchDispatchReport {
            hit_count: summary.hit_count,
            dropped_hits: summary.dropped_hits,
            hits,
            items_processed: summary.items_processed,
            wall_time: summary.wall_time,
            rejected_rules: summary.rejected_rules,
            telemetry: summary.telemetry,
        })
    }

    /// Dispatch one `FileBatch` while decoding sparse hits into caller-owned
    /// storage.
    ///
    /// Reusing `hits` avoids a fresh hit-vector allocation on hot repeated
    /// megakernel calls. The vector is cleared before decode and keeps its
    /// capacity unless the actual hit count exceeds it.
    ///
    /// # Errors
    ///
    /// Returns [`PipelineError::Backend`] on pipeline, upload, or readback
    /// failures.
    pub fn dispatch_into(
        &mut self,
        batch: &FileBatch,
        rules: &[BatchRuleProgram],
        hits: &mut Vec<HitRecord>,
    ) -> Result<BatchDispatchSummary, PipelineError> {
        if rules.is_empty() {
            hits.clear();
            let dynamic_plan = self.dispatch_plan(batch)?;
            return Ok(BatchDispatchSummary {
                hit_count: 0,
                dropped_hits: 0,
                items_processed: 0,
                wall_time: Duration::ZERO,
                rejected_rules: Vec::new(),
                telemetry: BatchDispatchTelemetry {
                    topology: dynamic_plan.plan.topology,
                    frontier_density_bps: self.config.frontier_density_bps,
                    estimated_peak_device_bytes: dynamic_plan.plan.estimated_peak_device_bytes,
                    device_memory_budget_bytes: dynamic_plan.plan.device_memory_budget_bytes,
                    dispatch_plan_cache_hit: dynamic_plan.cache_hit,
                    dispatch_plan_cache_entries: dynamic_plan.cache_entries,
                    ..BatchDispatchTelemetry::default()
                },
            });
        }
        let dynamic_plan = self.dispatch_plan(batch)?;
        let pipeline = self.pipeline_for_plan(dynamic_plan.plan)?;
        let rule_update = self.ensure_rule_buffers(rules)?;
        batch.reset_queue_state()?;

        let Some(class_maps) = self.class_maps.as_ref() else {
            return Err(PipelineError::Backend(
                "byte-class map buffer missing after ensure_rule_buffers. Fix: keep megakernel rule buffer initialization atomic.".to_string(),
            ));
        };
        let Some(rule_meta) = self.rule_meta.as_ref() else {
            return Err(PipelineError::Backend(
                "rule metadata buffer missing after ensure_rule_buffers. Fix: keep megakernel rule buffer initialization atomic.".to_string(),
            ));
        };
        let Some(transitions) = self.transitions.as_ref() else {
            return Err(PipelineError::Backend(
                "transition buffer missing after ensure_rule_buffers. Fix: keep megakernel rule buffer initialization atomic.".to_string(),
            ));
        };
        let Some(accept) = self.accept.as_ref() else {
            return Err(PipelineError::Backend(
                "accept buffer missing after ensure_rule_buffers. Fix: keep megakernel rule buffer initialization atomic.".to_string(),
            ));
        };
        // Input order MUST match the non-Shared storage buffer DECLARATION order
        // in `batch_program_buffers` (offsets, metadata, class_maps, haystack,
        // rule_meta, transitions, accept, segments), not literal binding numbers
        // — the persistent pipeline binds inputs positionally in that order.
        // `segments` is declared last so it is the final positional input.
        let inputs = [
            batch.offsets(),
            batch.metadata(),
            class_maps,
            batch.haystack(),
            rule_meta,
            transitions,
            accept,
            batch.segments(),
        ];
        let outputs = [batch.queue_state(), batch.hit_ring()];
        let start = Instant::now();
        pipeline.dispatch_persistent_borrowed(
            &inputs,
            &outputs,
            None,
            [dynamic_plan.plan.worker_groups, 1, 1],
        )?;

        let (device, queue) = &*self.backend.device_queue();
        wait_for_persistent_dispatch(device, start, self.config.timeout)?;
        let wall_time = start.elapsed();
        self.queue_state_bytes.clear();
        let queue_state_readback_bytes = batch_fixed_resident_overhead_bytes();
        batch.queue_state().readback_prefix(
            device,
            queue,
            queue_state_readback_bytes,
            &mut self.queue_state_bytes,
        )?;
        let queue_state_word_count =
            validate_u32_readback_words(&self.queue_state_bytes, "queue-state")?;
        if queue_state_word_count < QUEUE_STATE_WORDS {
            return Err(PipelineError::Backend(format!(
                "queue-state readback exposed {} words, expected at least {}. Fix: keep the queue-state buffer sized for every control word.",
                queue_state_word_count,
                QUEUE_STATE_WORDS
            )));
        }
        // The kernel `atomicAdd(HIT_HEAD, 1)`s for EVERY match it finds, then
        // writes only when `slot < hit_capacity` — so the raw head is the true
        // number of matches the device produced, which can exceed the ring. We
        // can only read back `hit_capacity` slots, but the overflow is a
        // recall-critical signal: clamping it away silently would hide dropped
        // matches (Law 10). Split the raw head into the readable count and the
        // dropped count and surface the latter to the caller.
        let raw_hit_head = read_u32_word(
            &self.queue_state_bytes,
            "queue-state",
            queue_state_word::HIT_HEAD,
        )?;
        let (hit_count, dropped_hits) = split_hit_overflow(raw_hit_head, batch.hit_capacity());
        let items_processed = read_u32_word(
            &self.queue_state_bytes,
            "queue-state",
            queue_state_word::DONE_COUNT,
        )?;

        // Fail-closed drain-completion guard (Law 10: no silent recall loss).
        //
        // The claim loop now DRAINS: every resident lane keeps issuing
        // `atomicAdd(HEAD, 1)` until it claims past the end of the queue, so after
        // a COMPLETE drain `HEAD == queue_len + resident_lanes >= queue_len` (one
        // past-the-end claim per lane). The only way `HEAD < queue_len` can be
        // observed is an INCOMPLETE drain — the dispatch was cut short (e.g. the
        // dispatch timeout fired) before the queue was exhausted, leaving the
        // indices `[HEAD, queue_len)` unscanned and their matches missing from the
        // ring with `dropped_hits == 0`: an INVISIBLE recall loss. (HEAD, not
        // DONE_COUNT: a claimed-but-rejected rule still advances HEAD, so HEAD is
        // the rejected-rule-independent "was every work-item handed out?" signal.)
        // Surface it loudly instead of returning a partial hit set.
        let claims_attempted = read_u32_word(
            &self.queue_state_bytes,
            "queue-state",
            queue_state_word::HEAD,
        )?;
        let expected_items = batch.queue_len();
        if claims_attempted < expected_items {
            return Err(PipelineError::Backend(format!(
                "megakernel drain incomplete: only {claims_attempted} of {expected_items} work-items were \
                 claimed before the dispatch ended, so {} work-item(s) went unscanned and their matches were \
                 dropped. This dispatch's hit set is INCOMPLETE. Fix: raise the dispatch timeout so the drain \
                 loop can exhaust the queue, or shard the batch into smaller queues.",
                expected_items.saturating_sub(claims_attempted)
            )));
        }

        self.hit_bytes.clear();
        let hit_readback_bytes = u64::from(hit_count)
            .checked_mul(dispatcher_usize_to_u64(
                HIT_RECORD_WORDS,
                "hit-record word count",
            ))
            .and_then(|words| {
                words.checked_mul(dispatcher_usize_to_u64(
                    std::mem::size_of::<u32>(),
                    "u32 byte width",
                ))
            })
            .ok_or_else(|| {
                PipelineError::Backend(
                    "hit-ring readback length overflowed u64. Fix: reduce hit_capacity or shard the batch."
                        .to_string(),
                )
            })?;
        batch
            .hit_ring()
            .readback_prefix(device, queue, hit_readback_bytes, &mut self.hit_bytes)?;
        decode_hits_from_readback_into(&self.hit_bytes, hit_count, hits)?;
        let bytes_read_back = queue_state_readback_bytes
            .checked_add(hit_readback_bytes)
            .ok_or_else(|| {
                PipelineError::Backend(
                    "batch readback byte accounting overflowed u64. Fix: shard the batch before readback."
                        .to_string(),
                )
            })?;
        let bytes_moved = rule_update
            .uploaded_bytes
            .checked_add(bytes_read_back)
            .ok_or_else(|| {
                PipelineError::Backend(
                    "batch moved-byte accounting overflowed u64. Fix: shard the batch before dispatch."
                        .to_string(),
                )
            })?;

        Ok(BatchDispatchSummary {
            hit_count,
            dropped_hits,
            items_processed,
            wall_time,
            rejected_rules: rule_update.rejected_rules,
            telemetry: BatchDispatchTelemetry {
                bytes_uploaded: rule_update.uploaded_bytes,
                bytes_read_back,
                bytes_moved,
                resident_allocations: rule_update.resident_allocations,
                kernel_launches: 1,
                sync_points: 2,
                occupancy_proxy_bps: occupancy_proxy_bps(
                    items_processed,
                    dynamic_plan.plan.worker_groups,
                    self.config.workgroup_size_x,
                ),
                frontier_density_bps: self.config.frontier_density_bps,
                queue_state_readback_bytes,
                hit_readback_bytes,
                estimated_peak_device_bytes: dynamic_plan.plan.estimated_peak_device_bytes,
                device_memory_budget_bytes: dynamic_plan.plan.device_memory_budget_bytes,
                topology: dynamic_plan.plan.topology,
                dispatch_plan_cache_hit: dynamic_plan.cache_hit,
                dispatch_plan_cache_entries: dynamic_plan.cache_entries,
            },
        })
    }

    fn pipeline_for_plan(
        &mut self,
        plan: BatchDispatchPlan,
    ) -> Result<Arc<WgpuPipeline>, PipelineError> {
        let shape = BatchPipelineShape {
            workgroup_size_x: plan.workgroup_size_x,
            worker_groups: plan.worker_groups,
            hit_capacity: plan.hit_capacity,
        };
        if let Some(pipeline) = self.pipeline_cache.get(shape) {
            return Ok(pipeline);
        }
        let program = build_batch_program(
            plan.workgroup_size_x,
            plan.worker_groups,
            plan.hit_capacity,
            self.hit_writer,
        );
        let pipeline = self
            .backend
            .compile_persistent(&program, &DispatchConfig::default())?;
        self.pipeline_cache.insert(shape, pipeline.clone());
        Ok(pipeline)
    }

    fn dispatch_plan(
        &mut self,
        batch: &FileBatch,
    ) -> Result<BatchDispatchPlanLookup, PipelineError> {
        let queue_len = batch.queue_len();
        if let Some(plan) = self.dispatch_plan_cache.get(queue_len) {
            return Ok(BatchDispatchPlanLookup {
                plan,
                cache_hit: true,
                cache_entries: self.dispatch_plan_cache.len_u16(),
            });
        }
        let mut recommendation = self
            .config
            .launch_recommendation(self.backend.device_limits(), queue_len)?;
        let resident_hit_capacity = batch.hit_capacity();
        if recommendation.hit_capacity > resident_hit_capacity {
            let removed_hit_bytes = u64::from(recommendation.hit_capacity - resident_hit_capacity)
                .checked_mul(dispatcher_usize_to_u64(
                    HIT_RECORD_WORDS,
                    "hit-record word count",
                ))
                .and_then(|words| {
                    words.checked_mul(dispatcher_usize_to_u64(
                        std::mem::size_of::<u32>(),
                        "u32 byte width",
                    ))
                })
                .ok_or_else(|| {
                    PipelineError::Backend(
                        "resident hit-capacity byte adjustment overflowed u64. Fix: shard the batch before dispatch planning."
                            .to_string(),
                    )
                })?;
            recommendation.hit_capacity = resident_hit_capacity;
            recommendation.estimated_peak_device_bytes = recommendation
                .estimated_peak_device_bytes
                .checked_sub(removed_hit_bytes)
                .ok_or_else(|| {
                    PipelineError::Backend(
                        "resident hit-capacity adjustment exceeded peak device estimate. Fix: keep launch recommendation and resident batch capacity synchronized."
                            .to_string(),
                    )
                })?;
        }
        let plan = BatchDispatchPlan::from_recommendation(queue_len, &self.config, recommendation);
        self.dispatch_plan_cache.insert(plan);
        Ok(BatchDispatchPlanLookup {
            plan,
            cache_hit: false,
            cache_entries: self.dispatch_plan_cache.len_u16(),
        })
    }

    fn ensure_rule_buffers(
        &mut self,
        rules: &[BatchRuleProgram],
    ) -> Result<RuleBufferUpdate, PipelineError> {
        accepted_rule_fingerprints_and_rejections_into(
            rules,
            &mut self.fingerprint_scratch,
            &mut self.fingerprint_occupied_scratch,
            &mut self.fingerprint_addressed_scratch,
            &mut self.rejection_scratch,
        );
        if self.fingerprint_scratch == self.active_rule_fingerprints {
            return Ok(RuleBufferUpdate {
                rejected_rules: if self.rejection_scratch.is_empty() {
                    Vec::new()
                } else {
                    self.rejection_scratch.clone()
                },
                uploaded_bytes: 0,
                resident_allocations: 0,
            });
        }

        pack_rule_catalog_into(rules, &mut self.packing_scratch)?;
        // rule_meta words = entries * RULE_META_WORDS (each RuleMeta is
        // RULE_META_WORDS u32s); transitions + accept + class_maps are flat u32
        // vecs. Account for all four uploaded device buffers.
        let rule_meta_words = self
            .packing_scratch
            .rule_meta
            .len()
            .checked_mul(RULE_META_WORDS)
            .ok_or_else(|| {
                PipelineError::Backend(
                    "rule metadata upload word count overflowed usize. Fix: shard the rule catalog before upload."
                        .to_string(),
                )
            })?;
        let uploaded_words = rule_meta_words
            .checked_add(self.packing_scratch.transitions.len())
            .and_then(|words| words.checked_add(self.packing_scratch.accept.len()))
            .and_then(|words| words.checked_add(self.packing_scratch.class_maps.len()))
            .ok_or_else(|| {
                PipelineError::Backend(
                    "rule catalog upload word count overflowed usize. Fix: shard the rule catalog before upload."
                        .to_string(),
                )
            })?;
        let uploaded_bytes = uploaded_words
            .checked_mul(std::mem::size_of::<u32>())
            .and_then(|bytes| u64::try_from(bytes).ok())
            .ok_or_else(|| {
                PipelineError::Backend(
                    "rule catalog upload byte count overflowed u64. Fix: shard the rule catalog before upload."
                        .to_string(),
                )
            })?;
        let (device, queue) = &*self.backend.device_queue();
        self.rule_meta = Some(GpuBufferHandle::upload(
            device,
            queue,
            bytemuck::cast_slice(&self.packing_scratch.rule_meta),
            persistent_storage_binding_usage(),
        )?);
        self.transitions = Some(GpuBufferHandle::upload(
            device,
            queue,
            bytemuck::cast_slice(&self.packing_scratch.transitions),
            persistent_storage_binding_usage(),
        )?);
        self.accept = Some(GpuBufferHandle::upload(
            device,
            queue,
            bytemuck::cast_slice(&self.packing_scratch.accept),
            persistent_storage_binding_usage(),
        )?);
        self.class_maps = Some(GpuBufferHandle::upload(
            device,
            queue,
            bytemuck::cast_slice(&self.packing_scratch.class_maps),
            persistent_storage_binding_usage(),
        )?);
        if self.active_rule_fingerprints.len() == self.fingerprint_scratch.len() {
            self.active_rule_fingerprints
                .copy_from_slice(&self.fingerprint_scratch);
        } else {
            self.active_rule_fingerprints.clear();
            self.active_rule_fingerprints
                .extend_from_slice(&self.fingerprint_scratch);
        }
        Ok(RuleBufferUpdate {
            rejected_rules: if self.packing_scratch.rejected_rules.is_empty() {
                Vec::new()
            } else {
                self.packing_scratch.rejected_rules.clone()
            },
            uploaded_bytes,
            resident_allocations: 4,
        })
    }
}

fn occupancy_proxy_bps(items_processed: u32, worker_groups: u32, workgroup_size_x: u32) -> u16 {
    let lanes = u64::from(worker_groups.max(1))
        .checked_mul(u64::from(workgroup_size_x.max(1)))
        .unwrap_or(u64::MAX);
    crate::numeric::ratio_basis_points_u64_wide(
        u64::from(items_processed),
        lanes.max(1),
        0,
        "batch occupancy proxy",
    )
    .min(10_000) as u16
}

fn validate_u32_readback_words(bytes: &[u8], label: &'static str) -> Result<usize, PipelineError> {
    if bytes.len() % std::mem::size_of::<u32>() != 0 {
        return Err(PipelineError::Backend(format!(
            "{label} readback exposed {} bytes, which is not a whole number of u32 words. Fix: keep readback lengths 4-byte aligned.",
            bytes.len()
        )));
    }
    Ok(bytes.len() / std::mem::size_of::<u32>())
}

fn read_u32_word(
    bytes: &[u8],
    label: &'static str,
    word_index: usize,
) -> Result<u32, PipelineError> {
    let offset = word_index
        .checked_mul(std::mem::size_of::<u32>())
        .ok_or_else(|| {
            PipelineError::Backend(format!(
                "{label} word offset overflowed usize. Fix: split the readback before decoding."
            ))
        })?;
    let word = bytes.get(offset..offset + std::mem::size_of::<u32>()).ok_or_else(|| {
        PipelineError::Backend(format!(
            "{label} readback is missing u32 word {word_index}. Fix: request a large enough readback prefix."
        ))
    })?;
    Ok(u32::from_le_bytes([word[0], word[1], word[2], word[3]]))
}

fn wait_for_persistent_dispatch(
    device: &wgpu::Device,
    start: Instant,
    timeout: Duration,
) -> Result<(), PipelineError> {
    let mut backoff = crate::wait_backoff::AdaptiveWaitBackoff::from_micros(64, 5, 50, 8);
    loop {
        if crate::runtime::device::poll_device_once(device)
            .map_err(|error| PipelineError::Backend(error.to_string()))?
            .is_queue_empty()
        {
            return Ok(());
        }
        let elapsed = start.elapsed();
        if elapsed >= timeout {
            return Err(PipelineError::Backend(format!(
                "batch megakernel dispatch exceeded timeout before readback: took {elapsed:?}, budget {timeout:?}. Fix: raise BatchDispatchConfig.timeout or split the batch.",
            )));
        }
        let remaining = timeout.checked_sub(elapsed).ok_or_else(|| {
            PipelineError::Backend(format!(
                "batch megakernel timeout arithmetic underflowed after elapsed {elapsed:?} exceeded budget {timeout:?}. Fix: split the batch or raise BatchDispatchConfig.timeout deliberately.",
            ))
        })?;
        backoff.idle_for(remaining);
    }
}

fn build_batch_program(
    workgroup_size_x: u32,
    worker_groups: u32,
    hit_capacity: u32,
    hit_writer: BatchHitWriter,
) -> Program {
    // Persistent DRAIN loop: every resident lane keeps claiming work-items with
    // `atomicAdd(HEAD, 1)` until its claim lands past the end of the queue
    // (`claim >= QUEUE_LEN`), then returns. This drains the full
    // `segment_count * rule_count` queue for ANY number of resident lanes.
    //
    // It replaces a fixed `claim_budget = ceil(QUEUE_LEN / total_workers)` loop
    // that assumed exactly `total_workers` lanes each ran their full budget. When
    // fewer lanes were actually resident than that budget assumed, the queue was
    // never fully claimed — `found < expected` with `dropped_hits == 0`: a SILENT
    // recall loss (Law 10). The drain removes the dependency on the resident-lane
    // count entirely. Overhead is one extra past-the-end `atomicAdd` per resident
    // lane (the claim that observes `>= QUEUE_LEN` and returns), NOT per
    // work-item — a rounding error, not a 1/queue_len-scale pessimization.
    //
    // `worker_groups` now sizes only the dispatch grid (more resident lanes =
    // more parallelism); kernel correctness no longer depends on it.
    let _ = worker_groups;
    let queue_len = atomic_load_relaxed(
        "queue_state",
        Expr::u32(dispatcher_abi_u32(
            queue_state_word::QUEUE_LEN,
            "queue-state length word",
        )),
    );
    let mut loop_body = vec![
        Node::let_bind(
            "claim",
            Expr::atomic_add(
                "queue_state",
                Expr::u32(dispatcher_abi_u32(
                    queue_state_word::HEAD,
                    "queue-state head word",
                )),
                Expr::u32(1),
            ),
        ),
        // Past-the-end claim ⇒ the queue is drained for this lane. `Return` exits
        // the kernel: safe because the drain loop is the only top-level statement
        // (no post-loop finalization to skip) and `execute_batch_claim_body`
        // contains no workgroup barrier (no divergence deadlock).
        Node::if_then(
            Expr::ge(Expr::var("claim"), queue_len),
            vec![Node::Return],
        ),
    ];
    loop_body.extend(execute_batch_claim_body(hit_writer));

    Program::wrapped(
        batch_program_buffers(hit_capacity),
        [workgroup_size_x, 1, 1],
        vec![Node::forever(loop_body)],
    )
}

fn batch_program_buffers(hit_capacity: u32) -> Vec<BufferDecl> {
    let hit_ring_words = hit_capacity.saturating_mul(4);
    vec![
        BufferDecl::storage("file_offsets", 0, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("file_metadata", 1, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("class_maps", 2, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("haystack", 3, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("rule_meta", 4, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("transitions", 5, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("accept", 6, BufferAccess::ReadOnly, DataType::U32),
        BufferDecl::storage("queue_state", 7, BufferAccess::ReadWrite, DataType::U32).with_count(
            dispatcher_abi_u32(QUEUE_STATE_WORDS, "queue-state word count"),
        ),
        BufferDecl::output("hit_ring", 8, DataType::U32).with_count(hit_ring_words),
        // Flat segment table (`segment_count * SEGMENT_WORDS` u32s). Declared
        // LAST among the read-only inputs so it occupies the final positional
        // input slot (the persistent pipeline binds non-output buffers to
        // `inputs[]` in declaration order — see `dispatch_persistent_borrowed`).
        // The kernel reads row `seg_idx = claim / rule_count` to derive the
        // window; the host sizes the queue from the same table so a claim never
        // indexes past it.
        BufferDecl::storage("segments", 9, BufferAccess::ReadOnly, DataType::U32),
    ]
}

fn execute_batch_claim_body(hit_writer: BatchHitWriter) -> Vec<Node> {
    vec![
        Node::let_bind(
            "rule_count",
            atomic_load_relaxed(
                "queue_state",
                Expr::u32(dispatcher_abi_u32(
                    queue_state_word::RULE_COUNT,
                    "queue-state rule-count word",
                )),
            ),
        ),
        // A claim decodes to `(seg_idx, rule_idx)`. `seg_idx` indexes the flat
        // `segments` table; each 4-word row is `[file_idx, scan_start, emit_start,
        // emit_end]` with FILE-RELATIVE offsets (see `segmentation::Segment`). The
        // dense default (one segment per file, `seg_len = u32::MAX`) makes the row
        // `[file_idx, 0, 0, file_len]`, so the window is the whole file and this
        // path is byte-for-byte the legacy `file_idx = claim / rule_count` scan.
        Node::let_bind(
            "seg_idx",
            Expr::div(Expr::var("claim"), Expr::var("rule_count")),
        ),
        Node::let_bind(
            "rule_idx",
            Expr::rem(Expr::var("claim"), Expr::var("rule_count")),
        ),
        Node::let_bind(
            "seg_base",
            Expr::mul(
                Expr::var("seg_idx"),
                Expr::u32(dispatcher_abi_u32(SEGMENT_WORDS, "segment table word count")),
            ),
        ),
        Node::let_bind("file_idx", Expr::load("segments", Expr::var("seg_base"))),
        Node::let_bind(
            "scan_start_rel",
            Expr::load("segments", Expr::add(Expr::var("seg_base"), Expr::u32(1))),
        ),
        Node::let_bind(
            "emit_start_rel",
            Expr::load("segments", Expr::add(Expr::var("seg_base"), Expr::u32(2))),
        ),
        Node::let_bind(
            "emit_end_rel",
            Expr::load("segments", Expr::add(Expr::var("seg_base"), Expr::u32(3))),
        ),
        Node::let_bind(
            "metadata_base",
            Expr::mul(
                Expr::var("file_idx"),
                Expr::u32(dispatcher_abi_u32(
                    FILE_METADATA_WORDS,
                    "file metadata word count",
                )),
            ),
        ),
        Node::let_bind(
            "layer_idx",
            Expr::load(
                "file_metadata",
                Expr::add(Expr::var("metadata_base"), Expr::u32(3)),
            ),
        ),
        Node::let_bind(
            "file_start",
            Expr::load("file_offsets", Expr::var("file_idx")),
        ),
        // Absolute (packed-haystack) window bounds: file base + file-relative
        // segment offsets. `scan_start <= emit_start < emit_end` by construction.
        Node::let_bind(
            "scan_start",
            Expr::add(Expr::var("file_start"), Expr::var("scan_start_rel")),
        ),
        Node::let_bind(
            "emit_start",
            Expr::add(Expr::var("file_start"), Expr::var("emit_start_rel")),
        ),
        Node::let_bind(
            "emit_end",
            Expr::add(Expr::var("file_start"), Expr::var("emit_end_rel")),
        ),
        Node::let_bind(
            "rule_base",
            Expr::mul(
                Expr::var("rule_idx"),
                Expr::u32(dispatcher_abi_u32(
                    RULE_META_WORDS,
                    "rule metadata word count",
                )),
            ),
        ),
        Node::let_bind(
            "transition_base",
            Expr::load("rule_meta", Expr::var("rule_base")),
        ),
        Node::let_bind(
            "accept_base",
            Expr::load("rule_meta", Expr::add(Expr::var("rule_base"), Expr::u32(1))),
        ),
        // Byte-class compression metadata (rule_meta words 3 and 4): the
        // per-rule 256-entry byte->class map base and the compressed row width.
        Node::let_bind(
            "class_map_base",
            Expr::load("rule_meta", Expr::add(Expr::var("rule_base"), Expr::u32(3))),
        ),
        Node::let_bind(
            "num_classes",
            Expr::load("rule_meta", Expr::add(Expr::var("rule_base"), Expr::u32(4))),
        ),
        // Delegate core evaluation to Tier-2 LEGO Primitive
        Node::Block(dfa_byte_scanner(hit_writer)),
        // Mark work completion
        Node::let_bind(
            "done_prev",
            Expr::atomic_add(
                "queue_state",
                Expr::u32(dispatcher_abi_u32(
                    queue_state_word::DONE_COUNT,
                    "queue-state done-count word",
                )),
                Expr::u32(1),
            ),
        ),
    ]
}

fn dfa_byte_scanner(hit_writer: BatchHitWriter) -> Vec<Node> {
    vec![
        Node::let_bind("state", Expr::u32(0)),
        // Scan the window `[scan_start, emit_end)` from state 0. The
        // `[scan_start, emit_start)` prefix is DFA warm-up — it advances the
        // state but emits nothing (the emit guard below). For the dense default
        // `scan_start == emit_start == file_start`, so the loop is the whole file
        // with no warm-up — identical to the pre-segmentation scan.
        Node::loop_for(
            "byte_pos",
            Expr::var("scan_start"),
            Expr::var("emit_end"),
            vec![
                Node::let_bind(
                    "haystack_word_index",
                    Expr::div(Expr::var("byte_pos"), Expr::u32(4)),
                ),
                Node::let_bind(
                    "haystack_shift",
                    Expr::mul(Expr::rem(Expr::var("byte_pos"), Expr::u32(4)), Expr::u32(8)),
                ),
                Node::let_bind(
                    "byte",
                    Expr::bitand(
                        Expr::shr(
                            Expr::load("haystack", Expr::var("haystack_word_index")),
                            Expr::var("haystack_shift"),
                        ),
                        Expr::u32(0xFF),
                    ),
                ),
                // Byte-class compressed transition load (lossless): fold the
                // byte through this rule's 256-entry class map, then index the
                // compressed `state * num_classes + class` row. Bytes that share
                // a transition column across every state collapse to one class,
                // shrinking each per-state row from 256 words to `num_classes`
                // words. Firings are byte-for-byte identical to the dense
                // `state * 256 + byte` table (proved in the CPU parity tests).
                Node::let_bind(
                    "byte_class",
                    Expr::load(
                        "class_maps",
                        Expr::add(Expr::var("class_map_base"), Expr::var("byte")),
                    ),
                ),
                Node::assign(
                    "state",
                    Expr::load(
                        "transitions",
                        Expr::add(
                            Expr::var("transition_base"),
                            Expr::add(
                                Expr::mul(Expr::var("state"), Expr::var("num_classes")),
                                Expr::var("byte_class"),
                            ),
                        ),
                    ),
                ),
                Node::let_bind(
                    "accepting",
                    Expr::load(
                        "accept",
                        Expr::add(Expr::var("accept_base"), Expr::var("state")),
                    ),
                ),
                // Emit guard mirrors the CPU parity oracle (`segmentation.rs`):
                // a match is owned by this window iff its end offset lies in
                // `[emit_start, emit_end)`. The loop bound already enforces
                // `byte_pos < emit_end`; the remaining condition `end > emit_start`
                // (end = byte_pos + 1) is exactly `byte_pos >= emit_start`. Bytes in
                // the warm-up prefix (`byte_pos < emit_start`) advance state but
                // never emit, so adjacent windows tile each file with no double
                // count and no miss.
                Node::let_bind(
                    "is_hit",
                    Expr::and(
                        Expr::ne(Expr::var("accepting"), Expr::u32(0)),
                        Expr::ge(Expr::var("byte_pos"), Expr::var("emit_start")),
                    ),
                ),
                hit_writer_node(hit_writer),
            ],
        ),
    ]
}

fn hit_writer_node(hit_writer: BatchHitWriter) -> Node {
    match hit_writer {
        BatchHitWriter::HierarchicalSubgroup => {
            Node::Block(record_hit_to_ring_hierarchical("is_hit"))
        }
        BatchHitWriter::Auto | BatchHitWriter::Scalar => {
            Node::if_then(Expr::var("is_hit"), record_hit_to_ring())
        }
    }
}

fn record_hit_to_ring() -> Vec<Node> {
    vec![
        Node::let_bind(
            "hit_slot",
            Expr::atomic_add(
                "queue_state",
                Expr::u32(dispatcher_abi_u32(
                    queue_state_word::HIT_HEAD,
                    "queue-state hit-head word",
                )),
                Expr::u32(1),
            ),
        ),
        Node::if_then(
            Expr::lt(
                Expr::var("hit_slot"),
                atomic_load_relaxed(
                    "queue_state",
                    Expr::u32(dispatcher_abi_u32(
                        queue_state_word::HIT_CAPACITY,
                        "queue-state hit-capacity word",
                    )),
                ),
            ),
            vec![
                Node::let_bind("hit_base", Expr::mul(Expr::var("hit_slot"), Expr::u32(4))),
                Node::store("hit_ring", Expr::var("hit_base"), Expr::var("file_idx")),
                Node::store(
                    "hit_ring",
                    Expr::add(Expr::var("hit_base"), Expr::u32(1)),
                    Expr::var("rule_idx"),
                ),
                Node::store(
                    "hit_ring",
                    Expr::add(Expr::var("hit_base"), Expr::u32(2)),
                    Expr::var("layer_idx"),
                ),
                Node::store(
                    "hit_ring",
                    Expr::add(Expr::var("hit_base"), Expr::u32(3)),
                    Expr::sub(Expr::var("byte_pos"), Expr::var("file_start")),
                ),
            ],
        ),
    ]
}

/// Split the device's raw atomic hit-head into `(readable, dropped)`.
///
/// The kernel increments `HIT_HEAD` for EVERY match but only writes ring slots
/// below `hit_capacity`, so a `raw_head > capacity` means `raw_head - capacity`
/// matches were produced-but-dropped. `readable` is what can be decoded from the
/// ring (`min(raw_head, capacity)`); `dropped` is the overflow the caller must
/// recover. Pure so the overflow accounting is unit-tested without a device.
const fn split_hit_overflow(raw_head: u32, capacity: u32) -> (u32, u32) {
    if raw_head > capacity {
        (capacity, raw_head - capacity)
    } else {
        (raw_head, 0)
    }
}

#[cfg(test)]
fn decode_hits_from_readback(
    bytes: &[u8],
    hit_count: u32,
) -> Result<Vec<HitRecord>, PipelineError> {
    let mut hits = Vec::new();
    decode_hits_from_readback_into(bytes, hit_count, &mut hits)?;
    Ok(hits)
}

fn decode_hits_from_readback_into(
    bytes: &[u8],
    hit_count: u32,
    hits: &mut Vec<HitRecord>,
) -> Result<(), PipelineError> {
    let word_count = validate_u32_readback_words(bytes, "hit-ring")?;
    let needed_words = usize::try_from(hit_count)
        .ok()
        .and_then(|count| count.checked_mul(4))
        .ok_or_else(|| PipelineError::Backend("hit-count overflowed usize".to_string()))?;
    if word_count < needed_words {
        return Err(PipelineError::Backend(format!(
            "hit-ring exposed {} words, expected at least {needed_words}. Fix: size the sparse hit ring for the configured hit_capacity.",
            word_count
        )));
    }
    let needed_bytes = needed_words
        .checked_mul(std::mem::size_of::<u32>())
        .ok_or_else(|| PipelineError::Backend(
            "hit-ring readback byte count overflowed usize. Fix: reduce hit_capacity or shard the batch."
                .to_string(),
        ))?;
    let hit_count = usize::try_from(hit_count).map_err(|source| {
        PipelineError::Backend(format!(
            "hit count cannot fit usize for host decode: {source}. Fix: reduce hit_capacity or run on a supported host pointer width."
        ))
    })?;
    let same_len = hits.len() == hit_count;
    if !same_len {
        hits.clear();
    }
    if hits.capacity() < hit_count {
        hits.try_reserve_exact(hit_count - hits.len())
            .map_err(|source| {
                PipelineError::Backend(format!(
                    "hit-ring decode could not reserve {hit_count} HitRecord slots: {source}. Fix: lower hit_capacity or shard the batch."
                ))
            })?;
    }
    if cfg!(target_endian = "little") {
        let record_bytes = std::mem::size_of::<HitRecord>();
        let expected_record_bytes = HIT_RECORD_WORDS * std::mem::size_of::<u32>();
        if record_bytes != expected_record_bytes {
            return Err(PipelineError::Backend(format!(
                "hit-ring host record layout is {record_bytes} bytes, expected {expected_record_bytes}. Fix: keep HitRecord as four packed u32 words."
            )));
        }
        if hit_count != 0 {
            let records: &[HitRecord] =
                bytemuck::try_cast_slice(&bytes[..needed_bytes]).map_err(|source| {
                    PipelineError::Backend(format!(
                        "hit-ring readback bytes were not aligned as HitRecord records: {source}. Fix: keep the hit ring byte layout aligned to four u32 words."
                    ))
                })?;
            if same_len {
                hits.copy_from_slice(records);
            } else {
                hits.extend_from_slice(records);
            }
        }
        return Ok(());
    }
    for (index, chunk) in bytes[..needed_bytes]
        .chunks_exact(HIT_RECORD_WORDS * std::mem::size_of::<u32>())
        .enumerate()
    {
        let record = HitRecord {
            file_idx: u32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]),
            rule_idx: u32::from_le_bytes([chunk[4], chunk[5], chunk[6], chunk[7]]),
            layer_idx: u32::from_le_bytes([chunk[8], chunk[9], chunk[10], chunk[11]]),
            match_offset: u32::from_le_bytes([chunk[12], chunk[13], chunk[14], chunk[15]]),
        };
        if same_len {
            hits[index] = record;
        } else {
            hits.push(record);
        }
    }
    Ok(())
}

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

    #[test]
    fn hit_overflow_split_reports_dropped_matches() {
        // No overflow: every produced match fits the ring.
        assert_eq!(split_hit_overflow(0, 1_000), (0, 0));
        assert_eq!(split_hit_overflow(254, 1_000), (254, 0));
        // Exactly full: readable == capacity, nothing dropped.
        assert_eq!(split_hit_overflow(1_000, 1_000), (1_000, 0));
        // Overflow: readable clamps to capacity, the rest are reported dropped —
        // the recall-critical signal the old `.min()` clamp threw away.
        assert_eq!(split_hit_overflow(1_001, 1_000), (1_000, 1));
        assert_eq!(split_hit_overflow(1_500_000, 1_000_000), (1_000_000, 500_000));
        // Saturated raw head (kernel produced u32::MAX-worth of matches).
        assert_eq!(split_hit_overflow(u32::MAX, 1_000), (1_000, u32::MAX - 1_000));
    }

    #[test]
    fn default_worker_groups_is_at_least_four_on_live_adapter() {
        if let Ok(backend) = WgpuBackend::new() {
            let wg = BatchDispatchConfig::default()
                .launch_recommendation(backend.device_limits(), 64)
                .expect("Fix: live adapter limits must produce a launch recommendation")
                .worker_groups;
            assert!(
                wg >= 4,
                "Fix: default worker_groups should be >= 4 on any live adapter, got {wg}"
            );
        }
    }

    #[test]
    fn launch_recommendation_is_consumed_for_worker_groups_and_hit_capacity() {
        let src = include_str!("dispatcher.rs");
        let prod_src = src.split("#[cfg(test)]").next().unwrap_or(src);
        assert!(
            prod_src.contains("config.worker_groups = launch.worker_groups"),
            "BatchDispatcher::new must consume launch policy worker group recommendations"
        );
        assert!(
            prod_src.contains("config.hit_capacity = launch.hit_capacity"),
            "BatchDispatcher::new must consume launch policy hit-capacity recommendations"
        );
    }

    #[test]
    fn dynamic_dispatch_plan_controls_pipeline_and_launch_geometry() {
        let src = include_str!("dispatcher.rs");
        let prod_src = src.split("#[cfg(test)]").next().unwrap_or(src);
        assert!(
            prod_src.contains("let pipeline = self.pipeline_for_plan(dynamic_plan.plan)?"),
            "dispatch must compile or reuse the pipeline for the per-batch scale-aware plan"
        );
        assert!(
            prod_src.contains("[dynamic_plan.plan.worker_groups, 1, 1]"),
            "dispatch must submit the policy-selected worker group count, not config.worker_groups"
        );
        assert!(
            prod_src.contains("dynamic_plan.plan.worker_groups,\n                    self.config.workgroup_size_x"),
            "occupancy telemetry must use the actual dynamic launch geometry"
        );
    }

    #[test]
    fn dynamic_pipeline_cache_is_bounded_lru() {
        let src = include_str!("dispatcher.rs");
        let prod_src = src.split("#[cfg(test)]").next().unwrap_or(src);
        assert!(
            prod_src.contains("const BATCH_PIPELINE_CACHE_CAP: usize = 32"),
            "scale-aware pipeline variants must have a fixed retention bound"
        );
        assert!(
            prod_src.contains("BatchPipelineCache::with_cap(BATCH_PIPELINE_CACHE_CAP)")
                && prod_src.contains("self.pipeline_cache.get(shape)")
                && prod_src.contains("self.pipeline_cache.insert(shape, pipeline.clone())")
                && !prod_src.contains("min_by_key(|(_, entry)| entry.last_seen)")
                && !prod_src.contains("swap_remove(evict_idx)"),
            "scale-aware pipeline cache must use the indexed heap-backed LRU instead of scanning entries"
        );
        assert!(
            prod_src.contains("workgroup_size_x: plan.workgroup_size_x")
                && prod_src.contains("worker_groups: plan.worker_groups")
                && prod_src.contains("hit_capacity: plan.hit_capacity"),
            "scale-aware pipeline cache key must include every program-shaping field"
        );
    }

    #[test]
    fn dynamic_plan_hit_capacity_is_clamped_to_resident_batch_ring() {
        let src = include_str!("dispatcher.rs");
        let prod_src = src.split("#[cfg(test)]").next().unwrap_or(src);
        assert!(
            prod_src.contains("let resident_hit_capacity = batch.hit_capacity()")
                && prod_src.contains("recommendation.hit_capacity = resident_hit_capacity")
                && prod_src.contains("estimated_peak_device_bytes"),
            "dynamic dispatch plans must not compile a hit-ring shape larger than the resident FileBatch output buffer"
        );
    }

    #[test]
    fn launch_recommendation_uses_explicit_graph_hints_for_topology() {
        let limits = wgpu::Limits::default();
        let config = BatchDispatchConfig::default()
            .with_graph_hints(8192, 131_072, 9_000, 0)
            .with_execution_hints(8, 0, 0, 0);

        let rec = config
            .launch_recommendation(&limits, 8192)
            .expect("Fix: explicit graph hints must produce a launch recommendation");

        assert_eq!(rec.topology, MegakernelDispatchTopology::FusedDense);
    }

    #[test]
    fn launch_recommendation_default_does_not_invent_dense_frontier() {
        let limits = wgpu::Limits::default();
        let rec = BatchDispatchConfig::default()
            .launch_recommendation(&limits, 8192)
            .expect("Fix: default graph hints must produce a launch recommendation");

        assert_ne!(rec.topology, MegakernelDispatchTopology::FusedDense);
        assert_eq!(BatchDispatchConfig::default().frontier_density_bps, 0);
    }

    #[test]
    fn timeout_field_is_plumbed_into_dispatch_path() {
        let src = include_str!("dispatcher.rs");
        let prod_src = src.split("#[cfg(test)]").next().unwrap_or(src);
        assert!(
            prod_src.contains("timeout"),
            "BatchDispatchConfig exposes timeout; this test documents that it must stay wired"
        );
        assert!(
            prod_src.contains("dispatch_config.timeout")
                || prod_src.contains(".with_timeout(")
                || prod_src.contains("config.timeout"),
            "BatchDispatchConfig.timeout appears publicly configurable but is not consumed during dispatch"
        );
    }

    #[test]
    fn hit_readback_decodes_without_intermediate_word_vector() {
        let mut bytes = Vec::new();
        for word in [7u32, 3, 2, 99, 8, 4, 1, 100] {
            bytes.extend_from_slice(&word.to_le_bytes());
        }

        let hits = decode_hits_from_readback(&bytes, 2)
            .expect("Fix: aligned hit readback bytes must decode directly");

        assert_eq!(hits.len(), 2);
        assert_eq!(hits[0].file_idx, 7);
        assert_eq!(hits[0].rule_idx, 3);
        assert_eq!(hits[1].match_offset, 100);
    }

    #[test]
    fn hit_readback_into_reuses_caller_capacity() {
        let mut bytes = Vec::new();
        for word in [7u32, 3, 2, 99, 8, 4, 1, 100] {
            bytes.extend_from_slice(&word.to_le_bytes());
        }
        let mut hits = Vec::with_capacity(8);
        let ptr = hits.as_ptr();

        decode_hits_from_readback_into(&bytes, 2, &mut hits)
            .expect("Fix: aligned hit readback bytes must decode into caller scratch");

        assert_eq!(hits.len(), 2);
        assert_eq!(hits.as_ptr(), ptr);
    }

    #[test]
    fn occupancy_proxy_caps_at_full_utilization() {
        assert_eq!(occupancy_proxy_bps(32, 1, 64), 5_000);
        assert_eq!(occupancy_proxy_bps(128, 1, 64), 10_000);
        assert_eq!(occupancy_proxy_bps(0, 0, 0), 0);
        assert_eq!(occupancy_proxy_bps(u32::MAX, 1, 1), 10_000);
    }

    #[test]
    fn dispatch_report_exposes_release_telemetry_counters() {
        let src = include_str!("dispatcher.rs");
        for field in [
            "bytes_uploaded",
            "bytes_read_back",
            "bytes_moved",
            "resident_allocations",
            "kernel_launches",
            "sync_points",
            "occupancy_proxy_bps",
            "frontier_density_bps",
            "queue_state_readback_bytes",
            "hit_readback_bytes",
            "estimated_peak_device_bytes",
            "device_memory_budget_bytes",
            "topology",
        ] {
            assert!(
                src.contains(field),
                "BatchDispatchReport telemetry must expose `{field}` for megakernel performance gates"
            );
        }
    }
}

#[cfg(test)]
mod scan_batch_segmentation_tests {
    use super::{
        wgpu_scan_batch_segmentation_evidence, WgpuScanBatchSegmentationError,
        WgpuScanBatchSegmentationRequest, WGPU_SCAN_BATCH_SEGMENTATION_SCHEMA_VERSION,
    };

    #[test]
    fn segmentation_evidence_records_command_copy_bind_group_counts_and_match_digest() {
        let evidence =
            wgpu_scan_batch_segmentation_evidence(WgpuScanBatchSegmentationRequest::new(
                10, 4, 2, 1, 10, 3, 0x1234, 0x1234,
            ))
            .expect("Fix: valid WGPU scan segmentation evidence should be accepted");

        assert_eq!(
            evidence.schema_version,
            WGPU_SCAN_BATCH_SEGMENTATION_SCHEMA_VERSION
        );
        assert_eq!(evidence.chunk_count, 10);
        assert_eq!(evidence.segment_count, 3);
        assert_eq!(evidence.command_encoder_count, 3);
        assert_eq!(evidence.bind_group_reuse_count, 2);
        assert_eq!(evidence.bind_group_create_count, 1);
        assert_eq!(evidence.copy_count, 13);
        assert_eq!(evidence.match_digest, 0x1234);
        assert!(evidence.match_parity);
        assert!(evidence.all_command_counts_recorded);
        assert!(evidence.is_complete());
    }

    #[test]
    fn segmentation_evidence_rejects_missing_bind_group_accounting() {
        let error = wgpu_scan_batch_segmentation_evidence(WgpuScanBatchSegmentationRequest::new(
            9, 4, 1, 1, 9, 3, 0x1234, 0x1234,
        ))
        .expect_err("Fix: bind group counts must account for every segment");

        assert!(matches!(
            error,
            WgpuScanBatchSegmentationError::BindGroupCountMismatch {
                command_encoder_count: 3,
                bind_group_reuse_count: 1,
                bind_group_create_count: 1
            }
        ));
    }

    #[test]
    fn segmentation_evidence_rejects_match_digest_drift() {
        let error = wgpu_scan_batch_segmentation_evidence(WgpuScanBatchSegmentationRequest::new(
            4, 4, 0, 1, 4, 1, 0xaaaa, 0xbbbb,
        ))
        .expect_err("Fix: segmented WGPU scan output must match the oracle digest");

        assert!(matches!(
            error,
            WgpuScanBatchSegmentationError::MatchDigestMismatch {
                expected_match_digest: 0xaaaa,
                actual_match_digest: 0xbbbb
            }
        ));
    }
}