asupersync 0.3.1

//! Cache-local intrusive ring queue for scheduler hot paths.
//!
//! This module provides intrusive data structures that use link fields embedded
//! in `TaskRecord` rather than allocating separate nodes. This eliminates
//! per-operation allocations and improves cache locality.
//!
//! # Structures
//!
//! - [`IntrusiveRing`]: FIFO queue with O(1) push_back, pop_front, remove
//! - [`IntrusiveStack`]: LIFO stack with FIFO steal (work-stealing pattern)
//!
//! # Design
//!
//! - Links (`next_in_queue`, `prev_in_queue`, `queue_tag`) are stored in `TaskRecord`
//! - Queues maintain only head/tail indices into the task arena
//! - Each task can be in at most one queue (enforced by `queue_tag`)
//! - All operations are O(1) with zero allocations
//!
//! # When to Use
//!
//! Use intrusive structures when:
//! - You have exclusive `&mut Arena<TaskRecord>` access
//! - You need zero-allocation queue operations (hot paths)
//! - You're in a single-owner context (not shared across threads)
//!
//! Use regular queues (VecDeque, BinaryHeap) when:
//! - You need shared access (`Arc<Mutex<...>>`)
//! - You don't have arena access (TaskId-only APIs)
//! - You need priority ordering (heap-based)
//!
//! # Integration Points
//!
//! | Component | Uses | Why |
//! |-----------|------|-----|
//! | `ThreeLaneWorker` | Can use intrusive | Has arena via `state` |
//! | `PriorityScheduler` | Regular queues | No arena access, needs priority |
//! | `LocalQueue` | Intrusive stack | Shared via `Arc<Mutex>` + RuntimeState arena |
//! | `GlobalInjector` | Regular queues | Concurrent access |
//!
//! # Safety Proof: No ABA, No Use-After-Free
//!
//! ## Invariants (maintained by all operations):
//!
//! **INV-1 (Exclusive Access):** Every operation takes `&mut Arena<TaskRecord>`,
//! guaranteeing no concurrent mutation. This eliminates data races entirely.
//!
//! **INV-2 (Tag Consistency):** A task's `queue_tag` equals the queue's `tag`
//! if and only if the task is logically in that queue. On removal, `clear_queue_links()`
//! sets `queue_tag = 0` atomically with link erasure.
//!
//! **INV-3 (Link Validity):** If `task.next_in_queue = Some(id)`, then `id` is a
//! valid arena index with `queue_tag == self.tag`. Conversely on removal.
//!
//! ## No ABA:
//! ABA requires a slot to be freed and reallocated while a stale reference exists.
//! Since we use arena indices (not pointers), and INV-2 ensures `queue_tag` is zeroed
//! on removal, any stale index would fail the `is_in_queue_tag(tag)` check. The arena
//! itself is `&mut`-borrowed, preventing concurrent reuse of slots.
//!
//! ## No Use-After-Free:
//! Tasks are never freed while in a queue. The arena is `&mut`-borrowed during all
//! operations, so no external code can free arena entries during queue manipulation.
//! After `clear_queue_links()`, the task's link fields are zeroed, and subsequent
//! operations will not follow stale links.
//!
//! # Queue Tags
//!
//! | Tag | Queue |
//! |-----|-------|
//! | 0 | Not in any queue |
//! | 1 | Local ready queue |
//! | 2 | Local cancel queue |
//! | 3 | Reserved |

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

/// Queue tag for the local ready queue.
pub const QUEUE_TAG_READY: u8 = 1;

/// Queue tag for the local cancel queue.
pub const QUEUE_TAG_CANCEL: u8 = 2;

/// An intrusive doubly-linked ring queue.
///
/// The queue stores only head/tail indices; the actual links are stored
/// in `TaskRecord` fields. This provides O(1) operations with zero
/// per-operation allocations.
///
/// # Invariants
///
/// - If `head.is_none()`, then `tail.is_none()` and `len == 0`
/// - If `head.is_some()`, then `tail.is_some()` and `len > 0`
/// - For all tasks in the queue: `task.queue_tag == self.tag`
/// - The list forms a proper doubly-linked chain from head to tail
#[derive(Debug)]
pub struct IntrusiveRing {
    /// First task in the queue (front for pop_front).
    head: Option<TaskId>,
    /// Last task in the queue (back for push_back).
    tail: Option<TaskId>,
    /// Number of tasks in the queue.
    len: usize,
    /// Queue tag for membership detection.
    tag: u8,
}

impl IntrusiveRing {
    /// Creates a new empty intrusive ring with the given queue tag.
    #[must_use]
    pub const fn new(tag: u8) -> Self {
        assert!(tag != 0, "queue tag 0 is reserved for \"not in any queue\"");
        Self {
            head: None,
            tail: None,
            len: 0,
            tag,
        }
    }

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

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

    /// Returns the queue tag.
    #[must_use]
    #[inline]
    pub const fn tag(&self) -> u8 {
        self.tag
    }

    /// Pushes a task to the back of the queue.
    ///
    /// # Panics
    ///
    /// Panics if the task is already in a queue (queue_tag != 0).
    ///
    /// # Complexity
    ///
    /// O(1) time, O(0) allocations.
    #[inline]
    pub fn push_back(&mut self, task_id: TaskId, arena: &mut Arena<TaskRecord>) {
        let Some(record) = arena.get_mut(task_id.arena_index()) else {
            return;
        };

        // Check for double-enqueue
        let in_queue = record.is_in_queue();
        debug_assert!(
            !in_queue,
            "task {:?} already in queue (tag={})",
            task_id, record.queue_tag
        );

        if in_queue {
            return;
        }

        match self.tail {
            None => {
                // Empty queue: new task becomes both head and tail
                record.set_queue_links(None, None, self.tag);
                self.head = Some(task_id);
                self.tail = Some(task_id);
            }
            Some(old_tail) => {
                // Link new task after current tail
                record.set_queue_links(Some(old_tail), None, self.tag);

                // Update old tail's next pointer
                if let Some(old_tail_record) = arena.get_mut(old_tail.arena_index()) {
                    old_tail_record.next_in_queue = Some(task_id);
                }

                self.tail = Some(task_id);
            }
        }

        self.len += 1;
    }

