// 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(())
    }
}