asupersync 0.3.4

//! Cache-aware intrusive priority heap for scheduler hot paths.
//!
//! This module provides [`IntrusivePriorityHeap`], a binary max-heap that stores
//! scheduling metadata (priority, generation, heap position) directly in
//! [`TaskRecord`] fields rather than in separate heap-allocated entries.
//!
//! # Design (SoA Layout)
//!
//! Traditional `BinaryHeap<SchedulerEntry>` uses an Array-of-Structs (AoS) layout:
//! one contiguous Vec of `{task, priority, generation}` tuples. This works but
//! allocates per-entry and mixes TaskId lookup keys with scheduling metadata.
//!
//! The intrusive heap uses a Struct-of-Arrays (SoA) split:
//! - **Heap backbone**: `Vec<TaskId>` — compact, cache-friendly array of 8-byte indices
//! - **Per-task metadata**: `heap_index`, `sched_priority`, `sched_generation` stored
//!   inline in `TaskRecord` — accessed only during sift operations
//!
//! This provides:
//! - **Zero allocations** after initial Vec capacity is established
//! - **Better cache locality** for heap traversal (compact Vec<TaskId>)
//! - **O(1) removal** by task ID (via stored heap_index)
//! - **O(log n) push/pop** with fewer cache misses than AoS
//!
//! # Ordering
//!
//! Higher priority first (max-heap). Within equal priority, earlier generation
//! (lower number) wins for FIFO tie-breaking.
//!
//! # Integration
//!
//! ```text
//! ┌─────────────────────────────────┐
//! │  IntrusivePriorityHeap          │
//! │  ┌────────────────────────────┐ │
//! │  │  heap: Vec<TaskId>         │ │  ← compact backbone
//! │  │  [T3, T1, T7, T2, ...]    │ │
//! │  └────────────────────────────┘ │
//! │  next_generation: u64           │
//! └─────────────────────────────────┘
//!          │ sift_up / sift_down
//!          ▼
//! ┌──────────────────────────────────────┐
//! │  Arena<TaskRecord>                   │
//! │  ┌──────────────────────────────────┐│
//! │  │ T1: heap_index=1, priority=5    ││ ← metadata in-record
//! │  │ T2: heap_index=3, priority=3    ││
//! │  │ T3: heap_index=0, priority=7    ││
//! │  │ T7: heap_index=2, priority=5    ││
//! │  └──────────────────────────────────┘│
//! └──────────────────────────────────────┘
//! ```

use crate::record::task::TaskRecord;
use crate::types::TaskId;
use crate::util::Arena;

/// An intrusive binary max-heap for scheduling tasks by priority.
///
/// The heap backbone is a compact `Vec<TaskId>`. Per-task scheduling metadata
/// (priority, generation, heap index) is stored in `TaskRecord` fields, giving
/// a SoA (Struct-of-Arrays) layout that minimises allocations and improves
/// cache utilisation during sift operations.
///
/// # Invariants
///
/// - For every entry at position `i` in `self.heap`:
///   `arena[heap[i]].heap_index == Some(i as u32)`
/// - For every entry at position `i` with parent `p = (i-1)/2`:
///   `priority(heap[p]) >= priority(heap[i])` (max-heap)
/// - Tasks not in this heap have `heap_index == None`
///
/// # Complexity
///
/// | Operation | Time       | Allocations |
/// |-----------|------------|-------------|
/// | push      | O(log n)   | 0 (amortised) |
/// | pop       | O(log n)   | 0           |
/// | remove    | O(log n)   | 0           |
/// | peek      | O(1)       | 0           |
/// | contains  | O(1)       | 0           |
#[derive(Debug, Default)]
pub struct IntrusivePriorityHeap {
    /// Compact array of TaskIds forming the heap structure.
    heap: Vec<TaskId>,
    /// Monotonic counter for FIFO tie-breaking within equal priorities.
    ///
    /// The counter wraps explicitly and is reset when the heap drains, so an
    /// empty heap always starts a fresh generation epoch.
    next_generation: u64,
}

impl IntrusivePriorityHeap {
    /// Creates a new empty intrusive priority heap.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Creates a new heap with pre-allocated capacity.
    #[must_use]
    pub fn with_capacity(capacity: usize) -> Self {
        Self {
            heap: Vec::with_capacity(capacity),
            next_generation: 0,
        }
    }

    /// Returns the number of tasks in the heap.
    #[must_use]
    #[inline]
    pub fn len(&self) -> usize {
        self.heap.len()
    }