    /// Pops a task from the front of the queue.
    ///
    /// Returns `None` if the queue is empty.
    ///
    /// # Complexity
    ///
    /// O(1) time, O(0) allocations.
    #[inline]
    #[must_use]
    pub fn pop_front(&mut self, arena: &mut Arena<TaskRecord>) -> Option<TaskId> {
        let head_id = self.head?;

        let next = {
            let record = arena
                .get_mut(head_id.arena_index())
                .expect("intrusive list broken: task removed from arena while in queue");

            // Verify the task is actually in this queue
            debug_assert!(
                record.is_in_queue_tag(self.tag),
                "head task {:?} has wrong tag (expected {}, got {})",
                head_id,
                self.tag,
                record.queue_tag
            );

            let next = record.next_in_queue;
            record.clear_queue_links();
            next
        };

        self.head = next;

        match next {
            None => {
                // Queue is now empty
                self.tail = None;
            }
            Some(new_head) => {
                // Update new head's prev pointer
                if let Some(new_head_record) = arena.get_mut(new_head.arena_index()) {
                    new_head_record.prev_in_queue = None;
                }
            }
        }

        self.len -= 1;
        Some(head_id)
    }

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

        // Check if task is in this queue
        if !record.is_in_queue_tag(self.tag) {
            return false;
        }

        let prev = record.prev_in_queue;
        let next = record.next_in_queue;

        // Clear the removed task's links
        record.clear_queue_links();

        // Update predecessor's next pointer
        match prev {
            None => {
                // Task was the head
                self.head = next;
            }
            Some(prev_id) => {
                if let Some(prev_record) = arena.get_mut(prev_id.arena_index()) {
                    prev_record.next_in_queue = next;
                }
            }
        }

        // Update successor's prev pointer
        match next {
            None => {
                // Task was the tail
                self.tail = prev;
            }
            Some(next_id) => {
                if let Some(next_record) = arena.get_mut(next_id.arena_index()) {
                    next_record.prev_in_queue = prev;
                }
            }
        }

        self.len -= 1;
        true
    }

    /// Returns true if the given task is in this queue.
    ///
    /// # Complexity
    ///
    /// O(1) time.
    #[must_use]
    pub fn contains(&self, task_id: TaskId, arena: &Arena<TaskRecord>) -> bool {
        arena
            .get(task_id.arena_index())
            .is_some_and(|record| record.is_in_queue_tag(self.tag))
    }

    /// Returns the head task ID without removing it.
    #[must_use]
    #[inline]
    pub const fn peek_front(&self) -> Option<TaskId> {
        self.head
    }

    /// Clears the queue, removing all tasks.
    ///
    /// # Complexity
    ///
    /// O(n) time to clear all links.
    pub fn clear(&mut self, arena: &mut Arena<TaskRecord>) {
        let mut current = self.head;
        while let Some(task_id) = current {
            if let Some(record) = arena.get_mut(task_id.arena_index()) {
                let next = record.next_in_queue;
                record.clear_queue_links();
                current = next;
            } else {
                break;
            }
        }

        self.head = None;
        self.tail = None;
        self.len = 0;
    }
}

impl Default for IntrusiveRing {
    fn default() -> Self {
        Self::new(QUEUE_TAG_READY)
    }
}

/// An intrusive LIFO stack for work-stealing local queues.
///
/// Unlike `IntrusiveRing` which is FIFO, this stack provides LIFO
/// semantics for the owner while supporting FIFO stealing.
/// This matches the cache-locality optimization of processing
/// recently-pushed work first.
///
/// # Invariants
///
/// - If `top.is_none()`, then `len == 0`
/// - For all tasks in the stack: `task.queue_tag == self.tag`
#[derive(Debug)]
pub struct IntrusiveStack {
    /// Top of the stack (most recently pushed).
    top: Option<TaskId>,
    /// Bottom of the stack (oldest, for stealing).
    bottom: Option<TaskId>,
    /// Number of tasks in the stack.
    len: usize,
    /// Number of local (`!Send`) tasks currently in the stack.
    local_count: usize,
    /// Queue tag for membership detection.
    tag: u8,
}

impl IntrusiveStack {
    /// Creates a new empty intrusive stack with the given queue tag.
    #[must_use]
    pub const fn new(tag: u8) -> Self {
        assert!(tag != 0, "queue tag 0 is reserved for \"not in any queue\"");
        Self {
            top: None,
            bottom: None,
            len: 0,
            local_count: 0,
            tag,
        }
    }

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

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

    /// Returns true if the stack currently holds any local (`!Send`) tasks.
    #[must_use]
    #[inline]
    pub const fn has_local_tasks(&self) -> bool {
        self.local_count != 0
    }

    /// Pushes a task onto the top of the stack.
    ///
    /// # Complexity
    ///
    /// O(1) time, O(0) allocations.
    #[inline]
    pub fn push(&mut self, task_id: TaskId, arena: &mut Arena<TaskRecord>) {
        let Some(record) = arena.get_mut(task_id.arena_index()) else {
            return;
        };

        if record.is_in_queue() {
            return;
        }
        let is_local = record.is_local();

        match self.top {
            None => {
                // Empty stack
                record.set_queue_links(None, None, self.tag);
                self.top = Some(task_id);
                self.bottom = Some(task_id);
            }
            Some(old_top) => {
                // Link new task as new top, pointing down to old top
                record.set_queue_links(None, Some(old_top), self.tag);

                // Update old top's prev pointer (points up to new top)
                if let Some(old_top_record) = arena.get_mut(old_top.arena_index()) {
                    old_top_record.prev_in_queue = Some(task_id);
                }

                self.top = Some(task_id);
            }
        }

        self.len += 1;
        if is_local {
            self.local_count += 1;
        }
    }

