//! Bounded single CPU ready membership, selection and quantum custody.
//! The backend exclusively retains the frozen context and thread services.

use core::marker::PhantomData;
use crate::context::{self as context, Record as ContextRecord};
use turkos_os_threads::threads::{self as thread, StackObservation, ThreadId, ThreadRecord};

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Error {
    Configuration,
    Context,
    Capacity,
    Exhausted,
    Duplicate,
    Stale,
    State,
    Guard,
    Empty,
    Priority,
    Tick,
    Expired,
    Backend,
    Quarantined,
    Busy,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Class {
    Normal,
    Low,
}
impl Class {
    pub fn request(value: u8) -> Result<Self, Error> {
        match value {
            0 => Ok(Self::Normal),
            1 => Ok(Self::Low),
            _ => Err(Error::Priority),
        }
    }
    pub fn quantum(self) -> u32 {
        match self {
            Self::Normal => 4,
            Self::Low => 2,
        }
    }
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Execution {
    pub cpu: u32,
    pub masked: bool,
    pub in_irq: bool,
    pub thread: Option<ThreadId>,
    pub root: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Observation {
    pub thread: ThreadRecord,
    pub context: ContextRecord,
    pub stack: StackObservation,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TickObservation {
    pub cpu: u32,
    pub source: u64,
    pub sequence: u64,
    pub counter: u64,
    pub thread: ThreadId,
    pub slice: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Disposition {
    Ready,
    Sleeping,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Failure {
    pub error: Error,
    pub phase: context::FailurePhase,
}
/// # Safety
/// Backend exclusively owns live frozen Table/ContextBank services. Nonclone
/// Member values pin that exact ThreadId, ContextHandle and guarded stack; Drop
/// never frees backing. observe reads actual retained stack/guard/canary state
/// and validates the same live context binding, with no fabricated identity.
/// Each Tick owns one real timer/deferred delivery and is bound to its hardware
/// source, CPU, ThreadId generation and current slice epoch. Tick observations
/// and consumption never switch threads, allocate, sleep or run in IRQ here.
/// prepare failure leaves complete thread/register/resource inventory unchanged.
/// Prepared retains all staged ownership. commit uses the frozen context service
/// and its current actual capture; BeforeSwitch means nothing visible changed,
/// AfterSwitch covers any possible change. A completed switch closes IRQ custody
/// and observes the new actual Running context before returning. finish/abort
/// failures return the same Prepared; dropping it does not release resources.
/// All operations are bounded, serialized on the initial CPU with IRQ masked.
pub unsafe trait Backend {
    type Member;
    type Prepared;
    type Tick;
    fn execution(&self) -> Execution;
    fn observe(&self, member: &Self::Member) -> Result<Observation, Error>;
    fn tick(&self, tick: &Self::Tick) -> Result<TickObservation, Error>;
    fn consume_tick(&mut self, tick: Self::Tick);
    fn prepare(
        &mut self,
        source: Option<&Self::Member>,
        target: &Self::Member,
        epoch: u64,
        disposition: Disposition,
    ) -> Result<Self::Prepared, Error>;
    fn commit(
        &mut self,
        prepared: &mut Self::Prepared,
        source: Option<&Self::Member>,
        target: &Self::Member,
        epoch: u64,
        disposition: Disposition,
    ) -> Result<(), Failure>;
    fn finish(&mut self, prepared: Self::Prepared) -> Result<(), (Error, Self::Prepared)>;
    fn abort(&mut self, prepared: Self::Prepared) -> Result<(), (Error, Self::Prepared)>;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Handle {
    queue: u64,
    slot: usize,
    generation: u64,
}
impl Handle {
    pub fn slot(self) -> usize {
        self.slot
    }
    pub fn generation(self) -> u64 {
        self.generation
    }
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Record {
    pub handle: Handle,
    pub thread: ThreadId,
    pub class: Class,
    pub queued: bool,
    pub active: bool,
    pub remaining: u32,
    pub slice: u64,
    pub consumed_ticks: u64,
    pub slices: u64,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Snapshot {
    pub members: usize,
    pub queued: usize,
    pub active: usize,
    pub retired: usize,
    pub epoch: u64,
    pub consumed_ticks: u64,
    pub faulted: bool,
    pub retained_prepared: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Charge {
    pub remaining: u32,
    pub reschedule: bool,
    pub sequence: u64,
}
struct Entry<M> {
    generation: u64,
    member: Option<M>,
    thread: Option<ThreadId>,
    class: Class,
    remaining: u32,
    slice: u64,
    consumed_ticks: u64,
    slices: u64,
}
impl<M> Entry<M> {
    fn empty() -> Self {
        Self {
            generation: 0,
            member: None,
            thread: None,
            class: Class::Normal,
            remaining: 0,
            slice: 0,
            consumed_ticks: 0,
            slices: 0,
        }
    }
}
pub struct Queue<B: Backend, const N: usize> {
    backend: B,
    identity: u64,
    cpu: u32,
    entries: [Entry<B::Member>; N],
    ready: [usize; N],
    len: usize,
    active: Option<usize>,
    epoch: u64,
    tick_source: Option<u64>,
    last_tick: u64,
    last_counter: u64,
    total_ticks: u64,
    faulted: bool,
    retained: Option<B::Prepared>,
    _local: PhantomData<*mut ()>,
}
#[cfg(not(target_os = "none"))]
fn identity() -> Result<u64, Error> {
    use core::sync::atomic::{AtomicU64, Ordering};
    static NEXT: AtomicU64 = AtomicU64::new(1);
    NEXT.try_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1))
        .map_err(|_| Error::Exhausted)
}
#[cfg(target_os = "none")]
fn identity() -> Result<u64, Error> {
    unsafe extern "C" {
        fn __os_k1060_runqueue_identity() -> u64;
    }
    let value = unsafe { __os_k1060_runqueue_identity() };
    if value == 0 {
        Err(Error::Exhausted)
    } else {
        Ok(value)
    }
}
impl<B: Backend, const N: usize> Queue<B, N> {
    pub fn new(backend: B) -> Result<Self, (Error, B)> {
        let e = backend.execution();
        if N == 0 || N > 32 {
            return Err((Error::Configuration, backend));
        }
        if !e.masked || e.in_irq || e.thread.is_some() {
            return Err((Error::Context, backend));
        }
        let identity = match identity() {
            Ok(v) => v,
            Err(e) => return Err((e, backend)),
        };
        Ok(Self {
            backend,
            identity,
            cpu: e.cpu,
            entries: core::array::from_fn(|_| Entry::empty()),
            ready: [0; N],
            len: 0,
            active: None,
            epoch: 0,
            tick_source: None,
            last_tick: 0,
            last_counter: 0,
            total_ticks: 0,
            faulted: false,
            retained: None,
            _local: PhantomData,
        })
    }
    fn execution(&self) -> Result<Execution, Error> {
        if self.faulted {
            return Err(Error::Quarantined);
        }
        let e = self.backend.execution();
        if !e.masked || e.in_irq || e.cpu != self.cpu {
            return Err(Error::Context);
        }
        Ok(e)
    }
    fn handle(&self, index: usize) -> Handle {
        Handle {
            queue: self.identity,
            slot: index,
            generation: self.entries[index].generation,
        }
    }
    fn index(&self, h: Handle) -> Result<usize, Error> {
        if h.queue != self.identity || h.slot >= N {
            return Err(Error::Stale);
        }
        let s = &self.entries[h.slot];
        if s.member.is_none() || s.generation != h.generation {
            return Err(Error::Stale);
        }
        Ok(h.slot)
    }
    fn member(&self, i: usize) -> Result<&B::Member, Error> {
        self.entries
            .get(i)
            .and_then(|e| e.member.as_ref())
            .ok_or(Error::Stale)
    }
    fn observed(&self, i: usize) -> Result<Observation, Error> {
        let o = self.backend.observe(self.member(i)?)?;
        if Some(o.thread.id) != self.entries[i].thread || o.context.thread != o.thread.id {
            return Err(Error::Stale);
        }
        if o.context.state == context::State::Vacant
            || o.context.state == context::State::Quarantined
            || o.thread.state == thread::State::Quarantined
        {
            return Err(Error::State);
        }
        if o.stack.layout != o.thread.layout
            || o.stack.guard_descriptor != 0
            || o.stack.low_canary != thread::LOW_CANARY
            || o.stack.high_canary != thread::HIGH_CANARY
        {
            return Err(Error::Guard);
        }
        if o.context.root == 0 || !o.context.root.is_multiple_of(4096) || o.context.root >> 48 != 0
        {
            return Err(Error::State);
        }
        Ok(o)
    }
    fn eligible(&self, i: usize) -> Result<bool, Error> {
        let o = self.observed(i)?;
        Ok(o.thread.state == thread::State::Ready && o.context.state == context::State::Ready)
    }
    fn audit_at(&self, e: Execution) -> Result<(), Error> {
        if self.len > N {
            return Err(Error::State);
        }
        for i in 0..self.len {
            let index = self.ready[i];
            if index >= N || self.active == Some(index) || self.ready[..i].contains(&index) {
                return Err(Error::Duplicate);
            }
            self.observed(index)?;
        }
        for (i, entry) in self.entries.iter().enumerate() {
            if entry.member.is_some() {
                let o = self.observed(i)?;
                if self.entries[..i]
                    .iter()
                    .any(|prior| prior.member.is_some() && prior.thread == entry.thread)
                {
                    return Err(Error::Duplicate);
                }
                if self.active == Some(i) {
                    if o.thread.state != thread::State::Running
                        || o.context.state != context::State::Active
                        || e.thread != Some(o.thread.id)
                        || e.root != o.context.root
                    {
                        return Err(Error::State);
                    }
                } else if o.thread.state == thread::State::Running
                    || o.context.state == context::State::Active
                {
                    return Err(Error::State);
                }
            }
        }
        if self.active.is_none() && e.thread.is_some() {
            return Err(Error::State);
        }
        Ok(())
    }
    pub fn audit(&self) -> Result<(), Error> {
        let e = self.execution()?;
        self.audit_at(e)
    }
    pub fn enqueue(
        &mut self,
        member: B::Member,
        class: Class,
    ) -> Result<Handle, (Error, B::Member)> {
        let result = (|| {
            self.audit()?;
            let o = self.backend.observe(&member)?;
            if self
                .entries
                .iter()
                .any(|s| s.member.is_some() && s.thread == Some(o.thread.id))
            {
                return Err(Error::Duplicate);
            }
            if o.thread.state != thread::State::Ready
                || o.context.state != context::State::Ready
                || o.context.thread != o.thread.id
            {
                return Err(Error::State);
            }
            if o.stack.layout != o.thread.layout
                || o.stack.guard_descriptor != 0
                || o.stack.low_canary != thread::LOW_CANARY
                || o.stack.high_canary != thread::HIGH_CANARY
            {
                return Err(Error::Guard);
            }
            if o.context.root == 0
                || !o.context.root.is_multiple_of(4096)
                || o.context.root >> 48 != 0
            {
                return Err(Error::State);
            }
            let i = self
                .entries
                .iter()
                .position(|s| s.member.is_none() && s.generation != u64::MAX)
                .ok_or(Error::Capacity)?;
            if self.len == N {
                return Err(Error::Capacity);
            }
            let generation = self.entries[i]
                .generation
                .checked_add(1)
                .ok_or(Error::Exhausted)?;
            Ok((i, generation, o.thread.id))
        })();
        match result {
            Ok((i, generation, id)) => {
                self.entries[i] = Entry {
                    generation,
                    member: Some(member),
                    thread: Some(id),
                    class,
                    remaining: 0,
                    slice: 0,
                    consumed_ticks: 0,
                    slices: 0,
                };
                self.ready[self.len] = i;
                self.len += 1;
                Ok(self.handle(i))
            }
            Err(e) => Err((e, member)),
        }
    }
    pub fn requeue(&mut self, h: Handle) -> Result<(), Error> {
        self.audit()?;
        let i = self.index(h)?;
        if self.active == Some(i) || self.ready[..self.len].contains(&i) {
            return Err(Error::Duplicate);
        }
        if !self.eligible(i)? {
            return Err(Error::State);
        }
        if self.len == N {
            return Err(Error::Capacity);
        }
        self.ready[self.len] = i;
        self.len += 1;
        Ok(())
    }
    pub fn request_priority(&mut self, h: Handle, request: u8) -> Result<(), Error> {
        let class = Class::request(request)?;
        self.audit()?;
        let i = self.index(h)?;
        if self.active == Some(i) {
            return Err(Error::Busy);
        }
        if !self.eligible(i)? {
            return Err(Error::State);
        }
        self.entries[i].class = class;
        Ok(())
    }
    pub fn select(&mut self, disposition: Disposition) -> Result<Selection<'_, B, N>, Error> {
        self.audit()?;
        if self.active.is_none() && disposition == Disposition::Sleeping {
            return Err(Error::State);
        }
        let mut selected = None;
        for position in 0..self.len {
            if self.eligible(self.ready[position])? {
                selected = Some(position);
                break;
            }
        }
        let position = selected.ok_or(Error::Empty)?;
        let target = self.ready[position];
        let epoch = self.epoch.checked_add(1).ok_or(Error::Exhausted)?;
        let slices = self.entries[target]
            .slices
            .checked_add(1)
            .ok_or(Error::Exhausted)?;
        let prepared = self.backend.prepare(
            self.active.and_then(|i| self.entries[i].member.as_ref()),
            self.entries[target].member.as_ref().ok_or(Error::Stale)?,
            epoch,
            disposition,
        )?;
        Ok(Selection {
            source: self.active,
            queue: self,
            target,
            position,
            epoch,
            slices,
            disposition,
            prepared: Some(prepared),
        })
    }
    /// Renew policy accounting for the sole runnable member. This changes no
    /// hardware context; the next owned timer delivery must bind the new slice.
    /// A live Ready neighbour requires selection and cannot be bypassed here.
    pub fn renew(&mut self) -> Result<Handle, Error> {
        self.audit()?;
        let i = self.active.ok_or(Error::State)?;
        if self.entries[i].remaining != 0 {
            return Err(Error::Busy);
        }
        for member in 0..N {
            if member != i && self.entries[member].member.is_some() && self.eligible(member)? {
                return Err(Error::Busy);
            }
        }
        let epoch = self.epoch.checked_add(1).ok_or(Error::Exhausted)?;
        let slices = self.entries[i]
            .slices
            .checked_add(1)
            .ok_or(Error::Exhausted)?;
        self.epoch = epoch;
        self.entries[i].slice = epoch;
        self.entries[i].remaining = self.entries[i].class.quantum();
        self.entries[i].slices = slices;
        Ok(self.handle(i))
    }
    #[allow(clippy::result_large_err)]
    pub fn charge(&mut self, tick: B::Tick) -> Result<Charge, (Error, B::Tick)> {
        let result = (|| {
            self.audit()?;
            let i = self.active.ok_or(Error::State)?;
            let entry = &self.entries[i];
            if entry.remaining == 0 {
                return Err(Error::Expired);
            }
            let o = self.backend.tick(&tick)?;
            if o.cpu != self.cpu
                || Some(o.thread) != entry.thread
                || o.slice != entry.slice
                || o.source == 0
                || self.tick_source.is_some_and(|source| source != o.source)
                || o.sequence == 0
                || o.sequence <= self.last_tick
                || o.counter < self.last_counter
            {
                return Err(Error::Tick);
            }
            let count = entry
                .consumed_ticks
                .checked_add(1)
                .ok_or(Error::Exhausted)?;
            let total = self.total_ticks.checked_add(1).ok_or(Error::Exhausted)?;
            Ok((i, o, count, total))
        })();
        match result {
            Ok((i, o, count, total)) => {
                let entry = &mut self.entries[i];
                entry.remaining -= 1;
                entry.consumed_ticks = count;
                self.total_ticks = total;
                self.tick_source = Some(o.source);
                self.last_tick = o.sequence;
                self.last_counter = o.counter;
                let remaining = entry.remaining;
                self.backend.consume_tick(tick);
                Ok(Charge {
                    remaining,
                    reschedule: remaining == 0,
                    sequence: o.sequence,
                })
            }
            Err(e) => Err((e, tick)),
        }
    }
    pub fn withdraw(&mut self, h: Handle) -> Result<B::Member, Error> {
        self.audit()?;
        let i = self.index(h)?;
        if self.active == Some(i) {
            return Err(Error::Busy);
        }
        let o = self.observed(i)?;
        if o.thread.state != thread::State::Dead && o.thread.state != thread::State::Sleeping {
            return Err(Error::State);
        }
        if let Some(p) = self.ready[..self.len].iter().position(|v| *v == i) {
            for j in p + 1..self.len {
                self.ready[j - 1] = self.ready[j]
            }
            self.len -= 1;
            self.ready[self.len] = 0;
        }
        let member = self.entries[i].member.take().ok_or(Error::Stale)?;
        let generation = self.entries[i].generation;
        self.entries[i] = Entry::empty();
        self.entries[i].generation = generation;
        Ok(member)
    }
    pub fn record(&self, h: Handle) -> Result<Record, Error> {
        let i = self.index(h)?;
        let e = &self.entries[i];
        Ok(Record {
            handle: h,
            thread: e.thread.ok_or(Error::Stale)?,
            class: e.class,
            queued: self.ready[..self.len].contains(&i),
            active: self.active == Some(i),
            remaining: e.remaining,
            slice: e.slice,
            consumed_ticks: e.consumed_ticks,
            slices: e.slices,
        })
    }
    pub fn snapshot(&self) -> Snapshot {
        Snapshot {
            members: self.entries.iter().filter(|s| s.member.is_some()).count(),
            queued: self.len,
            active: usize::from(self.active.is_some()),
            retired: self
                .entries
                .iter()
                .filter(|s| s.member.is_none() && s.generation == u64::MAX)
                .count(),
            epoch: self.epoch,
            consumed_ticks: self.total_ticks,
            faulted: self.faulted,
            retained_prepared: self.retained.is_some(),
        }
    }
    pub fn ready_order(&self) -> [Option<Handle>; N] {
        core::array::from_fn(|i| (i < self.len).then(|| self.handle(self.ready[i])))
    }
    pub fn with_backend_observation<R>(&self, f: impl FnOnce(&B) -> R) -> R {
        f(&self.backend)
    }
    pub fn with_retained_observation<R>(&self, f: impl FnOnce(Option<&B::Prepared>) -> R) -> R {
        f(self.retained.as_ref())
    }
    fn quarantine(&mut self, prepared: B::Prepared) {
        self.retained = Some(prepared);
        self.faulted = true;
    }
    #[allow(clippy::result_large_err)]
    pub fn into_backend(self) -> Result<B, (Error, Self)> {
        if self.faulted || self.entries.iter().any(|e| e.member.is_some()) {
            return Err((Error::Busy, self));
        }
        if let Err(e) = self.audit() {
            return Err((e, self));
        }
        Ok(self.backend)
    }
}
pub enum SelectionError<T> {
    Before(Error, T),
    Quarantined(Error),
}
pub struct Selection<'a, B: Backend, const N: usize> {
    queue: &'a mut Queue<B, N>,
    source: Option<usize>,
    target: usize,
    position: usize,
    epoch: u64,
    slices: u64,
    disposition: Disposition,
    prepared: Option<B::Prepared>,
}
impl<B: Backend, const N: usize> Selection<'_, B, N> {
    pub fn target(&self) -> Handle {
        self.queue.handle(self.target)
    }
    pub fn with_prepared_observation<R>(&self, f: impl FnOnce(&B::Prepared) -> R) -> R {
        f(self.prepared.as_ref().unwrap())
    }
    fn before(&self) -> Result<(), Error> {
        self.queue.audit()?;
        if self.queue.active != self.source
            || self.queue.ready.get(self.position) != Some(&self.target)
            || !self.queue.eligible(self.target)?
        {
            return Err(Error::State);
        }
        Ok(())
    }
    fn completed(&self) -> Result<(), Error> {
        let e = self.queue.execution()?;
        let target = self.queue.observed(self.target)?;
        if target.thread.state != thread::State::Running
            || target.context.state != context::State::Active
            || e.thread != Some(target.thread.id)
            || e.root != target.context.root
        {
            return Err(Error::State);
        }
        if let Some(i) = self.source {
            let source = self.queue.observed(i)?;
            let state = match self.disposition {
                Disposition::Ready => thread::State::Ready,
                Disposition::Sleeping => thread::State::Sleeping,
            };
            if source.thread.state != state || source.context.state != context::State::Ready {
                return Err(Error::State);
            }
        }
        Ok(())
    }
    #[allow(clippy::result_large_err)]
    pub fn complete(mut self) -> Result<Handle, SelectionError<Self>> {
        if let Err(e) = self.before() {
            return Err(SelectionError::Before(e, self));
        }
        let result = self.queue.backend.commit(
            self.prepared.as_mut().unwrap(),
            self.source
                .and_then(|i| self.queue.entries[i].member.as_ref()),
            self.queue.entries[self.target].member.as_ref().unwrap(),
            self.epoch,
            self.disposition,
        );
        if let Err(f) = result {
            return match f.phase {
                context::FailurePhase::BeforeSwitch => Err(SelectionError::Before(f.error, self)),
                context::FailurePhase::AfterSwitch => {
                    let p = self.prepared.take().unwrap();
                    self.queue.quarantine(p);
                    Err(SelectionError::Quarantined(f.error))
                }
            };
        }
        if let Err(e) = self.completed() {
            let p = self.prepared.take().unwrap();
            self.queue.quarantine(p);
            return Err(SelectionError::Quarantined(e));
        }
        let p = self.prepared.take().unwrap();
        if let Err((e, p)) = self.queue.backend.finish(p) {
            self.queue.quarantine(p);
            return Err(SelectionError::Quarantined(e));
        }
        for j in self.position + 1..self.queue.len {
            self.queue.ready[j - 1] = self.queue.ready[j]
        }
        self.queue.len -= 1;
        self.queue.ready[self.queue.len] = 0;
        if let Some(i) = self.source
            && self.disposition == Disposition::Ready
        {
            self.queue.ready[self.queue.len] = i;
            self.queue.len += 1;
        }
        self.queue.active = Some(self.target);
        self.queue.epoch = self.epoch;
        let entry = &mut self.queue.entries[self.target];
        entry.remaining = entry.class.quantum();
        entry.slice = self.epoch;
        entry.slices = self.slices;
        Ok(self.queue.handle(self.target))
    }
    pub fn abort(mut self) -> Result<(), Error> {
        let p = self.prepared.take().unwrap();
        if let Err((e, p)) = self.queue.backend.abort(p) {
            self.queue.quarantine(p);
            return Err(e);
        }
        Ok(())
    }
}
impl<B: Backend, const N: usize> Drop for Selection<'_, B, N> {
    fn drop(&mut self) {
        if let Some(p) = self.prepared.take()
            && let Err((_, p)) = self.queue.backend.abort(p)
        {
            self.queue.quarantine(p);
        }
    }
}

#[cfg(test)]
mod exhaustion_tests {
    use super::*;
    use crate::test_support::{backend, inventory, select, setup, tick};
    #[test]
    fn retired_slot_is_skipped_and_all_retired_preserve_same_member() {
        let (b, mut members, c) = backend(2);
        let mut q = Queue::<_, 4>::new(b).ok().unwrap();
        q.entries[0].generation = u64::MAX;
        let h = q.enqueue(members.remove(0), Class::Normal).ok().unwrap();
        assert_eq!(h.slot(), 1);
        q.entries[2].generation = u64::MAX;
        q.entries[3].generation = u64::MAX;
        let member = members.remove(0);
        let id = member.thread;
        let member = match q.enqueue(member, Class::Normal) {
            Err((Error::Capacity, m)) => m,
            _ => panic!(),
        };
        assert_eq!(member.thread, id);
        assert_eq!(q.snapshot().retired, 3);
        assert_eq!(inventory(&q), (2, 2, 8));
        assert_eq!(c.borrow().prepares, 0);
    }
    #[test]
    fn epoch_and_slice_counter_exhaustion_publish_no_prefix() {
        for counter in 0..2 {
            let (mut q, _, c) = setup(2);
            if counter == 0 {
                q.epoch = u64::MAX
            } else {
                q.entries[0].slices = u64::MAX
            }
            let before = q.snapshot();
            let order = q.ready_order();
            assert!(matches!(
                q.select(Disposition::Ready),
                Err(Error::Exhausted)
            ));
            assert_eq!(q.snapshot(), before);
            assert_eq!(q.ready_order(), order);
            assert_eq!(c.borrow().prepares, 0);
            assert_eq!(inventory(&q), (2, 2, 8));
        }
    }
    #[test]
    fn tick_counter_exhaustion_retains_exact_owned_tick() {
        for counter in 0..2 {
            let (mut q, h, c) = setup(2);
            select(&mut q, Disposition::Ready);
            if counter == 0 {
                q.total_ticks = u64::MAX
            } else {
                q.entries[0].consumed_ticks = u64::MAX
            }
            let t = tick(&q, h[0]);
            let sequence = t.nonce;
            let before = q.record(h[0]).unwrap();
            let t = match q.charge(t) {
                Err((Error::Exhausted, t)) => t,
                _ => panic!(),
            };
            assert_eq!(t.nonce, sequence);
            assert_eq!(q.record(h[0]).unwrap(), before);
            assert_eq!(c.borrow().consumed, 0);
            assert_eq!(inventory(&q), (2, 2, 8));
        }
    }
    #[test]
    fn private_foreign_generation_and_stale_slot_refuse_complete_inventory() {
        let (mut q, h, c) = setup(3);
        let before = q.snapshot();
        let forged = Handle {
            generation: h[0].generation + 1,
            ..h[0]
        };
        assert_eq!(q.request_priority(forged, 1), Err(Error::Stale));
        q.entries[2].thread = Some(q.record(h[1]).unwrap().thread);
        assert!(matches!(q.select(Disposition::Ready), Err(Error::Stale)));
        assert_eq!(q.snapshot(), before);
        assert_eq!(c.borrow().prepares, 0);
        assert_eq!(inventory(&q), (3, 3, 12));
    }
    #[test]
    fn damaged_ready_length_or_duplicate_never_prepares_selection() {
        for counter in 0..2 {
            let (mut q, _, c) = setup(3);
            if counter == 0 {
                q.len = 5
            } else {
                q.ready[2] = q.ready[0]
            }
            assert!(q.select(Disposition::Ready).is_err());
            assert_eq!(c.borrow().prepares, 0);
            assert_eq!(inventory(&q), (3, 3, 12));
        }
    }
    #[test]
    fn maximum_tick_sequence_is_consumed_once_without_successor_wrap() {
        let (mut q, h, c) = setup(2);
        select(&mut q, Disposition::Ready);
        q.last_tick = u64::MAX - 1;
        c.borrow_mut().sequence = u64::MAX - 1;
        let t = tick(&q, h[0]);
        let duplicate = crate::test_support::Tick {
            observation: t.observation,
            nonce: t.nonce,
        };
        assert_eq!(q.charge(t).ok().unwrap().sequence, u64::MAX);
        c.borrow_mut().tick_live = Some(u64::MAX);
        let before = q.record(h[0]).unwrap();
        assert!(matches!(q.charge(duplicate), Err((Error::Tick, _))));
        assert_eq!(q.record(h[0]).unwrap(), before);
        assert_eq!(c.borrow().consumed, 1);
        assert_eq!(inventory(&q), (2, 2, 8));
    }
}

// Appended only to the exact frozen Runqueue source in an isolated generated
// module. The original queue and every Member/Prepared remain exclusively
// owned here while assembly runs without a Rust queue or backend loan.
/// # Safety
/// Backend retains the actual Context AsyncBank and frozen Table. Departure
/// consumes that same staged context token with a short masked source-stack
/// loan. Arrival closes it only after actual target registers/SP/root were
/// captured by the assembly landing. No callback branches across a Rust loan.
pub unsafe trait SelectionHandoffBackend: Backend {
    /// BeforeSwitch preserves the complete Table/CPU/resource inventory and
    /// exact Prepared. Any possible Table or hardware change is AfterSwitch,
    /// including a failure before the physical assembly branch executes.
    fn depart(
        &mut self,
        prepared: &mut Self::Prepared,
        source: Option<&Self::Member>,
        target: &Self::Member,
        epoch: u64,
        disposition: Disposition,
        nonce: u64,
    ) -> Result<(), Failure>;
    fn arrive(
        &mut self,
        prepared: &mut Self::Prepared,
        target: &Self::Member,
        epoch: u64,
        nonce: u64,
    ) -> Result<(), Error>;
}

#[cfg(test)]
mod selection_handoff_exhaustion {
    use super::*;
    use crate::test_support as saved;
    unsafe impl SelectionHandoffBackend for saved::Backend {
        fn depart(
            &mut self,
            _: &mut Self::Prepared,
            _: Option<&Self::Member>,
            _: &Self::Member,
            _: u64,
            _: Disposition,
            _: u64,
        ) -> Result<(), Failure> {
            Err(Failure {
                error: Error::Backend,
                phase: context::FailurePhase::BeforeSwitch,
            })
        }
        fn arrive(
            &mut self,
            _: &mut Self::Prepared,
            _: &Self::Member,
            _: u64,
            _: u64,
        ) -> Result<(), Error> {
            Err(Error::Backend)
        }
    }
    #[test]
    fn sequence_exhaustion_never_stages_backend_or_changes_fifo() {
        let (queue, _, control) = saved::setup(2);
        let mut queue = AsyncQueue::new(queue).ok().unwrap();
        queue.sequence = u64::MAX;
        let before = queue.queue().snapshot();
        let order = queue.queue().ready_order();
        assert!(matches!(
            queue.select(Disposition::Ready),
            Err(Error::Exhausted)
        ));
        assert_eq!(queue.queue().snapshot(), before);
        assert_eq!(queue.queue().ready_order(), order);
        assert_eq!(control.borrow().prepares, 0);
        assert!(queue.quiescent().is_ok());
    }
    #[test]
    fn foreign_nonce_and_changed_target_generation_keep_exact_prepared() {
        for field in 0..3 {
            let (queue, _, control) = saved::setup(2);
            let mut queue = AsyncQueue::new(queue).ok().unwrap();
            let mut token = queue.select(Disposition::Ready).unwrap();
            let prepared = queue
                .queue()
                .with_retained_observation(|p| p.unwrap().nonce);
            match field {
                0 => token.queue += 1,
                1 => token.nonce += 1,
                _ => queue.queue.entries[queue.pending.unwrap().target].generation += 1,
            }
            assert!(matches!(
                queue.depart(token),
                Err(HandoffSelectionError::Before(Error::Stale, _))
            ));
            assert_eq!(
                queue
                    .queue()
                    .with_retained_observation(|p| p.unwrap().nonce),
                prepared
            );
            assert_eq!(queue.stage(), Some(SelectionStage::Staged));
            assert!(queue.quiescent().is_err());
            assert_eq!(control.borrow().commits, 0);
            assert_eq!(saved::inventory(queue.queue()), (2, 2, 8));
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SelectionStage {
    Staged,
    Departed,
}
/// Nonclone lookup authority. Forgetting this token leaves all actual members
/// and the staged Prepared retained, and blocks every ordinary queue mutation.
pub struct OwnedSelection {
    queue: u64,
    nonce: u64,
}
impl OwnedSelection {
    pub fn nonce(&self) -> u64 {
        self.nonce
    }
}
pub enum HandoffSelectionError {
    Before(Error, OwnedSelection),
    Quarantined(Error),
}
#[derive(Clone, Copy)]
struct PendingSelection {
    nonce: u64,
    stage: SelectionStage,
    source: Option<usize>,
    source_generation: Option<u64>,
    target: usize,
    target_generation: u64,
    position: usize,
    old_epoch: u64,
    epoch: u64,
    slices: u64,
    disposition: Disposition,
}
pub struct AsyncQueue<B: SelectionHandoffBackend, const N: usize> {
    queue: Queue<B, N>,
    sequence: u64,
    pending: Option<PendingSelection>,
    quarantined: bool,
}
impl<B: SelectionHandoffBackend, const N: usize> AsyncQueue<B, N> {
    #[allow(clippy::result_large_err)]
    pub fn new(queue: Queue<B, N>) -> Result<Self, (Error, Queue<B, N>)> {
        if queue.faulted || queue.retained.is_some() {
            return Err((Error::Quarantined, queue));
        }
        if let Err(error) = queue.audit() {
            return Err((error, queue));
        }
        Ok(Self {
            queue,
            sequence: 0,
            pending: None,
            quarantined: false,
        })
    }
    pub fn queue(&self) -> &Queue<B, N> {
        &self.queue
    }
    pub fn stage(&self) -> Option<SelectionStage> {
        self.pending.map(|p| p.stage)
    }
    pub fn quarantined(&self) -> bool {
        self.quarantined
    }
    pub fn quiescent(&mut self) -> Result<&mut Queue<B, N>, Error> {
        if self.pending.is_some() || self.quarantined || self.queue.faulted {
            return Err(Error::Busy);
        }
        Ok(&mut self.queue)
    }
    pub fn select(&mut self, disposition: Disposition) -> Result<OwnedSelection, Error> {
        if self.pending.is_some() || self.quarantined || self.queue.faulted {
            return Err(Error::Busy);
        }
        let nonce = self.sequence.checked_add(1).ok_or(Error::Exhausted)?;
        let mut selection = self.queue.select(disposition)?;
        let pending = PendingSelection {
            nonce,
            stage: SelectionStage::Staged,
            source: selection.source,
            source_generation: selection
                .source
                .map(|i| selection.queue.entries[i].generation),
            target: selection.target,
            target_generation: selection.queue.entries[selection.target].generation,
            position: selection.position,
            old_epoch: selection.queue.epoch,
            epoch: selection.epoch,
            slices: selection.slices,
            disposition: selection.disposition,
        };
        let token = OwnedSelection {
            queue: selection.queue.identity,
            nonce,
        };
        selection.queue.retained = selection.prepared.take();
        selection.queue.faulted = true;
        drop(selection);
        self.sequence = nonce;
        self.pending = Some(pending);
        Ok(token)
    }
    fn validate(&self, token: &OwnedSelection) -> Result<PendingSelection, Error> {
        let pending = self.pending.ok_or(Error::State)?;
        if self.quarantined || !self.queue.faulted || self.queue.retained.is_none() {
            return Err(Error::Quarantined);
        }
        if token.queue != self.queue.identity
            || token.nonce != pending.nonce
            || self.queue.epoch != pending.old_epoch
            || self.queue.active != pending.source
            || self.queue.entries[pending.target].generation != pending.target_generation
            || pending.source.is_some_and(|i| {
                Some(self.queue.entries[i].generation) != pending.source_generation
            })
        {
            return Err(Error::Stale);
        }
        Ok(pending)
    }
    pub fn target(&self, token: &OwnedSelection) -> Result<Handle, Error> {
        Ok(self.queue.handle(self.validate(token)?.target))
    }
    fn checked(&mut self, pending: PendingSelection, completed: bool) -> Result<(), Error> {
        self.queue.faulted = false;
        let result = {
            let selection = Selection {
                queue: &mut self.queue,
                source: pending.source,
                target: pending.target,
                position: pending.position,
                epoch: pending.epoch,
                slices: pending.slices,
                disposition: pending.disposition,
                prepared: None,
            };
            if completed {
                selection.completed()
            } else {
                selection.before()
            }
        };
        self.queue.faulted = true;
        result
    }
    fn quarantine(&mut self) {
        self.queue.faulted = true;
        self.quarantined = true;
    }
    #[allow(clippy::result_large_err)]
    pub fn depart(
        &mut self,
        token: OwnedSelection,
    ) -> Result<OwnedSelection, HandoffSelectionError> {
        let pending = match self.validate(&token) {
            Ok(pending) => pending,
            Err(error) => return Err(HandoffSelectionError::Before(error, token)),
        };
        if pending.stage != SelectionStage::Staged {
            return Err(HandoffSelectionError::Before(Error::State, token));
        }
        if let Err(error) = self.checked(pending, false) {
            return Err(HandoffSelectionError::Before(error, token));
        }
        let result = self.queue.backend.depart(
            self.queue.retained.as_mut().unwrap(),
            pending
                .source
                .and_then(|i| self.queue.entries[i].member.as_ref()),
            self.queue.entries[pending.target].member.as_ref().unwrap(),
            pending.epoch,
            pending.disposition,
            pending.nonce,
        );
        if let Err(failure) = result {
            return match failure.phase {
                context::FailurePhase::BeforeSwitch => {
                    Err(HandoffSelectionError::Before(failure.error, token))
                }
                context::FailurePhase::AfterSwitch => {
                    self.quarantine();
                    Err(HandoffSelectionError::Quarantined(failure.error))
                }
            };
        }
        self.pending.as_mut().unwrap().stage = SelectionStage::Departed;
        Ok(token)
    }
    pub fn landed(&mut self, token: OwnedSelection) -> Result<Handle, HandoffSelectionError> {
        let pending = match self.validate(&token) {
            Ok(pending) => pending,
            Err(error) => return Err(HandoffSelectionError::Before(error, token)),
        };
        if pending.stage != SelectionStage::Departed {
            return Err(HandoffSelectionError::Before(Error::State, token));
        }
        let result = self.queue.backend.arrive(
            self.queue.retained.as_mut().unwrap(),
            self.queue.entries[pending.target].member.as_ref().unwrap(),
            pending.epoch,
            pending.nonce,
        );
        if let Err(error) = result {
            self.quarantine();
            return Err(HandoffSelectionError::Quarantined(error));
        }
        if let Err(error) = self.checked(pending, true) {
            self.quarantine();
            return Err(HandoffSelectionError::Quarantined(error));
        }
        let prepared = self.queue.retained.take().unwrap();
        if let Err((error, prepared)) = self.queue.backend.finish(prepared) {
            self.queue.retained = Some(prepared);
            self.quarantine();
            return Err(HandoffSelectionError::Quarantined(error));
        }
        for position in pending.position + 1..self.queue.len {
            self.queue.ready[position - 1] = self.queue.ready[position];
        }
        self.queue.len -= 1;
        self.queue.ready[self.queue.len] = 0;
        if let Some(source) = pending.source
            && pending.disposition == Disposition::Ready
        {
            self.queue.ready[self.queue.len] = source;
            self.queue.len += 1;
        }
        self.queue.active = Some(pending.target);
        self.queue.epoch = pending.epoch;
        let entry = &mut self.queue.entries[pending.target];
        entry.remaining = entry.class.quantum();
        entry.slice = pending.epoch;
        entry.slices = pending.slices;
        self.pending = None;
        self.queue.faulted = false;
        Ok(self.queue.handle(pending.target))
    }
    /// Abort can close only a staged source-side transaction. After departure,
    /// the actual Prepared and all members remain retained for physical landing.
    pub fn abort(&mut self, token: OwnedSelection) -> Result<(), HandoffSelectionError> {
        let pending = match self.validate(&token) {
            Ok(pending) => pending,
            Err(error) => return Err(HandoffSelectionError::Before(error, token)),
        };
        if pending.stage != SelectionStage::Staged {
            return Err(HandoffSelectionError::Before(Error::State, token));
        }
        let prepared = self.queue.retained.take().unwrap();
        if let Err((error, prepared)) = self.queue.backend.abort(prepared) {
            self.queue.retained = Some(prepared);
            self.quarantine();
            return Err(HandoffSelectionError::Quarantined(error));
        }
        self.pending = None;
        self.queue.faulted = false;
        Ok(())
    }
}

// NEW extension after the exact frozen Queue and asynchronous selection source.
// A return owns its actual backend Prepared until physical bootstrap landing.
/// # Safety
/// The backend owns the same live CompletionBank used by ordinary selections.
/// ReturnPrepared retains its real nonclone OwnedReturn and all underlying
/// Context/Table/stack/root owners. prepare_exit has no visible effect on
/// failure. BeforeSwitch keeps the actual CPU/Table/resource inventory intact;
/// every possible change is AfterSwitch. depart_exit delegates real Completion
/// departure; returned_exit delegates real Completion landing, after assembly
/// saved the complete bootstrap frame/root/stack observations. returned_exit
/// MUST NOT consume the actual Completion Prepared; finish_exit calls the real
/// Completion.return_landed as the LAST fallible operation. finish/abort
/// failures return the actual Prepared. No method branches across a Rust loan,
/// allocates, releases a membership/backing owner or unmasks IRQs. Execution
/// and observations bind actual owning CPU/I1/nonIRQ before owner access.
pub unsafe trait QueueCompletionBackend: SelectionHandoffBackend {
    type ReturnPrepared;
    fn prepare_exit(
        &mut self,
        source: &Self::Member,
        capture: context::Capture,
    ) -> Result<Self::ReturnPrepared, Error>;
    fn depart_exit(
        &mut self,
        prepared: &mut Self::ReturnPrepared,
        source: &Self::Member,
        nonce: u64,
    ) -> Result<(), Failure>;
    fn returned_exit(
        &mut self,
        prepared: &mut Self::ReturnPrepared,
        source: &Self::Member,
        nonce: u64,
    ) -> Result<(), Error>;
    fn finish_exit(
        &mut self,
        prepared: Self::ReturnPrepared,
    ) -> Result<(), (Error, Self::ReturnPrepared)>;
    fn abort_exit(
        &mut self,
        prepared: Self::ReturnPrepared,
    ) -> Result<(), (Error, Self::ReturnPrepared)>;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum QueueReturnStage {
    Staged,
    Departed,
}
/// Private nonclone lookup. Forgetting it retains the actual Prepared and all
/// memberships and blocks every ordinary mutation; it cannot release backing.
pub struct OwnedQueueReturn {
    queue: u64,
    nonce: u64,
}
impl OwnedQueueReturn {
    pub fn nonce(&self) -> u64 {
        self.nonce
    }
}
pub enum QueueReturnFailure {
    Before(Error, OwnedQueueReturn),
    Quarantined(Error),
}
struct PendingQueueReturn<P> {
    nonce: u64,
    source: usize,
    generation: u64,
    epoch: u64,
    next_epoch: u64,
    root: u64,
    cpu: u32,
    prepared: P,
    stage: QueueReturnStage,
}
pub struct ReturnQueue<B: QueueCompletionBackend, const N: usize> {
    queue: AsyncQueue<B, N>,
    sequence: u64,
    pending: Option<PendingQueueReturn<B::ReturnPrepared>>,
    quarantined: bool,
}
impl<B: QueueCompletionBackend, const N: usize> ReturnQueue<B, N> {
    #[allow(clippy::result_large_err)]
    pub fn new(queue: AsyncQueue<B, N>) -> Result<Self, (Error, AsyncQueue<B, N>)> {
        if queue.stage().is_some() || queue.quarantined() {
            return Err((Error::Busy, queue));
        }
        if let Err(error) = queue.queue.audit() {
            return Err((error, queue));
        }
        Ok(Self {
            queue,
            sequence: 0,
            pending: None,
            quarantined: false,
        })
    }
    pub fn queue(&self) -> &Queue<B, N> {
        self.queue.queue()
    }
    pub fn return_stage(&self) -> Option<QueueReturnStage> {
        self.pending.as_ref().map(|p| p.stage)
    }
    pub fn quarantined(&self) -> bool {
        self.quarantined || self.queue.quarantined()
    }
    pub fn with_return_observation<R>(&self, f: impl FnOnce(Option<&B::ReturnPrepared>) -> R) -> R {
        f(self.pending.as_ref().map(|p| &p.prepared))
    }
    fn ordinary(&self) -> Result<(), Error> {
        if self.pending.is_some() || self.quarantined {
            return Err(Error::Busy);
        }
        Ok(())
    }
    pub fn quiescent(&mut self) -> Result<&mut Queue<B, N>, Error> {
        self.ordinary()?;
        self.queue.quiescent()
    }
    pub fn select(&mut self, d: Disposition) -> Result<OwnedSelection, Error> {
        self.ordinary()?;
        self.queue.select(d)
    }
    pub fn target(&self, t: &OwnedSelection) -> Result<Handle, Error> {
        self.ordinary()?;
        self.queue.target(t)
    }
    pub fn depart(&mut self, t: OwnedSelection) -> Result<OwnedSelection, HandoffSelectionError> {
        if let Err(e) = self.ordinary() {
            return Err(HandoffSelectionError::Before(e, t));
        }
        self.queue.depart(t)
    }
    pub fn landed(&mut self, t: OwnedSelection) -> Result<Handle, HandoffSelectionError> {
        if let Err(e) = self.ordinary() {
            return Err(HandoffSelectionError::Before(e, t));
        }
        self.queue.landed(t)
    }
    pub fn abort(&mut self, t: OwnedSelection) -> Result<(), HandoffSelectionError> {
        if let Err(e) = self.ordinary() {
            return Err(HandoffSelectionError::Before(e, t));
        }
        self.queue.abort(t)
    }
    pub fn prepare_return(&mut self, capture: context::Capture) -> Result<OwnedQueueReturn, Error> {
        self.ordinary()?;
        if self.queue.pending.is_some() || self.queue.quarantined {
            return Err(Error::Busy);
        }
        let queue = &mut self.queue.queue;
        queue.audit()?;
        let source = queue.active.ok_or(Error::State)?;
        let execution = queue.execution()?;
        let observation = queue.observed(source)?;
        if execution.thread != Some(observation.thread.id)
            || execution.root != observation.context.root
        {
            return Err(Error::State);
        }
        let nonce = self.sequence.checked_add(1).ok_or(Error::Exhausted)?;
        let next_epoch = queue.epoch.checked_add(1).ok_or(Error::Exhausted)?;
        let prepared = queue.backend.prepare_exit(
            queue.entries[source].member.as_ref().ok_or(Error::Stale)?,
            capture,
        )?;
        self.pending = Some(PendingQueueReturn {
            nonce,
            source,
            generation: queue.entries[source].generation,
            epoch: queue.epoch,
            next_epoch,
            root: execution.root,
            cpu: execution.cpu,
            prepared,
            stage: QueueReturnStage::Staged,
        });
        self.sequence = nonce;
        queue.faulted = true;
        Ok(OwnedQueueReturn {
            queue: queue.identity,
            nonce,
        })
    }
    fn validate(
        &self,
        t: &OwnedQueueReturn,
    ) -> Result<&PendingQueueReturn<B::ReturnPrepared>, Error> {
        let p = self.pending.as_ref().ok_or(Error::State)?;
        let q = &self.queue.queue;
        if self.quarantined || self.queue.quarantined || !q.faulted {
            return Err(Error::Quarantined);
        }
        if t.queue != q.identity
            || t.nonce != p.nonce
            || q.epoch != p.epoch
            || q.active != Some(p.source)
            || q.entries[p.source].generation != p.generation
            || q.entries[p.source].member.is_none()
        {
            return Err(Error::Stale);
        }
        let e = q.backend.execution();
        if e.cpu != p.cpu || !e.masked || e.in_irq {
            return Err(Error::Context);
        }
        Ok(p)
    }
    fn quarantine(&mut self) {
        self.quarantined = true;
        self.queue.queue.faulted = true;
    }
    pub fn depart_return(
        &mut self,
        t: OwnedQueueReturn,
    ) -> Result<OwnedQueueReturn, QueueReturnFailure> {
        let p = match self.validate(&t) {
            Ok(p) => p,
            Err(e) => return Err(QueueReturnFailure::Before(e, t)),
        };
        if p.stage != QueueReturnStage::Staged {
            return Err(QueueReturnFailure::Before(Error::State, t));
        }
        let source = p.source;
        let root = p.root;
        // Validate the complete old graph without publishing any new metadata.
        self.queue.queue.faulted = false;
        let before = self.queue.queue.audit();
        self.queue.queue.faulted = true;
        if let Err(e) = before {
            return Err(QueueReturnFailure::Before(e, t));
        }
        if self.queue.queue.backend.execution().root != root {
            return Err(QueueReturnFailure::Before(Error::State, t));
        }
        let p = self.pending.as_mut().unwrap();
        let q = &mut self.queue.queue;
        let result = q.backend.depart_exit(
            &mut p.prepared,
            q.entries[source].member.as_ref().unwrap(),
            p.nonce,
        );
        match result {
            Ok(()) => {
                p.stage = QueueReturnStage::Departed;
                Ok(t)
            }
            Err(Failure {
                error,
                phase: context::FailurePhase::BeforeSwitch,
            }) => Err(QueueReturnFailure::Before(error, t)),
            Err(Failure {
                error,
                phase: context::FailurePhase::AfterSwitch,
            }) => {
                self.quarantine();
                Err(QueueReturnFailure::Quarantined(error))
            }
        }
    }
    pub fn return_landed(&mut self, t: OwnedQueueReturn) -> Result<(), QueueReturnFailure> {
        let p = match self.validate(&t) {
            Ok(p) => p,
            Err(e) => return Err(QueueReturnFailure::Before(e, t)),
        };
        if p.stage != QueueReturnStage::Departed {
            return Err(QueueReturnFailure::Before(Error::State, t));
        }
        let source = p.source;
        let p = self.pending.as_mut().unwrap();
        let q = &mut self.queue.queue;
        if let Err(error) = q.backend.returned_exit(
            &mut p.prepared,
            q.entries[source].member.as_ref().unwrap(),
            p.nonce,
        ) {
            self.quarantine();
            return Err(QueueReturnFailure::Quarantined(error));
        }
        let checked = (|| {
            let e = q.backend.execution();
            let o = q.observed(source)?;
            if e.cpu != p.cpu
                || !e.masked
                || e.in_irq
                || e.thread.is_some()
                || o.thread.state != thread::State::Dead
                || o.context.state != context::State::Active
                || q.len > N
                || q.ready[..q.len].contains(&source)
            {
                return Err(Error::State);
            }
            // The immutable backend proved physical bootstrap landing while
            // real Completion Prepared is STILL retained. Inspect every other
            // private owner and FIFO edge before that Prepared is consumed.
            for position in 0..q.len {
                let index = q.ready[position];
                if index >= N || index == source || q.ready[..position].contains(&index) {
                    return Err(Error::Duplicate);
                }
                let ready = q.observed(index)?;
                if ready.thread.state != thread::State::Ready
                    || ready.context.state != context::State::Ready
                {
                    return Err(Error::State);
                }
            }
            for (index, entry) in q.entries.iter().enumerate() {
                if entry.member.is_none() {
                    continue;
                }
                let observed = q.observed(index)?;
                if q.entries[..index]
                    .iter()
                    .any(|prior| prior.member.is_some() && prior.thread == entry.thread)
                {
                    return Err(Error::Duplicate);
                }
                if index != source
                    && (observed.thread.state == thread::State::Running
                        || observed.context.state == context::State::Active)
                {
                    return Err(Error::State);
                }
            }
            Ok(())
        })();
        if let Err(error) = checked {
            self.quarantine();
            return Err(QueueReturnFailure::Quarantined(error));
        }
        let p = self.pending.take().unwrap();
        if let Err((error, prepared)) = self.queue.queue.backend.finish_exit(p.prepared) {
            self.pending = Some(PendingQueueReturn { prepared, ..p });
            self.quarantine();
            return Err(QueueReturnFailure::Quarantined(error));
        }
        self.queue.queue.active = None;
        self.queue.queue.epoch = p.next_epoch;
        self.queue.queue.faulted = false;
        Ok(())
    }
    pub fn abort_return(&mut self, t: OwnedQueueReturn) -> Result<(), QueueReturnFailure> {
        let p = match self.validate(&t) {
            Ok(p) => p,
            Err(e) => return Err(QueueReturnFailure::Before(e, t)),
        };
        if p.stage != QueueReturnStage::Staged {
            return Err(QueueReturnFailure::Before(Error::State, t));
        }
        let p = self.pending.take().unwrap();
        if let Err((error, prepared)) = self.queue.queue.backend.abort_exit(p.prepared) {
            self.pending = Some(PendingQueueReturn { prepared, ..p });
            self.quarantine();
            return Err(QueueReturnFailure::Quarantined(error));
        }
        self.queue.queue.faulted = false;
        Ok(())
    }
    #[allow(clippy::result_large_err)]
    pub fn into_async(self) -> Result<AsyncQueue<B, N>, (Error, Self)> {
        if self.pending.is_some() || self.quarantined {
            Err((Error::Busy, self))
        } else {
            Ok(self.queue)
        }
    }
}
#[cfg(test)]
mod queue_return_tests {
    use super::*;
    use crate::return_support::*;
    #[test]
    fn foreign_or_stale_lookup_preserves_exact_actual_prepared() {
        for kind in 0..3 {
            let (mut q, _) = returned_setup();
            step(&mut q, false);
            let c = source_capture(&q);
            let mut t = q.prepare_return(c).unwrap();
            let before = q.queue().snapshot();
            let p = prepared_return(&q);
            match kind {
                0 => t.queue += 1,
                1 => t.nonce += 1,
                _ => q.queue.queue.entries[q.pending.as_ref().unwrap().source].generation += 1,
            };
            assert!(matches!(
                q.depart_return(t),
                Err(QueueReturnFailure::Before(Error::Stale, _))
            ));
            assert_eq!(prepared_return(&q), p);
            assert_eq!(q.queue().snapshot(), before);
            assert!(underlying_return_retained(&q));
            assert_eq!(resources(&q), (3, 3, 12));
        }
    }
    #[test]
    fn return_sequence_or_epoch_exhaustion_is_before_backend_preparation() {
        for field in 0..2 {
            let (mut q, _) = returned_setup();
            step(&mut q, false);
            let c = source_capture(&q);
            if field == 0 {
                q.sequence = u64::MAX;
            } else {
                q.queue.queue.epoch = u64::MAX;
            }
            let before = q.queue().snapshot();
            let prepared = hardware(&q).borrow().prepares;
            assert!(matches!(q.prepare_return(c), Err(Error::Exhausted)));
            assert_eq!(q.queue().snapshot(), before);
            assert_eq!(hardware(&q).borrow().prepares, prepared);
            assert!(!underlying_return_retained(&q));
            assert_eq!(resources(&q), (3, 3, 12));
        }
    }
    #[test]
    fn departed_return_cannot_abort_or_land_twice_without_real_landing() {
        let (mut q, _) = returned_setup();
        step(&mut q, false);
        let c = source_capture(&q);
        let t = q.prepare_return(c).unwrap();
        let t = q.depart_return(t).ok().unwrap();
        let t = match q.abort_return(t) {
            Err(QueueReturnFailure::Before(Error::State, t)) => t,
            _ => panic!("departed abort"),
        };
        assert!(underlying_return_retained(&q));
        boot_landing(&q);
        assert!(q.return_landed(t).is_ok());
        let fake = OwnedQueueReturn {
            queue: q.queue.queue.identity,
            nonce: q.sequence,
        };
        assert!(matches!(
            q.return_landed(fake),
            Err(QueueReturnFailure::Before(Error::State, _))
        ));
        assert_eq!(q.queue().snapshot().active, 0);
        assert_eq!(resources(&q), (3, 3, 12));
    }
    #[test]
    fn proposed_fifo_duplicate_keeps_real_completion_prepared_until_quarantine() {
        let (mut q, _) = returned_setup();
        step(&mut q, false);
        let c = source_capture(&q);
        let t = q.prepare_return(c).unwrap();
        let t = q.depart_return(t).ok().unwrap();
        boot_landing(&q);
        q.queue.queue.ready[1] = q.queue.queue.ready[0];
        let before = q.queue().snapshot();
        assert!(matches!(
            q.return_landed(t),
            Err(QueueReturnFailure::Quarantined(Error::Duplicate))
        ));
        assert_eq!(q.queue().snapshot(), before);
        assert!(underlying_return_retained(&q));
        assert_eq!(resources(&q), (3, 3, 12));
    }
    #[test]
    fn readonly_layout_words_bind_actual_private_member_and_prepared_addresses() {
        let (mut q, _) = returned_setup();
        step(&mut q, false);
        let c = source_capture(&q);
        let t = q.prepare_return(c).unwrap();
        let w = q.os_k1060_owner_words();
        let layout = os_k1060_return_layout::<crate::return_support::Device, 4>();
        assert_eq!(
            w[8],
            q.pending.as_ref().unwrap() as *const PendingQueueReturn<_> as u64
        );
        assert_eq!(
            w[13],
            &q.pending.as_ref().unwrap().prepared as *const _ as u64
        );
        assert_eq!(layout[39], 4);
        for (i, row) in q.os_k1060_member_words().iter().enumerate().take(3) {
            assert_eq!(
                row[2],
                q.queue.queue.entries[i].member.as_ref().unwrap() as *const _ as u64
            );
            assert_ne!(row[3], 0);
            assert_eq!(row[4], i as u64);
        }
        assert!(q.abort_return(t).is_ok());
    }
}

// NEW immutable layout/ownership observation appended only to generated Queue.
// No Member, Prepared, backend loan or release authority escapes.
impl<B: QueueCompletionBackend, const N: usize> ReturnQueue<B, N> {
    pub fn os_k1060_owner_words(&self) -> [u64; 16] {
        let q = &self.queue.queue;
        let p = self.pending.as_ref();
        [
            q.identity,
            q.cpu as u64,
            q.epoch,
            q.active.map_or(0, |i| i as u64 + 1),
            u64::from(q.faulted),
            q.total_ticks,
            q.len as u64,
            self.sequence,
            p.map_or(0, |p| {
                p as *const PendingQueueReturn<B::ReturnPrepared> as u64
            }),
            p.map_or(0, |p| p.nonce),
            p.map_or(0, |p| {
                if p.stage == QueueReturnStage::Staged {
                    1
                } else {
                    2
                }
            }),
            p.map_or(0, |p| p.source as u64),
            p.map_or(0, |p| p.generation),
            p.map_or(0, |p| &p.prepared as *const B::ReturnPrepared as u64),
            core::mem::size_of::<Self>() as u64,
            u64::from(self.quarantined),
        ]
    }
    pub fn os_k1060_member_words(&self) -> [[u64; 16]; N] {
        let q = &self.queue.queue;
        core::array::from_fn(|i| {
            let e = &q.entries[i];
            [
                i as u64,
                e.generation,
                e.member
                    .as_ref()
                    .map_or(0, |m| m as *const B::Member as u64),
                e.thread.map_or(0, |t| t.table()),
                e.thread.map_or(0, |t| t.slot() as u64),
                e.thread.map_or(0, |t| t.generation()),
                if e.class == Class::Normal { 0 } else { 1 },
                e.remaining as u64,
                e.slice,
                e.consumed_ticks,
                e.slices,
                u64::from(q.ready[..q.len].contains(&i)),
                u64::from(q.active == Some(i)),
                &q.backend as *const B as u64,
                core::mem::size_of::<B::Member>() as u64,
                core::mem::size_of::<B>() as u64,
            ]
        })
    }
}
pub const fn os_k1060_return_layout<B: QueueCompletionBackend, const N: usize>() -> [u64; 40] {
    use core::mem::{offset_of, size_of};
    [
        size_of::<ReturnQueue<B, N>>() as u64,
        offset_of!(ReturnQueue<B,N>,queue) as u64,
        offset_of!(ReturnQueue<B,N>,sequence) as u64,
        offset_of!(ReturnQueue<B,N>,pending) as u64,
        offset_of!(ReturnQueue<B,N>,quarantined) as u64,
        size_of::<AsyncQueue<B, N>>() as u64,
        offset_of!(AsyncQueue<B,N>,queue) as u64,
        offset_of!(AsyncQueue<B,N>,sequence) as u64,
        offset_of!(AsyncQueue<B,N>,pending) as u64,
        offset_of!(AsyncQueue<B,N>,quarantined) as u64,
        size_of::<Queue<B, N>>() as u64,
        offset_of!(Queue<B,N>,backend) as u64,
        offset_of!(Queue<B,N>,identity) as u64,
        offset_of!(Queue<B,N>,cpu) as u64,
        offset_of!(Queue<B,N>,entries) as u64,
        offset_of!(Queue<B,N>,ready) as u64,
        offset_of!(Queue<B,N>,len) as u64,
        offset_of!(Queue<B,N>,active) as u64,
        offset_of!(Queue<B,N>,epoch) as u64,
        offset_of!(Queue<B,N>,total_ticks) as u64,
        offset_of!(Queue<B,N>,faulted) as u64,
        size_of::<Entry<B::Member>>() as u64,
        offset_of!(Entry<B::Member>, generation) as u64,
        offset_of!(Entry<B::Member>, member) as u64,
        offset_of!(Entry<B::Member>, thread) as u64,
        offset_of!(Entry<B::Member>, class) as u64,
        offset_of!(Entry<B::Member>, remaining) as u64,
        offset_of!(Entry<B::Member>, slice) as u64,
        offset_of!(Entry<B::Member>, consumed_ticks) as u64,
        offset_of!(Entry<B::Member>, slices) as u64,
        size_of::<PendingQueueReturn<B::ReturnPrepared>>() as u64,
        offset_of!(PendingQueueReturn<B::ReturnPrepared>, nonce) as u64,
        offset_of!(PendingQueueReturn<B::ReturnPrepared>, source) as u64,
        offset_of!(PendingQueueReturn<B::ReturnPrepared>, generation) as u64,
        offset_of!(PendingQueueReturn<B::ReturnPrepared>, epoch) as u64,
        offset_of!(PendingQueueReturn<B::ReturnPrepared>, prepared) as u64,
        offset_of!(PendingQueueReturn<B::ReturnPrepared>, stage) as u64,
        size_of::<B::ReturnPrepared>() as u64,
        size_of::<B::Member>() as u64,
        N as u64,
    ]
}

// NEW narrow operation after generated frozen Queue ownership sources.
/// # Safety
/// The backend retires the same nonclone member/binding held by its actual
/// Context/Table owners. It retains and returns the real member on any error.
/// No callback branches, unmasks IRQ, or releases a queued/running member.
pub unsafe trait QueueRetirementBackend: Backend {
    // Ok returns only the retired logical member after actual backing release;
    // it carries no physical Binding. Keeping it until the final execution
    // check permits honest quarantine without inventing a consumed owner.
    fn retire_withdrawn(
        &mut self,
        member: Self::Member,
    ) -> Result<Self::Member, (Error, Self::Member)>;
}
impl<B: QueueRetirementBackend, const N: usize> Queue<B, N> {
    pub fn os_k1060_retire_withdrawn(
        &mut self,
        member: B::Member,
    ) -> Result<(), (Error, B::Member)> {
        let execution = match self.execution() {
            Ok(e) => e,
            Err(e) => return Err((e, member)),
        };
        if execution.thread.is_some() || self.active.is_some() || self.retained.is_some() {
            return Err((Error::Busy, member));
        }
        if let Err(error) = self.audit_at(execution) {
            return Err((error, member));
        }
        let observation = match self.backend.observe(&member) {
            Ok(o) => o,
            Err(error) => return Err((error, member)),
        };
        if self
            .entries
            .iter()
            .any(|e| e.member.is_some() && e.thread == Some(observation.thread.id))
        {
            return Err((Error::Duplicate, member));
        }
        match self.backend.retire_withdrawn(member) {
            Ok(retired) => {
                if self.backend.execution() != execution {
                    self.faulted = true;
                    return Err((Error::Quarantined, retired));
                }
                Ok(())
            }
            Err((error, actual)) => {
                if error == Error::Quarantined {
                    self.faulted = true;
                }
                Err((error, actual))
            }
        }
    }
}

// NEW exact Wait custody and wake publication after the frozen Queue bytes.
/// # Safety
/// Delegates only the retained member's actual Table Wait. Before preserves
/// owners; After quarantines. No callback switches or frees backing.
pub unsafe trait WaitQueueServices: Backend {
    fn wait_physical_landing(
        &self,
        prepared: &Self::Prepared,
        member: &Self::Member,
    ) -> Result<crate::wait_core::Execution, Error>;
    fn wait_capture(&self) -> Result<context::Capture, Error>;
    fn wait_custody(
        &mut self,
        member: &Self::Member,
        owned: &mut Option<thread::Wait>,
        resolve: bool,
    ) -> Result<(), Failure>;
}
impl<B: WaitQueueServices, const N: usize> Queue<B, N> {
    pub fn wait_guest_physical_landing(
        &self,
        h: Handle,
    ) -> Result<crate::wait_core::Execution, Error> {
        let e = self.backend.execution();
        if e.cpu != self.cpu || !e.masked || e.in_irq {
            return Err(Error::Context);
        }
        let i = self.index(h)?;
        let p = self.retained.as_ref().ok_or(Error::Busy)?;
        let observed = self.backend.wait_physical_landing(p, self.member(i)?)?;
        if self.backend.execution() != e
            || observed.current != self.entries[i].thread
            || (observed.cpu, observed.irq_masked, observed.in_irq) != (e.cpu, e.masked, e.in_irq)
        {
            return Err(Error::Context);
        }
        Ok(observed)
    }
    pub fn wait_guest_observe(&self, h: Handle) -> Result<Observation, Error> {
        // A real staged/departed async transaction deliberately faults ordinary
        // Queue mutations. This immutable owner observation remains available
        // for the WaitSet's same retained bindings; it does not audit/publish a
        // transient FIFO against the not-yet-landed target.
        let e = self.backend.execution();
        if e.cpu != self.cpu || !e.masked || e.in_irq {
            return Err(Error::Context);
        }
        self.observed(self.index(h)?)
    }
    pub fn wait_guest_custody(
        &mut self,
        h: Handle,
        owned: &mut Option<thread::Wait>,
        resolve: bool,
    ) -> Result<(), Failure> {
        let before = |error| Failure {
            error,
            phase: context::FailurePhase::BeforeSwitch,
        };
        let execution = self.execution().map_err(before)?;
        self.audit_at(execution).map_err(before)?;
        let i = self.index(h).map_err(before)?;
        if self.active == Some(i)
            || self.ready[..self.len].contains(&i)
            || (resolve && self.len >= N)
        {
            return Err(before(Error::State));
        }
        let old = self.observed(i).map_err(before)?;
        if old.thread.state != thread::State::Sleeping || old.context.state != context::State::Ready
        {
            return Err(before(Error::State));
        }
        let result = self.backend.wait_custody(
            self.entries[i]
                .member
                .as_ref()
                .ok_or_else(|| before(Error::Stale))?,
            owned,
            resolve,
        );
        if self.backend.execution() != execution {
            self.faulted = true;
            return Err(Failure {
                error: Error::Quarantined,
                phase: context::FailurePhase::AfterSwitch,
            });
        }
        if let Err(f) = result {
            if f.phase == context::FailurePhase::AfterSwitch {
                self.faulted = true;
            }
            return Err(f);
        }
        let valid = self.observed(i).is_ok_and(|o| {
            o.context == old.context
                && o.thread.id == old.thread.id
                && o.thread.wait_epoch == old.thread.wait_epoch
                && o.thread.state
                    == if resolve {
                        thread::State::Ready
                    } else {
                        thread::State::Sleeping
                    }
        });
        if !valid {
            self.faulted = true;
            return Err(Failure {
                error: Error::Quarantined,
                phase: context::FailurePhase::AfterSwitch,
            });
        }
        if resolve {
            self.ready[self.len] = i;
            self.len += 1;
        }
        Ok(())
    }
}