    /// Returns true if the heap is empty.
    #[must_use]
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.heap.is_empty()
    }

    /// Returns the highest-priority task without removing it.
    #[must_use]
    #[inline]
    pub fn peek(&self) -> Option<TaskId> {
        self.heap.first().copied()
    }

    /// Returns true if the given task is in this heap.
    ///
    /// O(1) via the stored `heap_index` field.
    #[must_use]
    pub fn contains(&self, task: TaskId, arena: &Arena<TaskRecord>) -> bool {
        arena.get(task.arena_index()).is_some_and(|record| {
            let Some(pos) = record.heap_index else {
                return false;
            };
            let pos = pos as usize;
            pos < self.heap.len() && self.heap[pos] == task
        })
    }

    /// Pushes a task into the heap with the given priority.
    ///
    /// If the task is already in the heap, this is a no-op.
    ///
    /// # Complexity
    ///
    /// O(log n) time, O(0) allocations (amortised, after Vec warmup).
    #[inline]
    pub fn push(&mut self, task: TaskId, priority: u8, arena: &mut Arena<TaskRecord>) {
        let Some(record) = arena.get_mut(task.arena_index()) else {
            return;
        };

        // Skip if already in heap
        if record.heap_index.is_some() {
            return;
        }

        let generation = self.next_generation;
        self.next_generation = self.next_generation.wrapping_add(1);

        record.sched_priority = priority;
        record.sched_generation = generation;

        let pos = self.heap.len();
        record.heap_index = Some(pos as u32);
        self.heap.push(task);

        self.sift_up(pos, arena);
    }

    /// Removes and returns the highest-priority task.
    ///
    /// # Complexity
    ///
    /// O(log n) time, O(0) allocations.
    #[inline]
    #[must_use]
    pub fn pop(&mut self, arena: &mut Arena<TaskRecord>) -> Option<TaskId> {
        if self.heap.is_empty() {
            return None;
        }

        let task = self.heap[0];
        self.remove_at(0, arena);
        Some(task)
    }

    /// Removes a specific task from the heap.
    ///
    /// Returns `true` if the task was found and removed.
    ///
    /// # Complexity
    ///
    /// O(log n) time, O(0) allocations.
    pub fn remove(&mut self, task: TaskId, arena: &mut Arena<TaskRecord>) -> bool {
        let Some(record) = arena.get(task.arena_index()) else {
            return false;
        };

        let Some(pos) = record.heap_index else {
            return false;
        };

        let pos = pos as usize;
        // Defensively validate slot ownership before removing. A stale or
        // corrupted heap_index must not remove arbitrary tasks or panic.
        if pos >= self.heap.len() || self.heap[pos] != task {
            if let Some(record) = arena.get_mut(task.arena_index()) {
                record.heap_index = None;
                record.sched_priority = 0;
                record.sched_generation = 0;
            }
            return false;
        }

        self.remove_at(pos, arena);
        true
    }

    /// Removes the element at position `pos` from the heap.
    fn remove_at(&mut self, pos: usize, arena: &mut Arena<TaskRecord>) {
        let last = self.heap.len() - 1;

        // Clear the removed task's heap index
        if let Some(record) = arena.get_mut(self.heap[pos].arena_index()) {
            record.heap_index = None;
            record.sched_priority = 0;
            record.sched_generation = 0;
        }

        if pos == last {
            self.heap.pop();
            self.reset_generation_if_empty();
            return;
        }

        // Swap with last element
        self.heap.swap(pos, last);
        self.heap.pop();

        // Update the swapped element's index
        if let Some(record) = arena.get_mut(self.heap[pos].arena_index()) {
            record.heap_index = Some(pos as u32);
        }

        // Restore heap property
        // Try sifting up first; if position didn't change, sift down
        let new_pos = self.sift_up(pos, arena);
        if new_pos == pos {
            self.sift_down(pos, arena);
        }
    }

    /// Sifts the element at `pos` up towards the root.
    /// Returns the final position.
    fn sift_up(&mut self, mut pos: usize, arena: &mut Arena<TaskRecord>) -> usize {
        while pos > 0 {
            let parent = (pos - 1) / 2;
            if self.higher_priority(pos, parent, arena) {
                self.swap_positions(pos, parent, arena);
                pos = parent;
            } else {
                break;
            }
        }
        pos
    }

    /// Sifts the element at `pos` down towards the leaves.
    fn sift_down(&mut self, mut pos: usize, arena: &mut Arena<TaskRecord>) {
        let len = self.heap.len();
        loop {
            let left = 2 * pos + 1;
            let right = 2 * pos + 2;
            let mut largest = pos;

            if left < len && self.higher_priority(left, largest, arena) {
                largest = left;
            }
            if right < len && self.higher_priority(right, largest, arena) {
                largest = right;
            }

            if largest == pos {
                break;
            }

            self.swap_positions(pos, largest, arena);
            pos = largest;
        }
    }

    /// Returns `true` if the task at position `a` has strictly higher scheduling
    /// priority than the task at position `b`.
    ///
    /// Higher priority value wins. For equal priorities, lower generation (FIFO) wins.
    fn higher_priority(&self, a: usize, b: usize, arena: &Arena<TaskRecord>) -> bool {
        let task_a = self.heap[a];
        let task_b = self.heap[b];

        let (prio_a, gen_a) = arena
            .get(task_a.arena_index())
            .map_or((0, u64::MAX), |r| (r.sched_priority, r.sched_generation));

        let (prio_b, gen_b) = arena
            .get(task_b.arena_index())
            .map_or((0, u64::MAX), |r| (r.sched_priority, r.sched_generation));

        match prio_a.cmp(&prio_b) {
            std::cmp::Ordering::Greater => true,
            std::cmp::Ordering::Less => false,
            std::cmp::Ordering::Equal => gen_b.wrapping_sub(gen_a).cast_signed() > 0, // Earlier generation = higher priority (FIFO)
        }
    }

    /// Swaps two positions in the heap and updates their stored indices.
    fn swap_positions(&mut self, a: usize, b: usize, arena: &mut Arena<TaskRecord>) {
        self.heap.swap(a, b);

        if let Some(record) = arena.get_mut(self.heap[a].arena_index()) {
            record.heap_index = Some(a as u32);
        }
        if let Some(record) = arena.get_mut(self.heap[b].arena_index()) {
            record.heap_index = Some(b as u32);
        }
    }

    /// Clears all entries from the heap, resetting all task heap indices.
    pub fn clear(&mut self, arena: &mut Arena<TaskRecord>) {
        for &task in &self.heap {
            if let Some(record) = arena.get_mut(task.arena_index()) {
                record.heap_index = None;
                record.sched_priority = 0;
                record.sched_generation = 0;
            }
        }
        self.heap.clear();
        self.reset_generation_if_empty();
    }

    #[inline]
    fn reset_generation_if_empty(&mut self) {
        if self.heap.is_empty() {
            self.next_generation = 0;
        }
    }
}