    /// Pushes a known non-local task onto the top of the stack.
    ///
    /// This avoids per-item locality bookkeeping in hot steal-batch paths that
    /// have already proven the source queue contains no local tasks.
    #[inline]
    #[allow(dead_code)] // reserved for future steal-batch optimization
    pub(crate) fn push_assume_non_local(&mut self, task_id: TaskId, arena: &mut Arena<TaskRecord>) {
        let Some(record) = arena.get(task_id.arena_index()) else {
            return;
        };

        if record.is_in_queue() {
            return;
        }
        if record.is_local() {
            // Harden against callers that violate the contract in release builds.
            // Fall back to the safe path that maintains local_count invariants.
            self.push(task_id, arena);
            return;
        }

        let Some(record) = arena.get_mut(task_id.arena_index()) else {
            return;
        };

        match self.top {
            None => {
                // Empty stack
                record.set_queue_links(None, None, self.tag);
                self.top = Some(task_id);
                self.bottom = Some(task_id);
            }
            Some(old_top) => {
                // Link new task as new top, pointing down to old top
                record.set_queue_links(None, Some(old_top), self.tag);

                // Update old top's prev pointer (points up to new top)
                if let Some(old_top_record) = arena.get_mut(old_top.arena_index()) {
                    old_top_record.prev_in_queue = Some(task_id);
                }

                self.top = Some(task_id);
            }
        }

        self.len += 1;
    }

    /// Pushes a task onto the bottom of the stack.
    ///
    /// This is used by steal paths that temporarily remove local (`!Send`) tasks
    /// while scanning for stealable work, then restore them without perturbing
    /// owner-observed ordering.
    #[inline]
    pub fn push_bottom(&mut self, task_id: TaskId, arena: &mut Arena<TaskRecord>) {
        let Some(record) = arena.get_mut(task_id.arena_index()) else {
            return;
        };

        if record.is_in_queue() {
            return;
        }
        let is_local = record.is_local();

        match self.bottom {
            None => {
                // Empty stack.
                record.set_queue_links(None, None, self.tag);
                self.top = Some(task_id);
                self.bottom = Some(task_id);
            }
            Some(old_bottom) => {
                // Link new task as older than current bottom.
                record.set_queue_links(Some(old_bottom), None, self.tag);

                if let Some(old_bottom_record) = arena.get_mut(old_bottom.arena_index()) {
                    old_bottom_record.next_in_queue = Some(task_id);
                }

                self.bottom = Some(task_id);
            }
        }

        self.len += 1;
        if is_local {
            self.local_count += 1;
        }
    }

    /// Pops a task from the top of the stack (LIFO).
    ///
    /// # Complexity
    ///
    /// O(1) time, O(0) allocations.
    #[inline]
    #[must_use]
    pub fn pop(&mut self, arena: &mut Arena<TaskRecord>) -> Option<TaskId> {
        let top_id = self.top?;

        let (next_down, is_local) = {
            let record = arena
                .get_mut(top_id.arena_index())
                .expect("intrusive list broken: task removed from arena while in queue");
            let is_local = record.is_local();
            let next_down = record.next_in_queue; // Points down to older task
            record.clear_queue_links();
            (next_down, is_local)
        };

        self.top = next_down;

        match next_down {
            None => {
                // Stack is now empty
                self.bottom = None;
            }
            Some(new_top) => {
                // Update new top's prev pointer
                if let Some(new_top_record) = arena.get_mut(new_top.arena_index()) {
                    new_top_record.prev_in_queue = None;
                }
            }
        }

        self.len -= 1;
        if is_local {
            debug_assert!(self.local_count > 0);
            self.local_count -= 1;
        }
        Some(top_id)
    }

    /// Steals tasks from the bottom of the stack (FIFO for stealing).
    ///
    /// Returns up to `max_steal` tasks, starting from the oldest.
    ///
    /// # Complexity
    ///
    /// O(k) time where k is the number stolen. Allocates a Vec for the result.
    pub fn steal_batch(
        &mut self,
        max_steal: usize,
        arena: &mut Arena<TaskRecord>,
        stolen: &mut Vec<TaskId>,
    ) {
        if self.is_empty() {
            return;
        }
        let steal_count = (self.len / 2).max(1).min(max_steal);

        for _ in 0..steal_count {
            if let Some((bottom_id, _)) = self.steal_one_with_locality(arena) {
                stolen.push(bottom_id);
            } else {
                break;
            }
        }
    }

    /// Steals up to `max_steal` tasks into the destination stack.
    ///
    /// Returns the number of tasks transferred.
    ///
    /// # Complexity
    ///
    /// O(k) time where k is the number stolen. No allocations.
    pub fn steal_batch_into(
        &mut self,
        max_steal: usize,
        arena: &mut Arena<TaskRecord>,
        dest: &mut Self,
    ) -> usize {
        if self.is_empty() {
            return 0;
        }
        let steal_count = (self.len / 2).max(1).min(max_steal);
        let mut stolen = 0;

        for _ in 0..steal_count {
            if let Some((task_id, _)) = self.steal_one_with_locality(arena) {
                dest.push(task_id, arena);
                stolen += 1;
            } else {
                break;
            }
        }

        stolen
    }

    /// Steals up to `max_steal` known non-local tasks into `dest`.
    ///
    /// Caller must ensure `self.has_local_tasks() == false`.
    #[inline]
    #[allow(dead_code)] // Work-stealing scheduler integration path
    pub(crate) fn steal_batch_into_non_local(
        &mut self,
        max_steal: usize,
        arena: &mut Arena<TaskRecord>,
        dest: &mut Self,
    ) -> usize {
        debug_assert!(
            !self.has_local_tasks(),
            "steal_batch_into_non_local called on stack with local tasks"
        );
        if self.is_empty() {
            return 0;
        }
        let steal_count = (self.len / 2).max(1).min(max_steal);

        if self.tag == dest.tag {
            return self.splice_same_tag_non_local_batch(steal_count, arena, dest);
        }

        self.rebuild_non_local_batch(steal_count, arena, dest)
    }

