// NEW custody bridge appended after the frozen CompletionBank/services bytes.
/// # Safety
/// Transfers only the exact nonclone Table Wait from the retained binding.
/// Errors leave any acquired Wait in `owned`; no backing is released. Resolve
/// consumes that exact Wait using the real Table, preserving it on Before.
/// No callback switches, allocates, releases backing, or unmasks interrupts.
pub unsafe trait WaitContextServices: QueueServices {
    /// Reads fresh physical landing evidence for this actual retained binding
    /// and Prepared image/nonce. It must check actual root, aligned SP on the
    /// binding's unique owned stack, executed landing PC, full retained wire
    /// frame/register image and nonce before
    /// returning target current. It never activates Table or changes owners.
    fn physical_wait_landing(
        &self,
        binding: &Self::Binding,
        image: &Image,
        nonce: u64,
    ) -> Result<turkos_os_thread_wait::thread_wait::Execution, Error>;
    fn take_sleep_wait(
        &mut self,
        binding: &Self::Binding,
        owned: &mut Option<thread::Wait>,
    ) -> Result<(), Failure>;
    fn resolve_sleep_wait(
        &mut self,
        binding: &Self::Binding,
        owned: &mut Option<thread::Wait>,
    ) -> Result<(), Failure>;
}
impl<B: WaitContextServices, const N: usize> CompletionBank<B, N> {
    pub fn wait_guest_physical_landing(
        &self,
        h: Handle,
    ) -> Result<turkos_os_thread_wait::thread_wait::Execution, Error> {
        self.os_k1060_queue_gate()?;
        if self.quarantined || self.pending.is_some() || self.handoff.quarantined {
            return Err(Error::Quarantined);
        }
        let p = self.handoff.pending.ok_or(Error::Pending)?;
        let bank = &self.handoff.bank;
        let before = bank.backend.execution();
        let i = bank.index(h)?;
        if p.stage != HandoffStage::Departed
            || p.target != i
            || p.nonce == 0
            || bank.retained_switch.is_none()
        {
            return Err(Error::State);
        }
        let actual = bank.validate(i)?;
        if actual.state != thread::State::Ready {
            return Err(Error::State);
        }
        let observed = bank
            .backend
            .physical_wait_landing(bank.binding(i)?, &p.image, p.nonce)?;
        if bank.backend.execution() != before
            || (observed.cpu, observed.irq_masked, observed.in_irq)
                != (before.cpu, before.daif & IRQ != 0, before.in_irq)
            || observed.current != Some(actual.id)
        {
            return Err(Error::Capture);
        }
        Ok(observed)
    }
    pub fn wait_guest_custody(
        &mut self,
        h: Handle,
        owned: &mut Option<thread::Wait>,
        resolve: bool,
    ) -> Result<(), Failure> {
        let before = |error| Failure {
            error,
            phase: FailurePhase::BeforeSwitch,
        };
        self.os_k1060_queue_gate().map_err(before)?;
        if self.quarantined
            || self.pending.is_some()
            || self.handoff.quarantined
            || self.handoff.pending.is_some()
        {
            return Err(before(Error::Pending));
        }
        let bank = &mut self.handoff.bank;
        let execution = bank.execution(false).map_err(before)?;
        let i = bank.index(h).map_err(before)?;
        let old = bank.validate(i).map_err(before)?;
        if old.state != thread::State::Sleeping
            || bank.slots[i].state != State::Ready
            || bank.active == Some(h)
            || (resolve && !owned.as_ref().is_some_and(|w| w.thread() == old.id))
            || (!resolve && owned.is_some())
        {
            return Err(before(Error::State));
        }
        let binding = bank.slots[i]
            .binding
            .as_ref()
            .ok_or_else(|| before(Error::State))?;
        let result = if resolve {
            bank.backend.resolve_sleep_wait(binding, owned)
        } else {
            bank.backend.take_sleep_wait(binding, owned)
        };
        // Fresh trusted execution must short circuit before any Table/stack
        // observation after a callback, including a successful real wake.
        if bank.backend.execution() != execution {
            bank.faulted = true;
            return Err(Failure {
                error: Error::Context,
                phase: FailurePhase::AfterSwitch,
            });
        }
        if let Err(f) = result {
            if f.phase == FailurePhase::AfterSwitch {
                bank.faulted = true;
            }
            return Err(f);
        }
        let valid = bank.validate(i).is_ok_and(|r| {
            r.id == old.id
                && r.wait_epoch == old.wait_epoch
                && r.state
                    == if resolve {
                        thread::State::Ready
                    } else {
                        thread::State::Sleeping
                    }
        });
        let custody = if resolve {
            owned.is_none()
        } else {
            owned.as_ref().is_some_and(|w| w.thread() == old.id)
        };
        if !valid || !custody {
            bank.faulted = true;
            return Err(Failure {
                error: Error::Context,
                phase: FailurePhase::AfterSwitch,
            });
        }
        Ok(())
    }
}