#[cfg(any(test, feature = "test-internals"))]
impl IntrusivePriorityHeap {
    #[doc(hidden)]
    pub fn decrease_key_for_test(
        &mut self,
        task: TaskId,
        new_priority: u8,
        arena: &mut Arena<TaskRecord>,
    ) -> bool {
        let Some(record) = arena.get(task.arena_index()) else {
            return false;
        };

        let Some(pos) = record.heap_index else {
            return false;
        };

        let pos = pos as usize;
        if pos >= self.heap.len() || self.heap[pos] != task {
            return false;
        }

        let current_priority = record.sched_priority;
        if new_priority >= current_priority {
            return false;
        }

        if let Some(record) = arena.get_mut(task.arena_index()) {
            record.sched_priority = new_priority;
        }
        self.sift_down(pos, arena);
        true
    }

    #[doc(hidden)]
    #[must_use]
    pub fn verify_invariants_for_test(&self, arena: &Arena<TaskRecord>) -> bool {
        for (idx, &task) in self.heap.iter().enumerate() {
            let Some(record) = arena.get(task.arena_index()) else {
                return false;
            };
            if record.heap_index != Some(idx as u32) {
                return false;
            }

            if idx > 0 {
                let parent = (idx - 1) / 2;
                if self.higher_priority(idx, parent, arena) {
                    return false;
                }
            }
        }
        true
    }
}