    #[inline]
    fn splice_same_tag_non_local_batch(
        &mut self,
        steal_count: usize,
        arena: &mut Arena<TaskRecord>,
        dest: &mut Self,
    ) -> usize {
        let Some(segment_bottom) = self.bottom else {
            return 0;
        };

        // When both stacks share the same tag (the LocalQueue hot path),
        // we can splice the validated non-local bottom segment directly
        // into the destination instead of clearing and rebuilding each
        // stolen node one at a time.
        let mut segment_top = segment_bottom;
        let mut new_src_bottom = None;
        let mut stolen = 0usize;
        let mut current = segment_bottom;

        while stolen < steal_count {
            let Some(record) = arena.get(current.arena_index()) else {
                break;
            };
            if record.is_local() {
                // Source queue contract was violated; repair the local-task
                // summary and stop this fast path to avoid stealing !Send work.
                self.local_count = self.local_count.max(1);
                break;
            }

            segment_top = current;
            new_src_bottom = record.prev_in_queue;
            stolen += 1;

            match record.prev_in_queue {
                Some(next_up) => current = next_up,
                None => break,
            }
        }

        if stolen == 0 {
            return 0;
        }

        if new_src_bottom.is_some_and(|task_id| arena.get(task_id.arena_index()).is_none())
            || dest
                .top
                .is_some_and(|task_id| arena.get(task_id.arena_index()).is_none())
        {
            return 0;
        }

        self.bottom = new_src_bottom;
        match new_src_bottom {
            None => {
                self.top = None;
            }
            Some(new_bottom) => {
                if let Some(new_bottom_record) = arena.get_mut(new_bottom.arena_index()) {
                    new_bottom_record.next_in_queue = None;
                }
            }
        }
        self.len -= stolen;

        if let Some(segment_top_record) = arena.get_mut(segment_top.arena_index()) {
            segment_top_record.prev_in_queue = None;
        }

        match dest.top {
            None => {
                if let Some(segment_bottom_record) = arena.get_mut(segment_bottom.arena_index()) {
                    segment_bottom_record.next_in_queue = None;
                }
                dest.top = Some(segment_top);
                dest.bottom = Some(segment_bottom);
            }
            Some(old_top) => {
                if let Some(segment_bottom_record) = arena.get_mut(segment_bottom.arena_index()) {
                    segment_bottom_record.next_in_queue = Some(old_top);
                }
                if let Some(old_top_record) = arena.get_mut(old_top.arena_index()) {
                    old_top_record.prev_in_queue = Some(segment_bottom);
                }
                dest.top = Some(segment_top);
            }
        }

        dest.len += stolen;
        stolen
    }

    #[inline]
    fn rebuild_non_local_batch(
        &mut self,
        steal_count: usize,
        arena: &mut Arena<TaskRecord>,
        dest: &mut Self,
    ) -> usize {
        let mut stolen = 0;

        for _ in 0..steal_count {
            let Some(bottom_id) = self.bottom else {
                break;
            };

            // Detach oldest task from source stack.
            let prev_up = {
                let Some(record) = arena.get_mut(bottom_id.arena_index()) else {
                    break;
                };
                if record.is_local() {
                    // Source queue contract was violated; repair the local-task
                    // summary and stop this fast path to avoid stealing !Send work.
                    self.local_count = self.local_count.max(1);
                    break;
                }
                record.prev_in_queue
            };

            self.bottom = prev_up;
            match prev_up {
                None => {
                    // Source stack is now empty.
                    self.top = None;
                }
                Some(new_bottom) => {
                    if let Some(new_bottom_record) = arena.get_mut(new_bottom.arena_index()) {
                        new_bottom_record.next_in_queue = None;
                    }
                }
            }
            self.len -= 1;

            // Attach directly to destination top.
            let Some(record) = arena.get_mut(bottom_id.arena_index()) else {
                break;
            };
            match dest.top {
                None => {
                    record.set_queue_links(None, None, dest.tag);
                    dest.top = Some(bottom_id);
                    dest.bottom = Some(bottom_id);
                }
                Some(old_top) => {
                    record.set_queue_links(None, Some(old_top), dest.tag);
                    if let Some(old_top_record) = arena.get_mut(old_top.arena_index()) {
                        old_top_record.prev_in_queue = Some(bottom_id);
                    }
                    dest.top = Some(bottom_id);
                }
            }

            dest.len += 1;
            stolen += 1;
        }

        stolen
    }

    /// Steals one task from the bottom of the stack.
    ///
    /// Returns the stolen task and whether it is local (`!Send`), allowing
    /// callers to avoid an extra arena lookup on steal paths that need locality.
    #[inline]
    #[must_use]
    pub(crate) fn steal_one_with_locality(
        &mut self,
        arena: &mut Arena<TaskRecord>,
    ) -> Option<(TaskId, bool)> {
        let bottom_id = self.bottom?;

        let (prev_up, is_local) = {
            let record = arena.get_mut(bottom_id.arena_index())?;
            let is_local = record.is_local();
            let prev_up = record.prev_in_queue; // Points up to newer task
            record.clear_queue_links();
            (prev_up, is_local)
        };

        self.bottom = prev_up;

        match prev_up {
            None => {
                // Stack is now empty
                self.top = None;
            }
            Some(new_bottom) => {
                // Update new bottom's next pointer
                if let Some(new_bottom_record) = arena.get_mut(new_bottom.arena_index()) {
                    new_bottom_record.next_in_queue = None;
                }
            }
        }

        self.len -= 1;
        if is_local {
            debug_assert!(self.local_count > 0);
            self.local_count -= 1;
        }
        Some((bottom_id, is_local))
    }

