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