#[cfg(test)]
mod tests {
    #![allow(
        clippy::pedantic,
        clippy::nursery,
        clippy::expect_fun_call,
        clippy::map_unwrap_or,
        clippy::cast_possible_wrap,
        clippy::future_not_send
    )]
    use super::*;
    use crate::types::{Budget, RegionId};
    use crate::util::ArenaIndex;

    fn region() -> RegionId {
        RegionId::from_arena(ArenaIndex::new(0, 0))
    }

    fn task(n: u32) -> TaskId {
        TaskId::from_arena(ArenaIndex::new(n, 0))
    }

    fn setup_arena(count: u32) -> Arena<TaskRecord> {
        let mut arena = Arena::new();
        for i in 0..count {
            let id = task(i);
            let record = TaskRecord::new(id, region(), Budget::INFINITE);
            let idx = arena.insert(record);
            assert_eq!(idx.index(), i);
        }
        arena
    }

    fn pop_all(heap: &mut IntrusivePriorityHeap, arena: &mut Arena<TaskRecord>) -> Vec<TaskId> {
        let mut popped = Vec::new();
        while let Some(task) = heap.pop(arena) {
            popped.push(task);
        }
        popped
    }

    #[test]
    fn empty_heap() {
        let heap = IntrusivePriorityHeap::new();
        assert!(heap.is_empty());
        assert_eq!(heap.len(), 0);
        assert!(heap.peek().is_none());
    }

    #[test]
    fn push_pop_single() {
        let mut arena = setup_arena(1);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 5, &mut arena);
        assert_eq!(heap.len(), 1);
        assert_eq!(heap.peek(), Some(task(0)));

        let popped = heap.pop(&mut arena);
        assert_eq!(popped, Some(task(0)));
        assert!(heap.is_empty());

        // Verify heap_index is cleared
        let record = arena.get(task(0).arena_index()).unwrap();
        assert!(record.heap_index.is_none());
    }

    #[test]
    fn priority_ordering() {
        let mut arena = setup_arena(5);
        let mut heap = IntrusivePriorityHeap::new();

        // Push with different priorities
        heap.push(task(0), 1, &mut arena); // lowest
        heap.push(task(1), 5, &mut arena); // highest
        heap.push(task(2), 3, &mut arena); // middle
        heap.push(task(3), 5, &mut arena); // equal to task 1
        heap.push(task(4), 2, &mut arena);

        // Pop should return highest priority first
        let first = heap.pop(&mut arena).unwrap();
        assert_eq!(first, task(1), "highest priority, earliest generation");

        let second = heap.pop(&mut arena).unwrap();
        assert_eq!(second, task(3), "same priority as task 1, later generation");

        let third = heap.pop(&mut arena).unwrap();
        assert_eq!(third, task(2), "priority 3");

        let fourth = heap.pop(&mut arena).unwrap();
        assert_eq!(fourth, task(4), "priority 2");

        let fifth = heap.pop(&mut arena).unwrap();
        assert_eq!(fifth, task(0), "priority 1 (lowest)");

        assert!(heap.is_empty());
    }

    #[test]
    fn fifo_within_same_priority() {
        let mut arena = setup_arena(5);
        let mut heap = IntrusivePriorityHeap::new();

        // All same priority
        for i in 0..5 {
            heap.push(task(i), 5, &mut arena);
        }

        // Should pop in insertion order (FIFO)
        for i in 0..5 {
            let popped = heap.pop(&mut arena).unwrap();
            assert_eq!(popped, task(i), "FIFO: expected task {i}");
        }
    }

    #[test]
    fn remove_by_task_id() {
        let mut arena = setup_arena(5);
        let mut heap = IntrusivePriorityHeap::new();

        for i in 0..5 {
            heap.push(task(i), u8::try_from(i).unwrap(), &mut arena);
        }
        assert_eq!(heap.len(), 5);

        // Remove task from middle
        let removed = heap.remove(task(2), &mut arena);
        assert!(removed);
        assert_eq!(heap.len(), 4);

        // Verify removed task's heap_index is cleared
        let record = arena.get(task(2).arena_index()).unwrap();
        assert!(record.heap_index.is_none());

        // Pop remaining in priority order: 4, 3, 1, 0
        assert_eq!(heap.pop(&mut arena), Some(task(4)));
        assert_eq!(heap.pop(&mut arena), Some(task(3)));
        assert_eq!(heap.pop(&mut arena), Some(task(1)));
        assert_eq!(heap.pop(&mut arena), Some(task(0)));
    }

    #[test]
    fn remove_not_in_heap() {
        let mut arena = setup_arena(2);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 5, &mut arena);
        let removed = heap.remove(task(1), &mut arena);
        assert!(!removed);
        assert_eq!(heap.len(), 1);
    }

    #[test]
    fn contains_check() {
        let mut arena = setup_arena(3);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 5, &mut arena);
        heap.push(task(1), 3, &mut arena);

        assert!(heap.contains(task(0), &arena));
        assert!(heap.contains(task(1), &arena));
        assert!(!heap.contains(task(2), &arena));

        let _ = heap.pop(&mut arena);
        assert!(!heap.contains(task(0), &arena)); // Was popped
        assert!(heap.contains(task(1), &arena));
    }

    #[test]
    fn no_duplicate_push() {
        let mut arena = setup_arena(1);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 5, &mut arena);
        heap.push(task(0), 10, &mut arena); // duplicate, should be no-op
        assert_eq!(heap.len(), 1);
    }

    #[test]
    fn clear_resets_all() {
        let mut arena = setup_arena(5);
        let mut heap = IntrusivePriorityHeap::new();

        for i in 0..5 {
            heap.push(task(i), u8::try_from(i).unwrap(), &mut arena);
        }
        assert_eq!(heap.len(), 5);

        heap.clear(&mut arena);
        assert!(heap.is_empty());

        // Verify all heap indices cleared
        for i in 0..5 {
            let record = arena.get(task(i).arena_index()).unwrap();
            assert!(record.heap_index.is_none());
        }
    }

    #[test]
    fn high_volume() {
        let count = 1000u32;
        let mut arena = setup_arena(count);
        let mut heap = IntrusivePriorityHeap::with_capacity(count as usize);

        // Push all with varying priorities
        for i in 0..count {
            let priority = (i % 10) as u8;
            heap.push(task(i), priority, &mut arena);
        }
        assert_eq!(heap.len(), count as usize);

        // Pop all and count
        let mut popped_count = 0u32;
        while heap.pop(&mut arena).is_some() {
            popped_count += 1;
        }
        assert_eq!(popped_count, count);
        assert!(heap.is_empty());
    }

    #[test]
    fn interleaved_push_pop() {
        let mut arena = setup_arena(10);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 3, &mut arena);
        heap.push(task(1), 7, &mut arena);
        assert_eq!(heap.pop(&mut arena), Some(task(1))); // priority 7

        heap.push(task(2), 5, &mut arena);
        heap.push(task(3), 9, &mut arena);
        assert_eq!(heap.pop(&mut arena), Some(task(3))); // priority 9
        assert_eq!(heap.pop(&mut arena), Some(task(2))); // priority 5
        assert_eq!(heap.pop(&mut arena), Some(task(0))); // priority 3
        assert!(heap.is_empty());
    }

    #[test]
    fn metamorphic_priority_permutation_preserves_descending_pop_order() {
        let fixtures = [(0, 10), (1, 40), (2, 20), (3, 60), (4, 30), (5, 50)];
        let canonical_order = [0, 1, 2, 3, 4, 5];
        let permuted_order = [2, 5, 1, 4, 0, 3];

        let mut canonical_arena = setup_arena(fixtures.len() as u32);
        let mut canonical_heap = IntrusivePriorityHeap::new();
        for index in canonical_order {
            let (task_id, priority) = fixtures[index];
            canonical_heap.push(task(task_id), priority, &mut canonical_arena);
        }

        let mut permuted_arena = setup_arena(fixtures.len() as u32);
        let mut permuted_heap = IntrusivePriorityHeap::new();
        for index in permuted_order {
            let (task_id, priority) = fixtures[index];
            permuted_heap.push(task(task_id), priority, &mut permuted_arena);
        }

        let canonical_popped = pop_all(&mut canonical_heap, &mut canonical_arena);
        let permuted_popped = pop_all(&mut permuted_heap, &mut permuted_arena);
        let expected = vec![task(3), task(5), task(1), task(4), task(2), task(0)];

        assert_eq!(
            canonical_popped, expected,
            "canonical insertion should pop in descending priority order"
        );
        assert_eq!(
            permuted_popped, expected,
            "permuting distinct-priority insertions must preserve pop order"
        );
    }

    #[test]
    fn metamorphic_low_priority_noise_preserves_fifo_within_urgent_band() {
        let urgent_band = [task(0), task(1), task(2)];

        let mut baseline_arena = setup_arena(6);
        let mut baseline_heap = IntrusivePriorityHeap::new();
        for urgent in urgent_band {
            baseline_heap.push(urgent, 9, &mut baseline_arena);
        }
        let baseline_prefix: Vec<_> = pop_all(&mut baseline_heap, &mut baseline_arena)
            .into_iter()
            .take(urgent_band.len())
            .collect();

        let mut noisy_arena = setup_arena(6);
        let mut noisy_heap = IntrusivePriorityHeap::new();
        noisy_heap.push(task(0), 9, &mut noisy_arena);
        noisy_heap.push(task(3), 1, &mut noisy_arena);
        noisy_heap.push(task(1), 9, &mut noisy_arena);
        noisy_heap.push(task(4), 2, &mut noisy_arena);
        noisy_heap.push(task(2), 9, &mut noisy_arena);
        noisy_heap.push(task(5), 0, &mut noisy_arena);

        let noisy_popped = pop_all(&mut noisy_heap, &mut noisy_arena);
        let noisy_prefix: Vec<_> = noisy_popped
            .iter()
            .copied()
            .take(urgent_band.len())
            .collect();

        assert_eq!(
            baseline_prefix, urgent_band,
            "equal-priority urgent tasks should pop FIFO without background noise"
        );
        assert_eq!(
            noisy_prefix, urgent_band,
            "lower-priority noise must not perturb FIFO ordering within the urgent band"
        );
        assert!(
            noisy_popped
                .iter()
                .copied()
                .skip(urgent_band.len())
                .eq([task(4), task(3), task(5)]),
            "noise should still drain by descending priority after the urgent band"
        );
    }

    #[test]
    fn reuse_after_pop() {
        let mut arena = setup_arena(1);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 5, &mut arena);
        let _ = heap.pop(&mut arena);

        // Re-push the same task
        heap.push(task(0), 8, &mut arena);
        assert_eq!(heap.len(), 1);
        assert_eq!(heap.peek(), Some(task(0)));

        let record = arena.get(task(0).arena_index()).unwrap();
        assert_eq!(record.sched_priority, 8);
    }

    #[test]
    fn remove_head() {
        let mut arena = setup_arena(3);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 1, &mut arena);
        heap.push(task(1), 9, &mut arena);
        heap.push(task(2), 5, &mut arena);

        // Remove the head (task 1, priority 9)
        let removed = heap.remove(task(1), &mut arena);
        assert!(removed);
        assert_eq!(heap.len(), 2);

        // Next pop should be task 2 (priority 5)
        assert_eq!(heap.pop(&mut arena), Some(task(2)));
        assert_eq!(heap.pop(&mut arena), Some(task(0)));
    }

    #[test]
    fn remove_tail() {
        let mut arena = setup_arena(3);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 9, &mut arena);
        heap.push(task(1), 5, &mut arena);
        heap.push(task(2), 1, &mut arena);

        // Remove lowest priority (task 2)
        let removed = heap.remove(task(2), &mut arena);
        assert!(removed);
        assert_eq!(heap.len(), 2);

        assert_eq!(heap.pop(&mut arena), Some(task(0)));
        assert_eq!(heap.pop(&mut arena), Some(task(1)));
    }

    #[test]
    fn contains_rejects_stale_heap_index() {
        let mut arena = setup_arena(2);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 9, &mut arena);

        // Corrupt task(1) metadata to point at task(0)'s slot.
        if let Some(record) = arena.get_mut(task(1).arena_index()) {
            record.heap_index = Some(0);
            record.sched_priority = 9;
            record.sched_generation = 0;
        }

        assert!(heap.contains(task(0), &arena));
        assert!(
            !heap.contains(task(1), &arena),
            "stale index must not be treated as membership"
        );
    }

    #[test]
    fn remove_with_stale_heap_index_is_safe_and_non_destructive() {
        let mut arena = setup_arena(2);
        let mut heap = IntrusivePriorityHeap::new();

        heap.push(task(0), 9, &mut arena);

        // Corrupt task(1) metadata to point at task(0)'s slot.
        if let Some(record) = arena.get_mut(task(1).arena_index()) {
            record.heap_index = Some(0);
            record.sched_priority = 9;
            record.sched_generation = 0;
        }

        assert!(
            !heap.remove(task(1), &mut arena),
            "stale index must not remove arbitrary task"
        );
        assert_eq!(heap.len(), 1, "heap content must be preserved");
        assert_eq!(heap.peek(), Some(task(0)));

        // The stale metadata is healed.
        let record = arena.get(task(1).arena_index()).unwrap();
        assert!(record.heap_index.is_none());
        assert_eq!(record.sched_priority, 0);
        assert_eq!(record.sched_generation, 0);
    }

    #[test]
    fn generation_wrap_preserves_fifo_within_same_priority() {
        let mut arena = setup_arena(2);
        let mut heap = IntrusivePriorityHeap::new();
        heap.next_generation = u64::MAX;

        heap.push(task(0), 9, &mut arena);
        heap.push(task(1), 9, &mut arena);

        assert_eq!(heap.pop(&mut arena), Some(task(0)));
        assert_eq!(heap.pop(&mut arena), Some(task(1)));
        assert_eq!(heap.next_generation, 0);
    }

    #[test]
    fn generation_epoch_resets_when_heap_becomes_empty() {
        let mut arena = setup_arena(3);
        let mut heap = IntrusivePriorityHeap::new();
        heap.next_generation = 41;

        heap.push(task(0), 5, &mut arena);
        assert_eq!(heap.next_generation, 42);
        assert_eq!(heap.pop(&mut arena), Some(task(0)));
        assert_eq!(heap.next_generation, 0);

        heap.push(task(1), 7, &mut arena);
        heap.push(task(2), 3, &mut arena);
        assert_eq!(heap.next_generation, 2);
        heap.clear(&mut arena);
        assert_eq!(heap.next_generation, 0);
    }

    /// Comprehensive metamorphic testing module for intrusive heap.
    mod metamorphic {
        use super::*;
        use proptest::prelude::*;

        fn extract_priorities(tasks: &[TaskId], arena: &Arena<TaskRecord>) -> Vec<u8> {
            tasks
                .iter()
                .map(|&task_id| {
                    arena
                        .get(task_id.arena_index())
                        .map_or(0, |record| record.sched_priority)
                })
                .collect()
        }

        fn extract_fixture_priorities(tasks: &[TaskId], fixtures: &[(u32, u8)]) -> Vec<u8> {
            tasks
                .iter()
                .map(|task_id| {
                    let index = task_id.arena_index().index();
                    fixtures
                        .iter()
                        .find_map(|&(fixture_id, priority)| {
                            (fixture_id == index).then_some(priority)
                        })
                        .unwrap_or(0)
                })
                .collect()
        }

        /// MR1: Priority Offset Additivity
        /// f(tasks + uniform_offset) should preserve relative ordering
        /// Category: Additive (f(x + c) = permute(f(x)))
        #[test]
        fn mr_priority_offset_preserves_relative_ordering() {
            let fixtures = [(0, 10), (1, 30), (2, 20), (3, 50), (4, 40)];
            let offset = 100u8;

            // Build baseline heap
            let mut baseline_arena = setup_arena(fixtures.len() as u32);
            let mut baseline_heap = IntrusivePriorityHeap::new();
            for &(task_id, priority) in &fixtures {
                baseline_heap.push(task(task_id), priority, &mut baseline_arena);
            }
            let baseline_popped = pop_all(&mut baseline_heap, &mut baseline_arena);
            let baseline_priorities = extract_fixture_priorities(&baseline_popped, &fixtures);

            // Build offset heap (all priorities shifted by constant)
            let mut offset_arena = setup_arena(fixtures.len() as u32);
            let mut offset_heap = IntrusivePriorityHeap::new();
            let mut offset_fixtures = Vec::with_capacity(fixtures.len());
            for &(task_id, priority) in &fixtures {
                let new_priority = priority.saturating_add(offset);
                offset_fixtures.push((task_id, new_priority));
                offset_heap.push(task(task_id), new_priority, &mut offset_arena);
            }
            let offset_popped = pop_all(&mut offset_heap, &mut offset_arena);

            assert_eq!(
                baseline_popped, offset_popped,
                "uniform priority offset must preserve task ordering"
            );

            // Verify the priorities were actually shifted
            let offset_priorities = extract_fixture_priorities(&offset_popped, &offset_fixtures);
            for (baseline_prio, offset_prio) in baseline_priorities.iter().zip(&offset_priorities) {
                assert_eq!(
                    *offset_prio,
                    baseline_prio.saturating_add(offset),
                    "each priority should be shifted by exactly {offset}"
                );
            }
        }

        /// MR2: Push-Pop Roundtrip Invertibility
        /// push(x); pop() should return x for single-element heaps
        /// Category: Invertive (f(T(T(x))) = f(x))
        #[test]
        fn mr_push_pop_roundtrip_invertibility() {
            proptest!(|(task_id in any::<u32>(), priority in any::<u8>())| {
                let task_id = task_id % 100; // Bound to reasonable range
                let mut arena = setup_arena(100);
                let mut heap = IntrusivePriorityHeap::new();

                // Push then immediately pop
                heap.push(task(task_id), priority, &mut arena);
                let popped = heap.pop(&mut arena);

                prop_assert_eq!(popped, Some(task(task_id)),
                    "single push-pop must be invertible");
                prop_assert!(heap.is_empty(),
                    "heap must be empty after pop in single-element case");
            });
        }

        /// MR3: Subset Monotonicity
        /// Removing tasks should never increase the priority of any remaining pop
        /// Category: Inclusive (subset input → subset-compatible output)
        #[test]
        fn mr_subset_monotonicity_preserves_priority_bounds() {
            let all_fixtures = [
                (0, 10),
                (1, 60),
                (2, 20),
                (3, 80),
                (4, 30),
                (5, 70),
                (6, 40),
                (7, 50),
            ];
            let subset_fixtures = [
                (1, 60),
                (3, 80),
                (5, 70),
                (7, 50), // Remove lower-priority items
            ];

            // Build full heap
            let mut full_arena = setup_arena(all_fixtures.len() as u32);
            let mut full_heap = IntrusivePriorityHeap::new();
            for &(task_id, priority) in &all_fixtures {
                full_heap.push(task(task_id), priority, &mut full_arena);
            }
            let full_popped = pop_all(&mut full_heap, &mut full_arena);
            let full_priorities = extract_priorities(&full_popped, &full_arena);

            // Build subset heap (removing some items)
            let mut subset_arena = setup_arena(all_fixtures.len() as u32);
            let mut subset_heap = IntrusivePriorityHeap::new();
            for &(task_id, priority) in &subset_fixtures {
                subset_heap.push(task(task_id), priority, &mut subset_arena);
            }
            let subset_popped = pop_all(&mut subset_heap, &mut subset_arena);
            let subset_priorities = extract_priorities(&subset_popped, &subset_arena);

            // Every priority in subset should exist in full (monotonicity)
            for &subset_prio in &subset_priorities {
                assert!(
                    full_priorities.contains(&subset_prio),
                    "subset priority {subset_prio} must exist in full heap"
                );
            }

            // Subset priorities should be non-increasing (heap property preserved)
            for window in subset_priorities.windows(2) {
                assert!(
                    window[0] >= window[1],
                    "subset must maintain descending priority order: {} >= {}",
                    window[0],
                    window[1]
                );
            }

            // Expected subset order: task(3)=80, task(5)=70, task(1)=60, task(7)=50
            assert_eq!(subset_popped, vec![task(3), task(5), task(1), task(7)]);
        }

        /// MR4: Clear-Rebuild Equivalence
        /// clear() + rebuild should equal building from scratch
        /// Category: Invertive (different paths to same state)
        #[test]
        fn mr_clear_rebuild_equivalence() {
            let fixtures = [(0, 25), (1, 75), (2, 50), (3, 100), (4, 10)];

            // Build fresh heap
            let mut fresh_arena = setup_arena(fixtures.len() as u32);
            let mut fresh_heap = IntrusivePriorityHeap::new();
            for &(task_id, priority) in &fixtures {
                fresh_heap.push(task(task_id), priority, &mut fresh_arena);
            }
            let fresh_popped = pop_all(&mut fresh_heap, &mut fresh_arena);

            // Build heap, then clear and rebuild
            let mut rebuild_arena = setup_arena(fixtures.len() as u32);
            let mut rebuild_heap = IntrusivePriorityHeap::new();

            // Initial build with different items
            rebuild_heap.push(task(10), 1, &mut rebuild_arena);
            rebuild_heap.push(task(11), 2, &mut rebuild_arena);

            // Clear and rebuild with target items
            rebuild_heap.clear(&mut rebuild_arena);
            for &(task_id, priority) in &fixtures {
                rebuild_heap.push(task(task_id), priority, &mut rebuild_arena);
            }
            let rebuild_popped = pop_all(&mut rebuild_heap, &mut rebuild_arena);

            assert_eq!(
                fresh_popped, rebuild_popped,
                "clear-rebuild must be equivalent to fresh construction"
            );
        }

        /// MR5: FIFO Preservation Under Priority Band Isolation
        /// Adding different-priority items shouldn't affect FIFO within a priority band
        /// Category: Equivalence (transformation preserves core property)
        #[test]
        fn mr_fifo_preservation_under_priority_band_isolation() {
            let same_priority = 50u8;
            let fifo_tasks = [task(10), task(11), task(12), task(13)];
            let noise_items = [(20, 30), (21, 70), (22, 10), (23, 90)];

            // Baseline: just the FIFO band
            let mut baseline_arena = setup_arena(30);
            let mut baseline_heap = IntrusivePriorityHeap::new();
            for &task_id in &fifo_tasks {
                baseline_heap.push(task_id, same_priority, &mut baseline_arena);
            }

            // Extract just the same-priority items in pop order
            let mut baseline_fifo = Vec::new();
            while let Some(popped) = baseline_heap.pop(&mut baseline_arena) {
                if fifo_tasks.contains(&popped) {
                    baseline_fifo.push(popped);
                }
            }

            // Noisy version: interleave different priorities
            let mut noisy_arena = setup_arena(30);
            let mut noisy_heap = IntrusivePriorityHeap::new();

            noisy_heap.push(fifo_tasks[0], same_priority, &mut noisy_arena);
            noisy_heap.push(task(noise_items[0].0), noise_items[0].1, &mut noisy_arena);
            noisy_heap.push(fifo_tasks[1], same_priority, &mut noisy_arena);
            noisy_heap.push(task(noise_items[1].0), noise_items[1].1, &mut noisy_arena);
            noisy_heap.push(fifo_tasks[2], same_priority, &mut noisy_arena);
            noisy_heap.push(task(noise_items[2].0), noise_items[2].1, &mut noisy_arena);
            noisy_heap.push(fifo_tasks[3], same_priority, &mut noisy_arena);
            noisy_heap.push(task(noise_items[3].0), noise_items[3].1, &mut noisy_arena);

            // Extract same-priority items from noisy heap
            let mut noisy_fifo = Vec::new();
            while let Some(popped) = noisy_heap.pop(&mut noisy_arena) {
                if fifo_tasks.contains(&popped) {
                    noisy_fifo.push(popped);
                }
            }

            assert_eq!(
                baseline_fifo, noisy_fifo,
                "different-priority noise must not disrupt FIFO ordering within priority band"
            );
            assert_eq!(
                baseline_fifo,
                fifo_tasks.to_vec(),
                "same-priority items must pop in FIFO order"
            );
        }

        /// MR6: Removal Non-Interference
        /// Removing items outside priority band shouldn't affect ordering within band
        /// Category: Exclusive (disjoint operations preserve properties)
        #[test]
        fn mr_removal_non_interference_with_priority_bands() {
            let target_band_priority = 60u8;
            let target_tasks = [task(5), task(6), task(7)];
            let removal_candidates = [(task(1), 20), (task(2), 40), (task(3), 80), (task(4), 100)];

            // Build baseline with only target band
            let mut baseline_arena = setup_arena(20);
            let mut baseline_heap = IntrusivePriorityHeap::new();
            for &task_id in &target_tasks {
                baseline_heap.push(task_id, target_band_priority, &mut baseline_arena);
            }
            let baseline_popped = pop_all(&mut baseline_heap, &mut baseline_arena);

            // Build full heap with target + removal candidates
            let mut full_arena = setup_arena(20);
            let mut full_heap = IntrusivePriorityHeap::new();
            for &task_id in &target_tasks {
                full_heap.push(task_id, target_band_priority, &mut full_arena);
            }
            for &(task_id, priority) in &removal_candidates {
                full_heap.push(task_id, priority, &mut full_arena);
            }

            // Remove the candidates (different priorities)
            for &(task_id, _) in &removal_candidates {
                let removed = full_heap.remove(task_id, &mut full_arena);
                assert!(removed, "removal candidate {task_id:?} should be removable");
            }

            let full_popped = pop_all(&mut full_heap, &mut full_arena);

            assert_eq!(
                baseline_popped, full_popped,
                "removing different-priority items must not affect target band ordering"
            );
        }

        /// MR7: Priority Scaling Linearity
        /// Scaling all priorities by constant factor preserves relative ordering
        /// Category: Multiplicative (f(k·x) = h(k)·f(x))
        #[test]
        fn mr_priority_scaling_preserves_relative_ordering() {
            let base_fixtures = [(0, 2), (1, 6), (2, 4), (3, 10), (4, 8)];
            let scale_factor = 10u8;

            // Build baseline heap
            let mut baseline_arena = setup_arena(base_fixtures.len() as u32);
            let mut baseline_heap = IntrusivePriorityHeap::new();
            for &(task_id, priority) in &base_fixtures {
                baseline_heap.push(task(task_id), priority, &mut baseline_arena);
            }
            let baseline_popped = pop_all(&mut baseline_heap, &mut baseline_arena);

            // Build scaled heap (all priorities multiplied by factor)
            let mut scaled_arena = setup_arena(base_fixtures.len() as u32);
            let mut scaled_heap = IntrusivePriorityHeap::new();
            for &(task_id, priority) in &base_fixtures {
                let scaled_priority = priority.saturating_mul(scale_factor);
                scaled_heap.push(task(task_id), scaled_priority, &mut scaled_arena);
            }
            let scaled_popped = pop_all(&mut scaled_heap, &mut scaled_arena);

            assert_eq!(
                baseline_popped, scaled_popped,
                "priority scaling must preserve task ordering"
            );

            // Verify scaling actually happened
            let baseline_priorities = extract_priorities(&baseline_popped, &baseline_arena);
            let scaled_priorities = extract_priorities(&scaled_popped, &scaled_arena);
            for (baseline_prio, scaled_prio) in baseline_priorities.iter().zip(&scaled_priorities) {
                assert_eq!(
                    *scaled_prio,
                    baseline_prio.saturating_mul(scale_factor),
                    "each priority should be scaled by factor {scale_factor}"
                );
            }
        }

        /// Property-Based MR: Heap Property Preservation Under Random Operations
        proptest! {
            #[test]
            fn property_heap_invariant_preserved_under_random_operations(
                operations in prop::collection::vec(
                    prop::strategy::Union::new([
                        (0..50u32, 0..255u8).prop_map(|(id, prio)| ("push", id, prio)).boxed(),
                        (0..50u32).prop_map(|id| ("pop", id, 0)).boxed(),
                        (0..50u32).prop_map(|id| ("remove", id, 0)).boxed(),
                    ]), 1..20
                )
            ) {
                let mut arena = setup_arena(50);
                let mut heap = IntrusivePriorityHeap::new();

                for (op, task_id, priority) in operations {
                    match op {
                        "push" => heap.push(task(task_id), priority, &mut arena),
                        "pop" => {
                            let _ = heap.pop(&mut arena);
                        }
                        "remove" => {
                            heap.remove(task(task_id), &mut arena);
                        }
                        _ => unreachable!(),
                    }

                    // Invariant: heap property must hold after every operation
                    prop_assert!(heap.verify_invariants_for_test(&arena),
                        "heap invariants violated after {op} operation");
                }
            }
        }
    }
}