    /// Steals one task from the bottom when the stack contains no local tasks.
    ///
    /// Caller must ensure `self.has_local_tasks() == false`.
    #[inline]
    #[must_use]
    #[allow(dead_code)] // Work-stealing scheduler integration path
    pub(crate) fn steal_one_assume_non_local(
        &mut self,
        arena: &mut Arena<TaskRecord>,
    ) -> Option<TaskId> {
        debug_assert!(
            !self.has_local_tasks(),
            "steal_one_assume_non_local called on stack with local tasks"
        );
        let bottom_id = self.bottom?;

        let prev_up = {
            let record = arena.get_mut(bottom_id.arena_index())?;
            if record.is_local() {
                // Source queue contract was violated; repair the local-task
                // summary and refuse to steal this task.
                self.local_count = self.local_count.max(1);
                return None;
            }
            let prev_up = record.prev_in_queue; // Points up to newer task
            record.clear_queue_links();
            prev_up
        };

        self.bottom = prev_up;

        match prev_up {
            None => {
                // Stack is now empty
                self.top = None;
            }
            Some(new_bottom) => {
                // Update new bottom's next pointer
                if let Some(new_bottom_record) = arena.get_mut(new_bottom.arena_index()) {
                    new_bottom_record.next_in_queue = None;
                }
            }
        }

        self.len -= 1;
        Some(bottom_id)
    }

    /// Steals one task from the bottom of the stack.
    #[inline]
    #[must_use]
    #[allow(dead_code)] // Work-stealing scheduler integration path
    pub(crate) fn steal_one(&mut self, arena: &mut Arena<TaskRecord>) -> Option<TaskId> {
        self.steal_one_with_locality(arena)
            .map(|(task_id, _)| task_id)
    }

