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