    /// Returns true if the given task is in this stack.
    #[must_use]
    pub fn contains(&self, task_id: TaskId, arena: &Arena<TaskRecord>) -> bool {
        arena
            .get(task_id.arena_index())
            .is_some_and(|record| record.is_in_queue_tag(self.tag))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::record::task::TaskRecord;
    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_ring(ring: &mut IntrusiveRing, arena: &mut Arena<TaskRecord>) -> Vec<TaskId> {
        let mut popped = Vec::new();
        while let Some(task_id) = ring.pop_front(arena) {
            popped.push(task_id);
        }
        popped
    }

    fn pop_all_stack(stack: &mut IntrusiveStack, arena: &mut Arena<TaskRecord>) -> Vec<TaskId> {
        let mut popped = Vec::new();
        while let Some(task_id) = stack.pop(arena) {
            popped.push(task_id);
        }
        popped
    }

    #[test]
    fn empty_queue() {
        let ring = IntrusiveRing::new(QUEUE_TAG_READY);
        assert!(ring.is_empty());
        assert_eq!(ring.len(), 0);
        assert!(ring.peek_front().is_none());
    }

    #[test]
    fn default_ring_uses_ready_tag() {
        let mut arena = setup_arena(1);
        let mut ring = IntrusiveRing::default();
        assert_eq!(ring.tag(), QUEUE_TAG_READY);

        ring.push_back(task(0), &mut arena);
        assert_eq!(ring.pop_front(&mut arena), Some(task(0)));
        assert!(ring.is_empty());
    }

    #[test]
    #[should_panic(expected = "queue tag 0 is reserved")]
    fn ring_rejects_zero_tag() {
        let _ring = IntrusiveRing::new(0);
    }

    #[test]
    #[should_panic(expected = "queue tag 0 is reserved")]
    fn stack_rejects_zero_tag() {
        let _stack = IntrusiveStack::new(0);
    }

    #[test]
    fn push_pop_single() {
        let mut arena = setup_arena(1);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        ring.push_back(task(0), &mut arena);
        assert_eq!(ring.len(), 1);
        assert!(!ring.is_empty());
        assert_eq!(ring.peek_front(), Some(task(0)));

        let popped = ring.pop_front(&mut arena);
        assert_eq!(popped, Some(task(0)));
        assert!(ring.is_empty());
        assert_eq!(ring.len(), 0);

        // Verify task links are cleared
        let record = arena.get(task(0).arena_index()).unwrap();
        assert!(!record.is_in_queue());
    }

    #[test]
    fn fifo_ordering() {
        let mut arena = setup_arena(5);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        // Push 0, 1, 2, 3, 4
        for i in 0..5 {
            ring.push_back(task(i), &mut arena);
        }
        assert_eq!(ring.len(), 5);

        // Pop should return 0, 1, 2, 3, 4 (FIFO)
        for i in 0..5 {
            let popped = ring.pop_front(&mut arena);
            assert_eq!(popped, Some(task(i)), "expected task {i}");
        }
        assert!(ring.is_empty());
    }

    #[test]
    fn remove_from_middle() {
        let mut arena = setup_arena(5);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        for i in 0..5 {
            ring.push_back(task(i), &mut arena);
        }

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

        // Verify task 2's links are cleared
        let record = arena.get(task(2).arena_index()).unwrap();
        assert!(!record.is_in_queue());

        // Pop remaining: 0, 1, 3, 4
        assert_eq!(ring.pop_front(&mut arena), Some(task(0)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(1)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(3)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(4)));
        assert!(ring.is_empty());
    }

    #[test]
    fn remove_head() {
        let mut arena = setup_arena(3);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        for i in 0..3 {
            ring.push_back(task(i), &mut arena);
        }

        // Remove head (task 0)
        let removed = ring.remove(task(0), &mut arena);
        assert!(removed);
        assert_eq!(ring.len(), 2);

        // Pop remaining: 1, 2
        assert_eq!(ring.pop_front(&mut arena), Some(task(1)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(2)));
    }

    #[test]
    fn remove_tail() {
        let mut arena = setup_arena(3);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        for i in 0..3 {
            ring.push_back(task(i), &mut arena);
        }

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

        // Pop remaining: 0, 1
        assert_eq!(ring.pop_front(&mut arena), Some(task(0)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(1)));
    }

    #[test]
    fn remove_only_element() {
        let mut arena = setup_arena(1);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        ring.push_back(task(0), &mut arena);
        let removed = ring.remove(task(0), &mut arena);

        assert!(removed);
        assert!(ring.is_empty());
        assert!(ring.head.is_none());
        assert!(ring.tail.is_none());
    }

    #[test]
    fn remove_not_in_queue() {
        let mut arena = setup_arena(2);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        ring.push_back(task(0), &mut arena);

        // Try to remove task 1 which is not in the queue
        let removed = ring.remove(task(1), &mut arena);
        assert!(!removed);
        assert_eq!(ring.len(), 1);
    }

    #[test]
    fn contains() {
        let mut arena = setup_arena(3);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        ring.push_back(task(0), &mut arena);
        ring.push_back(task(1), &mut arena);

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

        let _ = ring.pop_front(&mut arena);
        assert!(!ring.contains(task(0), &arena));
        assert!(ring.contains(task(1), &arena));
    }

    #[test]
    fn clear() {
        let mut arena = setup_arena(5);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        for i in 0..5 {
            ring.push_back(task(i), &mut arena);
        }

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

        // Verify all tasks have cleared links
        for i in 0..5 {
            let record = arena.get(task(i).arena_index()).unwrap();
            assert!(!record.is_in_queue());
        }
    }

    #[test]
    fn different_queue_tags() {
        let mut arena = setup_arena(4);
        let mut ready_ring = IntrusiveRing::new(QUEUE_TAG_READY);
        let mut cancel_ring = IntrusiveRing::new(QUEUE_TAG_CANCEL);

        // Put tasks 0,1 in ready queue and tasks 2,3 in cancel queue
        ready_ring.push_back(task(0), &mut arena);
        ready_ring.push_back(task(1), &mut arena);
        cancel_ring.push_back(task(2), &mut arena);
        cancel_ring.push_back(task(3), &mut arena);

        // Verify containment
        assert!(ready_ring.contains(task(0), &arena));
        assert!(ready_ring.contains(task(1), &arena));
        assert!(!ready_ring.contains(task(2), &arena));
        assert!(!ready_ring.contains(task(3), &arena));

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

        // Cannot remove task from wrong queue
        assert!(!ready_ring.remove(task(2), &mut arena));
        assert!(!cancel_ring.remove(task(0), &mut arena));

        // Can remove from correct queue
        assert!(ready_ring.remove(task(0), &mut arena));
        assert!(cancel_ring.remove(task(2), &mut arena));
    }

    #[test]
    fn interleaved_push_pop() {
        let mut arena = setup_arena(10);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        ring.push_back(task(0), &mut arena);
        ring.push_back(task(1), &mut arena);
        assert_eq!(ring.pop_front(&mut arena), Some(task(0)));

        ring.push_back(task(2), &mut arena);
        assert_eq!(ring.pop_front(&mut arena), Some(task(1)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(2)));

        ring.push_back(task(3), &mut arena);
        ring.push_back(task(4), &mut arena);
        ring.push_back(task(5), &mut arena);

        assert_eq!(ring.len(), 3);
        assert_eq!(ring.pop_front(&mut arena), Some(task(3)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(4)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(5)));
        assert!(ring.is_empty());
    }

    #[test]
    fn high_volume() {
        let count = 1000u32;
        let mut arena = setup_arena(count);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        for i in 0..count {
            ring.push_back(task(i), &mut arena);
        }
        assert_eq!(ring.len(), count as usize);

        for i in 0..count {
            let popped = ring.pop_front(&mut arena);
            assert_eq!(popped, Some(task(i)));
        }
        assert!(ring.is_empty());
    }

    #[test]
    fn reuse_after_pop() {
        let mut arena = setup_arena(2);
        let mut ring = IntrusiveRing::new(QUEUE_TAG_READY);

        // Push and pop task 0
        ring.push_back(task(0), &mut arena);
        assert_eq!(ring.pop_front(&mut arena), Some(task(0)));

        // Re-enqueue task 0
        ring.push_back(task(0), &mut arena);
        ring.push_back(task(1), &mut arena);

        // Should get task 0 first (FIFO)
        assert_eq!(ring.pop_front(&mut arena), Some(task(0)));
        assert_eq!(ring.pop_front(&mut arena), Some(task(1)));
    }

    #[test]
    fn metamorphic_ring_remove_matches_fifo_filter() {
        let mut baseline_arena = setup_arena(8);
        let mut filtered_arena = setup_arena(8);
        let mut baseline = IntrusiveRing::new(QUEUE_TAG_READY);
        let mut filtered = IntrusiveRing::new(QUEUE_TAG_READY);
        let removed = [task(1), task(4), task(6)];

        for i in 0..8 {
            baseline.push_back(task(i), &mut baseline_arena);
            filtered.push_back(task(i), &mut filtered_arena);
        }

        for task_id in removed {
            assert!(filtered.remove(task_id, &mut filtered_arena));
        }

        let expected: Vec<_> = pop_all_ring(&mut baseline, &mut baseline_arena)
            .into_iter()
            .filter(|task_id| !removed.contains(task_id))
            .collect();
        let actual = pop_all_ring(&mut filtered, &mut filtered_arena);

        assert_eq!(actual, expected);
        for task_id in removed {
            let record = filtered_arena
                .get(task_id.arena_index())
                .expect("removed task missing");
            assert!(!record.is_in_queue());
        }
    }

    // ── IntrusiveStack tests ─────────────────────────────────────────────

    #[test]
    fn stack_empty() {
        let stack = IntrusiveStack::new(QUEUE_TAG_READY);
        assert!(stack.is_empty());
        assert_eq!(stack.len(), 0);
    }

    #[test]
    fn stack_push_pop_single() {
        let mut arena = setup_arena(1);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        stack.push(task(0), &mut arena);
        assert_eq!(stack.len(), 1);
        assert!(!stack.is_empty());

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

    #[test]
    fn stack_tracks_local_task_presence() {
        let mut arena = setup_arena(2);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        arena
            .get_mut(task(0).arena_index())
            .expect("task record missing")
            .mark_local();

        assert!(!stack.has_local_tasks());
        stack.push(task(0), &mut arena);
        stack.push(task(1), &mut arena);
        assert!(stack.has_local_tasks());

        let (stolen_id, is_local) = stack
            .steal_one_with_locality(&mut arena)
            .expect("oldest task missing");
        assert_eq!(stolen_id, task(0));
        assert!(is_local);
        assert!(!stack.has_local_tasks());

        assert_eq!(stack.pop(&mut arena), Some(task(1)));
        assert!(!stack.has_local_tasks());
    }

    #[test]
    fn stack_lifo_ordering() {
        let mut arena = setup_arena(5);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        // Push 0, 1, 2, 3, 4
        for i in 0..5 {
            stack.push(task(i), &mut arena);
        }
        assert_eq!(stack.len(), 5);

        // Pop should return 4, 3, 2, 1, 0 (LIFO)
        for i in (0..5).rev() {
            let popped = stack.pop(&mut arena);
            assert_eq!(popped, Some(task(i)), "expected task {i}");
        }
        assert!(stack.is_empty());
    }

    #[test]
    fn stack_push_bottom_restores_owner_visible_order() {
        let mut arena = setup_arena(3);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        stack.push(task(0), &mut arena);
        stack.push(task(1), &mut arena);
        stack.push(task(2), &mut arena);

        // Temporarily remove oldest, then restore at bottom.
        let oldest = stack.steal_one(&mut arena).expect("oldest task missing");
        assert_eq!(oldest, task(0));
        stack.push_bottom(oldest, &mut arena);

        // Owner-observed order should remain unchanged.
        assert_eq!(stack.pop(&mut arena), Some(task(2)));
        assert_eq!(stack.pop(&mut arena), Some(task(1)));
        assert_eq!(stack.pop(&mut arena), Some(task(0)));
        assert_eq!(stack.pop(&mut arena), None);
    }

    #[test]
    fn stack_steal_fifo() {
        let mut arena = setup_arena(8);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        // Push 0, 1, 2, 3, 4, 5, 6, 7
        for i in 0..8 {
            stack.push(task(i), &mut arena);
        }

        // Steal should return oldest first (FIFO for stealing)
        let mut stolen = Vec::new();
        stack.steal_batch(4, &mut arena, &mut stolen);
        assert_eq!(stolen.len(), 4);
        // Should have stolen 0, 1, 2, 3 (the oldest)
        for (i, task_id) in stolen.into_iter().enumerate() {
            assert_eq!(task_id, task(i as u32), "stolen task {i}");
        }

        // Remaining should be 7, 6, 5, 4 (LIFO order)
        assert_eq!(stack.len(), 4);
        assert_eq!(stack.pop(&mut arena), Some(task(7)));
        assert_eq!(stack.pop(&mut arena), Some(task(6)));
        assert_eq!(stack.pop(&mut arena), Some(task(5)));
        assert_eq!(stack.pop(&mut arena), Some(task(4)));
    }

    #[test]
    fn stack_steal_batch_into_non_local_preserves_ordering() {
        let mut arena = setup_arena(8);
        let mut src = IntrusiveStack::new(QUEUE_TAG_READY);
        let mut dest = IntrusiveStack::new(QUEUE_TAG_CANCEL);

        for i in 0..8 {
            src.push(task(i), &mut arena);
        }

        let stolen = src.steal_batch_into_non_local(4, &mut arena, &mut dest);
        assert_eq!(stolen, 4);
        assert!(!src.has_local_tasks());
        assert!(!dest.has_local_tasks());

        // Destination receives oldest tasks (0..3), but push-to-top means
        // owner pop order is reverse in destination.
        assert_eq!(dest.pop(&mut arena), Some(task(3)));
        assert_eq!(dest.pop(&mut arena), Some(task(2)));
        assert_eq!(dest.pop(&mut arena), Some(task(1)));
        assert_eq!(dest.pop(&mut arena), Some(task(0)));
        assert_eq!(dest.pop(&mut arena), None);

        // Source retains newest half.
        assert_eq!(src.pop(&mut arena), Some(task(7)));
        assert_eq!(src.pop(&mut arena), Some(task(6)));
        assert_eq!(src.pop(&mut arena), Some(task(5)));
        assert_eq!(src.pop(&mut arena), Some(task(4)));
        assert_eq!(src.pop(&mut arena), None);
    }

    #[test]
    fn stack_steal_batch_into_same_tag_splices_above_existing_destination() {
        let mut arena = setup_arena(8);
        let mut src = IntrusiveStack::new(QUEUE_TAG_READY);
        let mut dest = IntrusiveStack::new(QUEUE_TAG_READY);

        for i in 0..6 {
            src.push(task(i), &mut arena);
        }
        dest.push(task(6), &mut arena);
        dest.push(task(7), &mut arena);

        let stolen = src.steal_batch_into_non_local(3, &mut arena, &mut dest);
        assert_eq!(stolen, 3);
        assert!(!src.has_local_tasks());
        assert!(!dest.has_local_tasks());

        // Stolen oldest source tasks (0..2) sit above the existing destination
        // stack while preserving thief-visible LIFO order.
        assert_eq!(dest.pop(&mut arena), Some(task(2)));
        assert_eq!(dest.pop(&mut arena), Some(task(1)));
        assert_eq!(dest.pop(&mut arena), Some(task(0)));
        assert_eq!(dest.pop(&mut arena), Some(task(7)));
        assert_eq!(dest.pop(&mut arena), Some(task(6)));
        assert_eq!(dest.pop(&mut arena), None);

        // Source retains the newest half.
        assert_eq!(src.pop(&mut arena), Some(task(5)));
        assert_eq!(src.pop(&mut arena), Some(task(4)));
        assert_eq!(src.pop(&mut arena), Some(task(3)));
        assert_eq!(src.pop(&mut arena), None);
    }

    #[test]
    fn stack_work_stealing_semantics() {
        // Simulates owner pushing and popping while thief steals
        let mut arena = setup_arena(10);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        // Owner pushes 0, 1, 2, 3, 4, 5
        for i in 0..6 {
            stack.push(task(i), &mut arena);
        }

        // Owner pops most recent (5)
        assert_eq!(stack.pop(&mut arena), Some(task(5)));

        // Stack now has [0, 1, 2, 3, 4] with len=5
        // Thief steals oldest - steal_batch takes at most half the queue
        // (5/2).max(1).min(2) = 2, so 2 tasks stolen
        let mut stolen = Vec::new();
        stack.steal_batch(2, &mut arena, &mut stolen);
        assert_eq!(stolen.len(), 2);
        assert_eq!(stolen[0], task(0)); // Oldest is stolen first
        assert_eq!(stolen[1], task(1));

        // Stack now has [2, 3, 4] with len=3
        // Owner pushes 6
        stack.push(task(6), &mut arena);

        // Stack is [2, 3, 4, 6] - owner pops LIFO (6, 4, 3, 2)
        assert_eq!(stack.pop(&mut arena), Some(task(6)));
        assert_eq!(stack.pop(&mut arena), Some(task(4)));
        assert_eq!(stack.pop(&mut arena), Some(task(3)));
        assert_eq!(stack.pop(&mut arena), Some(task(2)));
        assert!(stack.is_empty());
    }

    #[test]
    fn metamorphic_batch_steal_matches_repeated_single_steals() {
        let mut batch_arena = setup_arena(8);
        let mut single_arena = setup_arena(8);
        let mut batch_stack = IntrusiveStack::new(QUEUE_TAG_READY);
        let mut single_stack = IntrusiveStack::new(QUEUE_TAG_READY);

        for i in 0..8 {
            batch_stack.push(task(i), &mut batch_arena);
            single_stack.push(task(i), &mut single_arena);
        }

        let mut batch_stolen = Vec::new();
        batch_stack.steal_batch(4, &mut batch_arena, &mut batch_stolen);

        let mut single_stolen = Vec::new();
        while single_stolen.len() < batch_stolen.len() {
            single_stolen.push(
                single_stack
                    .steal_one(&mut single_arena)
                    .expect("single steal should match batch partition"),
            );
        }

        assert_eq!(single_stolen, batch_stolen);
        assert_eq!(
            pop_all_stack(&mut single_stack, &mut single_arena),
            pop_all_stack(&mut batch_stack, &mut batch_arena)
        );
    }

    #[test]
    fn metamorphic_restoring_stolen_suffix_reconstructs_owner_order() {
        let mut baseline_arena = setup_arena(6);
        let mut restored_arena = setup_arena(6);
        let mut baseline = IntrusiveStack::new(QUEUE_TAG_READY);
        let mut restored = IntrusiveStack::new(QUEUE_TAG_READY);

        for i in 0..6 {
            baseline.push(task(i), &mut baseline_arena);
            restored.push(task(i), &mut restored_arena);
        }

        let expected = pop_all_stack(&mut baseline, &mut baseline_arena);

        let mut stolen = Vec::new();
        restored.steal_batch(3, &mut restored_arena, &mut stolen);
        for task_id in stolen.into_iter().rev() {
            restored.push_bottom(task_id, &mut restored_arena);
        }

        assert_eq!(pop_all_stack(&mut restored, &mut restored_arena), expected);
        assert!(!restored.has_local_tasks());
    }

    #[test]
    fn stack_steal_from_small() {
        let mut arena = setup_arena(2);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        stack.push(task(0), &mut arena);

        // Steal from single-element stack
        let mut stolen = Vec::new();
        stack.steal_batch(4, &mut arena, &mut stolen);
        assert_eq!(stolen.len(), 1);
        assert_eq!(stolen[0], task(0));
        assert!(stack.is_empty());
    }

    #[test]
    fn stack_steal_from_empty() {
        let mut arena = setup_arena(0);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        let mut stolen = Vec::new();
        stack.steal_batch(4, &mut arena, &mut stolen);
        assert!(stolen.is_empty());
    }

    #[test]
    fn stack_contains() {
        let mut arena = setup_arena(3);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

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

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

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

    #[test]
    fn stack_steal_one_assume_non_local_rejects_local_when_counter_stale() {
        let mut arena = setup_arena(1);
        let mut stack = IntrusiveStack::new(QUEUE_TAG_READY);

        arena
            .get_mut(task(0).arena_index())
            .expect("task record missing")
            .mark_local();
        stack.push(task(0), &mut arena);
        assert!(stack.has_local_tasks());

        // Simulate stale bookkeeping (e.g., contract violation elsewhere).
        stack.local_count = 0;
        assert!(!stack.has_local_tasks());

        // Fast-path non-local steal must refuse to steal local tasks.
        let stolen = stack.steal_one_assume_non_local(&mut arena);
        assert!(stolen.is_none());
        assert!(stack.has_local_tasks(), "local_count should self-heal");
        assert_eq!(stack.len(), 1);
        assert_eq!(stack.pop(&mut arena), Some(task(0)));
    }

    #[test]
    fn stack_steal_batch_into_non_local_rejects_local_when_counter_stale() {
        let mut arena = setup_arena(3);
        let mut src = IntrusiveStack::new(QUEUE_TAG_READY);
        let mut dest = IntrusiveStack::new(QUEUE_TAG_CANCEL);

        arena
            .get_mut(task(0).arena_index())
            .expect("task record missing")
            .mark_local();

        // Keep local task at bottom (oldest).
        src.push(task(0), &mut arena);
        src.push(task(1), &mut arena);
        src.push(task(2), &mut arena);
        assert!(src.has_local_tasks());

        // Simulate stale bookkeeping.
        src.local_count = 0;
        assert!(!src.has_local_tasks());

        let stolen = src.steal_batch_into_non_local(3, &mut arena, &mut dest);
        assert_eq!(stolen, 0, "must not steal local task via fast path");
        assert!(src.has_local_tasks(), "local_count should self-heal");
        assert_eq!(dest.len(), 0, "destination remains untouched");

        // Source order remains intact for owner pop.
        assert_eq!(src.pop(&mut arena), Some(task(2)));
        assert_eq!(src.pop(&mut arena), Some(task(1)));
        assert_eq!(src.pop(&mut arena), Some(task(0)));
        assert_eq!(src.pop(&mut arena), None);
    }
}