//! Bounded, allocation-free channel cores: one-to-one blocking and a separate
//! two-producer/two-consumer FIFO. The EL0 send helper copies a complete user
//! range before publication. The exact TaskId exit hook reaps owned channels
//! and wakes blocked peers; multi-user handoff remains a separate gate.
//! A channel number is never a pointer or reusable slot ID.
use crate::task::{Scheduler, State, TaskId, WaitKind, WakeCause};
use crate::user_copy::UserPages;

pub const MAX_CHANNELS: usize = 2;
pub const QUEUE_CAPACITY: usize = 4;
pub const MAX_MESSAGE_BYTES: usize = 32;

/// Fixed-capacity receipt for channel objects removed by a task-exit hook.
/// The caller can revoke every user handle naming these exact channel IDs.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReapedChannels {
    pub ids: [u64; MAX_CHANNELS],
    pub count: usize,
    pub discarded_messages: usize,
    pub woken_peers: usize,
}

impl ReapedChannels {
    const EMPTY: Self = Self {
        ids: [0; MAX_CHANNELS],
        count: 0,
        discarded_messages: 0,
        woken_peers: 0,
    };
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Error {
    Full,
    IdExhausted,
    SameEndpoint,
    InvalidChannel,
    WrongProducer,
    WrongConsumer,
    TooLarge,
    QueueFull,
    QueueEmpty,
    NoReceiver,
    WaitState,
    RequiresScheduler,
    AlreadyWaiting,
    TimedOut,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SendOutcome {
    Sent,
    Blocked(Option<TaskId>),
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReceiveOutcome {
    Message(Message),
    Blocked(Option<TaskId>),
}

impl ReceiveOutcome {
    pub fn message_bytes(&self) -> Option<&[u8]> {
        match self {
            Self::Message(message) => Some(message.bytes()),
            Self::Blocked(_) => None,
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Message {
    bytes: [u8; MAX_MESSAGE_BYTES],
    len: u8,
}

impl Message {
    const EMPTY: Self = Self {
        bytes: [0; MAX_MESSAGE_BYTES],
        len: 0,
    };

    pub fn bytes(&self) -> &[u8] {
        &self.bytes[..usize::from(self.len)]
    }

    pub fn len(&self) -> usize {
        usize::from(self.len)
    }

    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
        if bytes.len() > MAX_MESSAGE_BYTES {
            return Err(Error::TooLarge);
        }
        let mut message = Self::EMPTY;
        message.bytes[..bytes.len()].copy_from_slice(bytes);
        message.len = bytes.len() as u8;
        Ok(message)
    }
}

#[derive(Clone, Copy)]
struct Channel {
    id: u64,
    producer: TaskId,
    consumer: TaskId,
    receiver_open: bool,
    queue: [Message; QUEUE_CAPACITY],
    head: usize,
    len: usize,
    sender_wait: Option<(TaskId, Message)>,
    receiver_wait: Option<TaskId>,
}

impl Channel {
    fn new(id: u64, producer: TaskId, consumer: TaskId) -> Self {
        Self {
            id,
            producer,
            consumer,
            receiver_open: true,
            queue: [Message::EMPTY; QUEUE_CAPACITY],
            head: 0,
            len: 0,
            sender_wait: None,
            receiver_wait: None,
        }
    }
}

pub struct ChannelTable {
    slots: [Option<Channel>; MAX_CHANNELS],
    next_id: u64,
}

impl ChannelTable {
    pub const fn new() -> Self {
        Self {
            slots: [None; MAX_CHANNELS],
            next_id: 1,
        }
    }

    pub fn live_channels(&self) -> usize {
        self.slots.iter().filter(|slot| slot.is_some()).count()
    }

    pub fn create(&mut self, producer: TaskId, consumer: TaskId) -> Result<u64, Error> {
        if producer == consumer {
            return Err(Error::SameEndpoint);
        }
        let vacant = self
            .slots
            .iter_mut()
            .find(|slot| slot.is_none())
            .ok_or(Error::Full)?;
        let following = self.next_id.checked_add(1).ok_or(Error::IdExhausted)?;
        let id = self.next_id;
        *vacant = Some(Channel::new(id, producer, consumer));
        self.next_id = following;
        Ok(id)
    }

    fn channel_mut(&mut self, id: u64) -> Result<&mut Channel, Error> {
        self.slots
            .iter_mut()
            .flatten()
            .find(|channel| channel.id == id)
            .ok_or(Error::InvalidChannel)
    }

    pub fn queued(&self, id: u64) -> Result<usize, Error> {
        self.slots
            .iter()
            .flatten()
            .find(|channel| channel.id == id)
            .map(|channel| channel.len)
            .ok_or(Error::InvalidChannel)
    }

    pub fn send(&mut self, id: u64, caller: TaskId, bytes: &[u8]) -> Result<(), Error> {
        let channel = self.channel_mut(id)?;
        if channel.receiver_wait.is_some() {
            return Err(Error::RequiresScheduler);
        }
        if channel.producer != caller {
            return Err(Error::WrongProducer);
        }
        if !channel.receiver_open {
            return Err(Error::NoReceiver);
        }
        let message = Message::from_bytes(bytes)?;
        if channel.len == QUEUE_CAPACITY {
            return Err(Error::QueueFull);
        }
        let tail = (channel.head + channel.len) % QUEUE_CAPACITY;
        channel.queue[tail] = message;
        channel.len += 1;
        Ok(())
    }

    pub fn receive(&mut self, id: u64, caller: TaskId) -> Result<Message, Error> {
        let channel = self.channel_mut(id)?;
        if channel.sender_wait.is_some() {
            return Err(Error::RequiresScheduler);
        }
        if channel.consumer != caller {
            return Err(Error::WrongConsumer);
        }
        if !channel.receiver_open {
            return Err(Error::NoReceiver);
        }
        if channel.len == 0 {
            return Err(Error::QueueEmpty);
        }
        let message = channel.queue[channel.head];
        channel.queue[channel.head] = Message::EMPTY;
        channel.head = (channel.head + 1) % QUEUE_CAPACITY;
        channel.len -= 1;
        Ok(message)
    }

    pub fn close_receiver(&mut self, id: u64, caller: TaskId) -> Result<(), Error> {
        let channel = self.channel_mut(id)?;
        if channel.sender_wait.is_some() || channel.receiver_wait.is_some() {
            return Err(Error::RequiresScheduler);
        }
        if channel.consumer != caller {
            return Err(Error::WrongConsumer);
        }
        if !channel.receiver_open {
            return Err(Error::NoReceiver);
        }
        channel.receiver_open = false;
        Ok(())
    }

    pub fn destroy(&mut self, id: u64, caller: TaskId) -> Result<(), Error> {
        let slot = self
            .slots
            .iter_mut()
            .find(|slot| slot.is_some_and(|channel| channel.id == id))
            .ok_or(Error::InvalidChannel)?;
        if slot
            .as_ref()
            .is_some_and(|channel| channel.producer != caller)
        {
            return Err(Error::WrongProducer);
        }
        if slot
            .as_ref()
            .is_some_and(|channel| channel.sender_wait.is_some() || channel.receiver_wait.is_some())
        {
            return Err(Error::RequiresScheduler);
        }
        *slot = None;
        Ok(())
    }

    /// Single-CPU critical section: validate both owners before publishing a
    /// wait registration or selecting the next runnable task.
    pub fn send_or_block(
        &mut self,
        id: u64,
        caller: TaskId,
        bytes: &[u8],
        scheduler: &mut Scheduler,
        now: u64,
    ) -> Result<SendOutcome, Error> {
        self.poll_deadlines(scheduler, now)?;
        if scheduler.running() != Some(caller) {
            return Err(Error::WaitState);
        }
        let channel = self.channel_mut(id)?;
        if channel.producer != caller {
            return Err(Error::WrongProducer);
        }
        if !channel.receiver_open {
            return Err(Error::NoReceiver);
        }
        let message = Message::from_bytes(bytes)?;
        if channel.len == QUEUE_CAPACITY {
            if channel.sender_wait.is_some() {
                return Err(Error::AlreadyWaiting);
            }
            let next = scheduler
                .block_current(id, WaitKind::Send, None, now)
                .map_err(|_| Error::WaitState)?;
            channel.sender_wait = Some((caller, message));
            return Ok(SendOutcome::Blocked(next));
        }
        let tail = (channel.head + channel.len) % QUEUE_CAPACITY;
        channel.queue[tail] = message;
        channel.len += 1;
        if let Some(receiver) = channel.receiver_wait.take() {
            scheduler
                .wake_blocked(receiver, id, WaitKind::Receive, WakeCause::MessageAvailable)
                .map_err(|_| Error::WaitState)?;
        }
        Ok(SendOutcome::Sent)
    }

    pub fn receive_or_block(
        &mut self,
        id: u64,
        caller: TaskId,
        scheduler: &mut Scheduler,
        deadline: Option<u64>,
        now: u64,
    ) -> Result<ReceiveOutcome, Error> {
        self.poll_deadlines(scheduler, now)?;
        if scheduler.running() != Some(caller) {
            return Err(Error::WaitState);
        }
        let channel = self.channel_mut(id)?;
        if channel.consumer != caller {
            return Err(Error::WrongConsumer);
        }
        if !channel.receiver_open {
            return Err(Error::NoReceiver);
        }
        if channel.len == 0 {
            if deadline.is_some_and(|until| until <= now) {
                return Err(Error::TimedOut);
            }
            if channel.receiver_wait.is_some() {
                return Err(Error::AlreadyWaiting);
            }
            let next = scheduler
                .block_current(id, WaitKind::Receive, deadline, now)
                .map_err(|_| Error::WaitState)?;
            channel.receiver_wait = Some(caller);
            return Ok(ReceiveOutcome::Blocked(next));
        }
        let message = channel.queue[channel.head];
        channel.queue[channel.head] = Message::EMPTY;
        channel.head = (channel.head + 1) % QUEUE_CAPACITY;
        channel.len -= 1;
        if let Some((sender, pending)) = channel.sender_wait.take() {
            let tail = (channel.head + channel.len) % QUEUE_CAPACITY;
            channel.queue[tail] = pending;
            channel.len += 1;
            scheduler
                .wake_blocked(sender, id, WaitKind::Send, WakeCause::SpaceAvailable)
                .map_err(|_| Error::WaitState)?;
        }
        Ok(ReceiveOutcome::Message(message))
    }

    pub fn poll_deadlines(&mut self, scheduler: &mut Scheduler, now: u64) -> Result<usize, Error> {
        let mut count = 0;
        for channel in self.slots.iter_mut().flatten() {
            let Some(receiver) = channel.receiver_wait else {
                continue;
            };
            match scheduler.state(receiver) {
                Ok(State::Blocked {
                    channel: id,
                    kind: WaitKind::Receive,
                    deadline,
                }) if id == channel.id => {
                    if deadline.is_some_and(|until| until <= now) {
                        scheduler
                            .wake_blocked(receiver, id, WaitKind::Receive, WakeCause::TimedOut)
                            .map_err(|_| Error::WaitState)?;
                        channel.receiver_wait = None;
                        count += 1;
                    }
                }
                _ => return Err(Error::WaitState),
            }
        }
        Ok(count)
    }

    pub fn close_receiver_wait(
        &mut self,
        id: u64,
        caller: TaskId,
        scheduler: &mut Scheduler,
    ) -> Result<(), Error> {
        let channel = self.channel_mut(id)?;
        if channel.consumer != caller {
            return Err(Error::WrongConsumer);
        }
        if !channel.receiver_open {
            return Err(Error::NoReceiver);
        }
        if let Some((sender, _)) = channel.sender_wait {
            scheduler
                .wake_blocked(sender, id, WaitKind::Send, WakeCause::ChannelClosed)
                .map_err(|_| Error::WaitState)?;
        }
        channel.sender_wait = None;
        channel.receiver_open = false;
        Ok(())
    }

    pub fn destroy_wait(
        &mut self,
        id: u64,
        caller: TaskId,
        scheduler: &mut Scheduler,
    ) -> Result<(), Error> {
        let slot = self
            .slots
            .iter_mut()
            .find(|slot| slot.is_some_and(|channel| channel.id == id))
            .ok_or(Error::InvalidChannel)?;
        let channel = slot.as_ref().ok_or(Error::InvalidChannel)?;
        if channel.producer != caller {
            return Err(Error::WrongProducer);
        }
        if let Some(receiver) = channel.receiver_wait {
            scheduler
                .wake_blocked(receiver, id, WaitKind::Receive, WakeCause::ChannelClosed)
                .map_err(|_| Error::WaitState)?;
        }
        *slot = None;
        Ok(())
    }

    /// Called after the scheduler has marked one exact TaskId Exited. Preflight
    /// all affected wait registrations before changing any channel or queue;
    /// a blocked surviving peer is awakened once with ChannelClosed.
    pub fn reap_exited_task(
        &mut self,
        dead: TaskId,
        scheduler: &mut Scheduler,
    ) -> Result<ReapedChannels, Error> {
        if !matches!(scheduler.state(dead), Ok(State::Exited { .. })) {
            return Err(Error::WaitState);
        }
        let mut wake_count = 0;
        for channel in self.slots.iter().flatten() {
            if channel.producer != dead && channel.consumer != dead {
                continue;
            }
            if channel.sender_wait.is_some() && channel.receiver_wait.is_some() {
                return Err(Error::WaitState);
            }
            for (waiter, kind) in [
                (channel.sender_wait.map(|(id, _)| id), WaitKind::Send),
                (channel.receiver_wait, WaitKind::Receive),
            ] {
                let Some(waiter) = waiter else { continue };
                if waiter == dead
                    || !matches!(scheduler.state(waiter), Ok(State::Blocked { channel: id, kind: actual, .. }) if id == channel.id && actual == kind)
                {
                    return Err(Error::WaitState);
                }
                wake_count += 1;
            }
        }
        if scheduler.ready_len() + wake_count > crate::task::MAX_TASKS {
            return Err(Error::WaitState);
        }
        let mut reaped = ReapedChannels::EMPTY;
        for slot in &mut self.slots {
            let Some(channel) = slot else { continue };
            if channel.producer != dead && channel.consumer != dead {
                continue;
            }
            if let Some((waiter, _)) = channel.sender_wait {
                scheduler
                    .wake_blocked(waiter, channel.id, WaitKind::Send, WakeCause::ChannelClosed)
                    .map_err(|_| Error::WaitState)?;
                reaped.woken_peers += 1;
            }
            if let Some(waiter) = channel.receiver_wait {
                scheduler
                    .wake_blocked(
                        waiter,
                        channel.id,
                        WaitKind::Receive,
                        WakeCause::ChannelClosed,
                    )
                    .map_err(|_| Error::WaitState)?;
                reaped.woken_peers += 1;
            }
            reaped.ids[reaped.count] = channel.id;
            reaped.count += 1;
            reaped.discarded_messages += channel.len;
            *slot = None;
        }
        Ok(reaped)
    }
}

/// Total order for exactly two authorized producers and two authorized
/// consumers on one CPU. This does not imply simultaneous execution or an
/// SMP synchronization guarantee.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct SequencedMessage {
    pub sequence: u64,
    pub message: Message,
}

impl SequencedMessage {
    const EMPTY: Self = Self {
        sequence: 0,
        message: Message::EMPTY,
    };
}

pub struct SharedChannel {
    producers: [TaskId; 2],
    consumers: [TaskId; 2],
    queue: [SequencedMessage; QUEUE_CAPACITY],
    head: usize,
    len: usize,
    next_sequence: u64,
}

/// A bounded sequential witness for one producer-to-consumer message path.
/// Sequence gaps, reversed counters and overlapping observations fail before
/// any aggregate is changed. Counter units are the platform's CNTFRQ ticks.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TimelineError {
    Capacity,
    Order,
    Clock,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MessageTimeline {
    count: u64,
    last_receive: u64,
    min: u64,
    max: u64,
    total: u64,
}

impl MessageTimeline {
    pub const fn new() -> Self {
        Self {
            count: 0,
            last_receive: 0,
            min: u64::MAX,
            max: 0,
            total: 0,
        }
    }

    pub fn record(
        &mut self,
        sequence: u64,
        sent: u64,
        received: u64,
    ) -> Result<u64, TimelineError> {
        if self.count == 5 {
            return Err(TimelineError::Capacity);
        }
        if sequence != self.count + 1 {
            return Err(TimelineError::Order);
        }
        if sent < self.last_receive || received <= sent {
            return Err(TimelineError::Clock);
        }
        let latency = received - sent;
        let total = self
            .total
            .checked_add(latency)
            .ok_or(TimelineError::Clock)?;
        self.count += 1;
        self.last_receive = received;
        self.min = self.min.min(latency);
        self.max = self.max.max(latency);
        self.total = total;
        Ok(latency)
    }

    pub fn summary(&self) -> (u64, u64, u64, u64) {
        (
            self.count,
            if self.count == 0 { 0 } else { self.min },
            self.max,
            self.total,
        )
    }
}

impl Default for MessageTimeline {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum UserSendError {
    Channel(Error),
    Copy(i64),
}

impl SharedChannel {
    pub fn new(producers: [TaskId; 2], consumers: [TaskId; 2]) -> Result<Self, Error> {
        let owners = [producers[0], producers[1], consumers[0], consumers[1]];
        for left in 0..owners.len() {
            for right in left + 1..owners.len() {
                if owners[left] == owners[right] {
                    return Err(Error::SameEndpoint);
                }
            }
        }
        Ok(Self {
            producers,
            consumers,
            queue: [SequencedMessage::EMPTY; QUEUE_CAPACITY],
            head: 0,
            len: 0,
            next_sequence: 1,
        })
    }

    pub fn queued(&self) -> usize {
        self.len
    }

    pub fn send(&mut self, caller: TaskId, bytes: &[u8]) -> Result<u64, Error> {
        if !self.producers.contains(&caller) {
            return Err(Error::WrongProducer);
        }
        let message = Message::from_bytes(bytes)?;
        if self.len == QUEUE_CAPACITY {
            return Err(Error::QueueFull);
        }
        let following = self
            .next_sequence
            .checked_add(1)
            .ok_or(Error::IdExhausted)?;
        let sequence = self.next_sequence;
        let tail = (self.head + self.len) % QUEUE_CAPACITY;
        self.queue[tail] = SequencedMessage { sequence, message };
        self.len += 1;
        self.next_sequence = following;
        Ok(sequence)
    }

    /// Copy an entire user range to a kernel-owned scratch message before
    /// publishing it. The caller must bind `pages` to `caller`'s live private
    /// address space; this EL1 helper is not itself an EL0 syscall gate.
    ///
    /// # Safety
    /// `pages` must name distinct, live, identity-mapped frames owned by the
    /// calling task. The scratch buffer must not alias any frame.
    pub unsafe fn send_from_user(
        &mut self,
        caller: TaskId,
        pages: UserPages,
        address: u64,
        length: usize,
    ) -> Result<u64, UserSendError> {
        if !self.producers.contains(&caller) {
            return Err(UserSendError::Channel(Error::WrongProducer));
        }
        if length > MAX_MESSAGE_BYTES {
            return Err(UserSendError::Channel(Error::TooLarge));
        }
        if self.len == QUEUE_CAPACITY {
            return Err(UserSendError::Channel(Error::QueueFull));
        }
        if self.next_sequence == u64::MAX {
            return Err(UserSendError::Channel(Error::IdExhausted));
        }
        let mut scratch = [0u8; MAX_MESSAGE_BYTES];
        // SAFETY: required live, non-aliasing frame ownership is the caller's
        // contract; UserPages validates the complete range before any read.
        unsafe { pages.copy_from(address, &mut scratch[..length]) }.map_err(UserSendError::Copy)?;
        self.send(caller, &scratch[..length])
            .map_err(UserSendError::Channel)
    }

    pub fn receive(&mut self, caller: TaskId) -> Result<SequencedMessage, Error> {
        if !self.consumers.contains(&caller) {
            return Err(Error::WrongConsumer);
        }
        if self.len == 0 {
            return Err(Error::QueueEmpty);
        }
        let result = self.queue[self.head];
        self.queue[self.head] = SequencedMessage::EMPTY;
        self.head = (self.head + 1) % QUEUE_CAPACITY;
        self.len -= 1;
        Ok(result)
    }
}

impl Default for ChannelTable {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    extern crate std;
    use super::*;
    use sha2::{Digest, Sha256};
    const A: TaskId = TaskId {
        slot: 0,
        generation: 1,
    };
    const B: TaskId = TaskId {
        slot: 1,
        generation: 2,
    };
    const C: TaskId = TaskId {
        slot: 2,
        generation: 3,
    };
    const D: TaskId = TaskId {
        slot: 3,
        generation: 4,
    };

    #[test]
    fn k445_deterministic_ipc_identity_fuzz_never_opens_foreign_channel() {
        let ids = [
            0,
            1,
            2,
            3,
            u64::MAX,
            u64::MAX - 1,
            0x10000,
            0x8000_0000_0000_0001,
        ];
        let callers = [
            A,
            B,
            C,
            D,
            TaskId {
                slot: 0,
                generation: 2,
            },
            TaskId {
                slot: 1,
                generation: 3,
            },
            TaskId {
                slot: u8::MAX,
                generation: u64::MAX,
            },
        ];
        let snapshot = |table: &ChannelTable| {
            table.slots.map(|slot| {
                slot.map(|channel| {
                    (
                        channel.id,
                        channel.producer,
                        channel.consumer,
                        channel.receiver_open,
                        channel.queue,
                        channel.head,
                        channel.len,
                        channel.sender_wait,
                        channel.receiver_wait,
                    )
                })
            })
        };
        let mut digest = Sha256::new();
        let mut cases = 0usize;
        let mut accepted = 0usize;
        let mut invalid = 0usize;
        let mut wrong_producer = 0usize;
        let mut wrong_consumer = 0usize;
        let mut check = |id: u64, caller: TaskId, operation: u8| {
            let mut table = ChannelTable::new();
            assert_eq!(table.create(A, B), Ok(1));
            assert_eq!(table.create(C, D), Ok(2));
            assert_eq!(table.send(1, A, b"A"), Ok(()));
            assert_eq!(table.send(2, C, b"C"), Ok(()));
            let before = snapshot(&table);
            let owner = match id {
                1 => Some((A, B, 0usize)),
                2 => Some((C, D, 1usize)),
                _ => None,
            };
            let expected = match owner {
                None => Err(Error::InvalidChannel),
                Some((producer, consumer, _)) => match operation {
                    0 | 2 if caller != producer => Err(Error::WrongProducer),
                    1 | 3 if caller != consumer => Err(Error::WrongConsumer),
                    _ => Ok(()),
                },
            };
            let actual = match operation {
                0 => table.send(id, caller, b"X"),
                1 => table.receive(id, caller).map(|_| ()),
                2 => table.destroy(id, caller),
                3 => table.close_receiver(id, caller),
                _ => unreachable!(),
            };
            assert_eq!(
                actual, expected,
                "id={id} caller={caller:?} operation={operation}"
            );
            if actual.is_err() {
                assert_eq!(
                    snapshot(&table),
                    before,
                    "rejected identity mutated channel state"
                );
            } else if let Some((_, _, slot)) = owner {
                assert_eq!(snapshot(&table)[1 - slot], before[1 - slot]);
                let expected_length = match operation {
                    0 => Ok(2),
                    1 => Ok(0),
                    2 => Err(Error::InvalidChannel),
                    3 => Ok(1),
                    _ => unreachable!(),
                };
                assert_eq!(table.queued(id), expected_length);
            }
            digest.update(id.to_le_bytes());
            digest.update([caller.slot]);
            digest.update(caller.generation.to_le_bytes());
            digest.update([operation]);
            digest.update([match actual {
                Ok(()) => 0,
                Err(Error::InvalidChannel) => 1,
                Err(Error::WrongProducer) => 2,
                Err(Error::WrongConsumer) => 3,
                _ => unreachable!(),
            }]);
            cases += 1;
            match actual {
                Ok(()) => accepted += 1,
                Err(Error::InvalidChannel) => invalid += 1,
                Err(Error::WrongProducer) => wrong_producer += 1,
                Err(Error::WrongConsumer) => wrong_consumer += 1,
                _ => unreachable!(),
            }
        };
        for id in ids {
            for caller in callers {
                for operation in 0..4 {
                    check(id, caller, operation);
                }
            }
        }
        let mut state = 0x4450_2026_1dc0_ffeeu64;
        for index in 0..32_768 {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            let id = if index % 4 == 0 {
                state
            } else {
                ids[state as usize % ids.len()]
            };
            let caller = callers[(state >> 17) as usize % callers.len()];
            let operation = ((state >> 35) & 3) as u8;
            check(id, caller, operation);
        }
        assert_eq!(cases, 32_992);
        assert!(accepted > 0 && invalid > 0 && wrong_producer > 0 && wrong_consumer > 0);

        let mut epochs = ChannelTable::new();
        for expected_id in 1..=64 {
            let id = epochs.create(A, B).unwrap();
            assert_eq!(id, expected_id);
            assert_eq!(epochs.destroy(id, A), Ok(()));
            for stale in 1..=id {
                assert_eq!(epochs.send(stale, A, b"stale"), Err(Error::InvalidChannel));
            }
        }
        assert_eq!(epochs.live_channels(), 0);
        let hash = digest
            .finalize()
            .iter()
            .map(|byte| std::format!("{byte:02x}"))
            .collect::<std::string::String>();
        std::eprintln!(
            "K445 CASES={cases} OK={accepted} INVALID={invalid} WRONG_PRODUCER={wrong_producer} WRONG_CONSUMER={wrong_consumer} STALE_EPOCHS=64 SHA256={hash}"
        );
    }

    #[test]
    fn k310_two_producers_share_one_total_fifo_order() {
        let mut channel = SharedChannel::new([A, B], [C, D]).unwrap();
        assert_eq!(channel.send(C, b"BAD"), Err(Error::WrongProducer));
        assert_eq!(channel.send(A, b"A1"), Ok(1));
        assert_eq!(channel.send(B, b"B1"), Ok(2));
        assert_eq!(channel.send(A, b"A2"), Ok(3));
        assert_eq!(channel.send(B, b"B2"), Ok(4));
        assert_eq!(channel.send(A, b"FULL"), Err(Error::QueueFull));
        assert_eq!(channel.queued(), 4);
        for (sequence, expected) in [(1, b"A1"), (2, b"B1"), (3, b"A2"), (4, b"B2")] {
            let item = channel.receive(C).unwrap();
            assert_eq!(item.sequence, sequence);
            assert_eq!(item.message.bytes(), expected);
        }
        assert_eq!(channel.queued(), 0);
        assert_eq!(channel.send(B, b"B3"), Ok(5));
        assert_eq!(channel.receive(D).unwrap().sequence, 5);
    }

    #[test]
    fn k311_two_consumers_cannot_receive_one_message_twice() {
        let mut channel = SharedChannel::new([A, B], [C, D]).unwrap();
        for byte in [b'0', b'1', b'2', b'3'] {
            channel.send(A, &[byte]).unwrap();
        }
        let stale_d = TaskId {
            slot: D.slot,
            generation: D.generation + 1,
        };
        assert_eq!(channel.receive(A), Err(Error::WrongConsumer));
        assert_eq!(channel.receive(stale_d), Err(Error::WrongConsumer));
        for (owner, sequence, byte) in [(C, 1, b'0'), (D, 2, b'1'), (C, 3, b'2'), (D, 4, b'3')] {
            let item = channel.receive(owner).unwrap();
            assert_eq!(
                (item.sequence, item.message.bytes()),
                (sequence, &[byte][..])
            );
        }
        assert_eq!(channel.receive(C), Err(Error::QueueEmpty));
        assert_eq!(channel.receive(D), Err(Error::QueueEmpty));
        assert_eq!(channel.queued(), 0);
    }

    #[test]
    fn k312_user_send_owns_bytes_after_source_mutates() {
        let code = [0x11u8; 4096];
        let mut data = [0u8; 4096];
        let stack = [0x22u8; 4096];
        data[..4].copy_from_slice(b"ABCD");
        let pages = UserPages {
            code_frame: code.as_ptr() as u64,
            data_frame: data.as_mut_ptr() as u64,
            stack_frame: stack.as_ptr() as u64,
        };
        let mut channel = SharedChannel::new([A, B], [C, D]).unwrap();
        assert_eq!(
            unsafe { channel.send_from_user(A, pages, crate::mmu::USER_DATA_VA, 4) },
            Ok(1)
        );
        data[..4].copy_from_slice(b"WXYZ");
        let item = channel.receive(C).unwrap();
        assert_eq!(item.sequence, 1);
        assert_eq!(item.message.bytes(), b"ABCD");
        assert_eq!(&data[..4], b"WXYZ");
        assert_eq!(channel.receive(D), Err(Error::QueueEmpty));
    }

    #[test]
    fn k313_bad_user_range_never_enqueues_a_partial_message() {
        let code = [0x11u8; 4096];
        let mut data = [0u8; 4096];
        let stack = [0x22u8; 4096];
        data[4094..].copy_from_slice(&[0xa5, 0x5a]);
        let pages = UserPages {
            code_frame: code.as_ptr() as u64,
            data_frame: data.as_mut_ptr() as u64,
            stack_frame: stack.as_ptr() as u64,
        };
        let mut channel = SharedChannel::new([A, B], [C, D]).unwrap();
        assert_eq!(channel.send(A, b"KEEP"), Ok(1));
        for address in [crate::mmu::USER_DATA_VA + 4094, u64::MAX - 1] {
            assert_eq!(
                unsafe { channel.send_from_user(A, pages, address, 4) },
                Err(UserSendError::Copy(crate::user_copy::EFAULT))
            );
            assert_eq!(channel.queued(), 1);
        }
        assert_eq!(
            unsafe { channel.send_from_user(A, pages, crate::mmu::USER_DATA_VA, 33) },
            Err(UserSendError::Channel(Error::TooLarge))
        );
        assert_eq!(
            unsafe { channel.send_from_user(C, pages, crate::mmu::USER_DATA_VA, 4) },
            Err(UserSendError::Channel(Error::WrongProducer))
        );
        assert_eq!(&data[4094..], &[0xa5, 0x5a]);
        assert_eq!(channel.receive(C).unwrap().message.bytes(), b"KEEP");
        assert_eq!(channel.receive(D), Err(Error::QueueEmpty));
        assert_eq!(channel.send(B, b"NEXT"), Ok(2));
    }

    #[test]
    fn shared_channel_duplicate_owners_and_sequence_overflow_are_atomic() {
        assert!(matches!(
            SharedChannel::new([A, A], [C, D]),
            Err(Error::SameEndpoint)
        ));
        assert!(matches!(
            SharedChannel::new([A, B], [C, A]),
            Err(Error::SameEndpoint)
        ));
        let mut channel = SharedChannel::new([A, B], [C, D]).unwrap();
        assert_eq!(
            channel.send(A, &[0x58; MAX_MESSAGE_BYTES + 1]),
            Err(Error::TooLarge)
        );
        assert_eq!(channel.queued(), 0);
        channel.next_sequence = u64::MAX;
        assert_eq!(channel.send(A, b"X"), Err(Error::IdExhausted));
        assert_eq!(channel.queued(), 0);
        assert_eq!(channel.next_sequence, u64::MAX);
    }

    #[test]
    fn k306_full_channel_blocks_sender_until_receiver_releases_space() {
        let mut scheduler = Scheduler::new();
        let a = scheduler.spawn().unwrap();
        let b = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        for byte in [b'0', b'1', b'2', b'3'] {
            assert_eq!(
                table.send_or_block(id, a, &[byte], &mut scheduler, 0),
                Ok(SendOutcome::Sent)
            );
        }
        let mut fifth = [b'4'];
        assert_eq!(
            table.send_or_block(id, a, &fifth, &mut scheduler, 0),
            Ok(SendOutcome::Blocked(Some(b)))
        );
        fifth[0] = b'X';
        assert_eq!(
            scheduler.state(a),
            Ok(State::Blocked {
                channel: id,
                kind: WaitKind::Send,
                deadline: None
            })
        );
        assert_eq!(scheduler.ready_len(), 0);
        assert_eq!(table.receive(id, b), Err(Error::RequiresScheduler));
        assert_eq!(
            table
                .receive_or_block(id, b, &mut scheduler, None, 0)
                .unwrap()
                .message_bytes(),
            Some(&b"0"[..])
        );
        assert_eq!(table.queued(id), Ok(4));
        for expected in [b"1", b"2", b"3", b"4"] {
            assert_eq!(
                table
                    .receive_or_block(id, b, &mut scheduler, None, 0)
                    .unwrap()
                    .message_bytes(),
                Some(&expected[..])
            );
        }
        assert_eq!(fifth[0], b'X');
        assert_eq!(table.queued(id), Ok(0));
        assert_eq!(scheduler.state(a), Ok(State::Ready));
        assert_eq!(scheduler.ready_len(), 1);
        assert_eq!(scheduler.yield_current(false), Ok(Some(a)));
        assert_eq!(
            scheduler.take_wake_cause(a),
            Ok(Some(WakeCause::SpaceAvailable))
        );
        assert_eq!(scheduler.take_wake_cause(a), Ok(None));
    }

    #[test]
    fn k307_empty_channel_blocks_only_its_receiver_then_wakes_on_send() {
        let mut scheduler = Scheduler::new();
        let b = scheduler.spawn().unwrap();
        let a = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(b)));
        assert_eq!(
            table.receive_or_block(id, b, &mut scheduler, None, 0),
            Ok(ReceiveOutcome::Blocked(Some(a)))
        );
        assert_eq!(
            scheduler.state(b),
            Ok(State::Blocked {
                channel: id,
                kind: WaitKind::Receive,
                deadline: None
            })
        );
        assert_eq!(scheduler.ready_len(), 0);
        assert_eq!(table.send(id, a, b"PING"), Err(Error::RequiresScheduler));
        assert_eq!(
            table.send_or_block(id, a, b"PING", &mut scheduler, 0),
            Ok(SendOutcome::Sent)
        );
        assert_eq!(scheduler.state(b), Ok(State::Ready));
        assert_eq!(scheduler.yield_current(false), Ok(Some(b)));
        assert_eq!(
            scheduler.take_wake_cause(b),
            Ok(Some(WakeCause::MessageAvailable))
        );
        assert_eq!(
            table
                .receive_or_block(id, b, &mut scheduler, None, 0)
                .unwrap()
                .message_bytes(),
            Some(&b"PING"[..])
        );
    }

    #[test]
    fn k308_receive_deadline_wakes_once_and_returns_timeout() {
        let mut scheduler = Scheduler::new();
        let b = scheduler.spawn().unwrap();
        let a = TaskId {
            slot: 1,
            generation: 2,
        };
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(b)));
        assert_eq!(
            table.receive_or_block(id, b, &mut scheduler, Some(10), 5),
            Ok(ReceiveOutcome::Blocked(None))
        );
        assert_eq!(table.poll_deadlines(&mut scheduler, 9), Ok(0));
        assert_eq!(scheduler.ready_len(), 0);
        assert_eq!(table.poll_deadlines(&mut scheduler, 10), Ok(1));
        assert_eq!(table.poll_deadlines(&mut scheduler, 10), Ok(0));
        assert_eq!(scheduler.ready_len(), 1);
        assert_eq!(scheduler.run_next(), Ok(Some(b)));
        assert_eq!(scheduler.take_wake_cause(b), Ok(Some(WakeCause::TimedOut)));
        assert_eq!(
            table.receive_or_block(id, b, &mut scheduler, Some(10), 10),
            Err(Error::TimedOut)
        );
        assert_eq!(table.queued(id), Ok(0));
    }

    #[test]
    fn k309_close_wakes_blocked_sender_and_destroy_wakes_receiver() {
        let mut scheduler = Scheduler::new();
        let a = scheduler.spawn().unwrap();
        let b = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        for byte in 0..QUEUE_CAPACITY {
            assert_eq!(
                table.send_or_block(id, a, &[byte as u8], &mut scheduler, 0),
                Ok(SendOutcome::Sent)
            );
        }
        assert_eq!(
            table.send_or_block(id, a, b"LATE", &mut scheduler, 0),
            Ok(SendOutcome::Blocked(Some(b)))
        );
        assert_eq!(table.close_receiver(id, b), Err(Error::RequiresScheduler));
        assert_eq!(table.close_receiver_wait(id, b, &mut scheduler), Ok(()));
        assert_eq!(scheduler.state(a), Ok(State::Ready));
        assert_eq!(scheduler.yield_current(false), Ok(Some(a)));
        assert_eq!(
            scheduler.take_wake_cause(a),
            Ok(Some(WakeCause::ChannelClosed))
        );
        assert_eq!(
            table.send_or_block(id, a, b"AGAIN", &mut scheduler, 0),
            Err(Error::NoReceiver)
        );

        let mut scheduler = Scheduler::new();
        let b = scheduler.spawn().unwrap();
        let a = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(b)));
        assert_eq!(
            table.receive_or_block(id, b, &mut scheduler, None, 0),
            Ok(ReceiveOutcome::Blocked(Some(a)))
        );
        assert_eq!(table.destroy(id, a), Err(Error::RequiresScheduler));
        assert_eq!(table.destroy_wait(id, a, &mut scheduler), Ok(()));
        assert_eq!(scheduler.yield_current(false), Ok(Some(b)));
        assert_eq!(
            scheduler.take_wake_cause(b),
            Ok(Some(WakeCause::ChannelClosed))
        );
        assert_eq!(
            table.receive_or_block(id, b, &mut scheduler, None, 0),
            Err(Error::InvalidChannel)
        );
    }

    #[test]
    fn k315_exited_producer_discards_queue_and_reuses_slot_with_new_id() {
        let mut scheduler = Scheduler::new();
        let a = scheduler.spawn().unwrap();
        let b = scheduler.spawn().unwrap();
        let c = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let first = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        assert_eq!(table.send(first, a, b"ABCD"), Ok(()));
        assert_eq!(table.queued(first), Ok(1));
        assert_eq!(scheduler.exit_current(0xa), Ok(Some(b)));
        assert_eq!(
            table.reap_exited_task(a, &mut scheduler),
            Ok(ReapedChannels {
                ids: [first, 0],
                count: 1,
                discarded_messages: 1,
                woken_peers: 0,
            })
        );
        assert_eq!(table.live_channels(), 0);
        assert_eq!(table.receive(first, b), Err(Error::InvalidChannel));
        assert_eq!(table.create(b, c), Ok(first + 1));
        assert_eq!(table.queued(first), Err(Error::InvalidChannel));
        assert_eq!(table.queued(first + 1), Ok(0));
    }

    #[test]
    fn k315_exited_endpoint_wakes_blocked_peer_once() {
        let mut scheduler = Scheduler::new();
        let a = scheduler.spawn().unwrap();
        let b = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        for byte in 0..QUEUE_CAPACITY {
            assert_eq!(table.send(id, a, &[byte as u8]), Ok(()));
        }
        assert_eq!(
            table.send_or_block(id, a, b"LATE", &mut scheduler, 0),
            Ok(SendOutcome::Blocked(Some(b)))
        );
        assert_eq!(scheduler.exit_current(0xb), Ok(None));
        assert_eq!(
            table
                .reap_exited_task(b, &mut scheduler)
                .unwrap()
                .woken_peers,
            1
        );
        assert_eq!(table.live_channels(), 0);
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        assert_eq!(
            scheduler.take_wake_cause(a),
            Ok(Some(WakeCause::ChannelClosed))
        );
        assert_eq!(scheduler.take_wake_cause(a), Ok(None));

        let mut scheduler = Scheduler::new();
        let b = scheduler.spawn().unwrap();
        let a = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(b)));
        assert_eq!(
            table.receive_or_block(id, b, &mut scheduler, None, 0),
            Ok(ReceiveOutcome::Blocked(Some(a)))
        );
        assert_eq!(scheduler.exit_current(0xa), Ok(None));
        assert_eq!(
            table
                .reap_exited_task(a, &mut scheduler)
                .unwrap()
                .woken_peers,
            1
        );
        assert_eq!(scheduler.run_next(), Ok(Some(b)));
        assert_eq!(
            scheduler.take_wake_cause(b),
            Ok(Some(WakeCause::ChannelClosed))
        );
        assert_eq!(table.receive(id, b), Err(Error::InvalidChannel));
    }

    #[test]
    fn k315_live_stale_or_corrupt_wait_registration_cannot_partially_reap() {
        let mut scheduler = Scheduler::new();
        let a = scheduler.spawn().unwrap();
        let b = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let id = table.create(a, b).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        assert_eq!(
            table.reap_exited_task(a, &mut scheduler),
            Err(Error::WaitState)
        );
        assert_eq!(
            table.reap_exited_task(
                TaskId {
                    slot: a.slot,
                    generation: a.generation + 9
                },
                &mut scheduler
            ),
            Err(Error::WaitState)
        );
        assert_eq!(table.live_channels(), 1);
        table.channel_mut(id).unwrap().sender_wait = Some((b, Message::EMPTY));
        assert_eq!(scheduler.exit_current(0xa), Ok(Some(b)));
        assert_eq!(
            table.reap_exited_task(a, &mut scheduler),
            Err(Error::WaitState)
        );
        assert_eq!(table.live_channels(), 1);
        assert_eq!(table.queued(id), Ok(0));
    }

    #[test]
    fn k315_two_channel_exit_preflights_all_before_removing_either() {
        let mut scheduler = Scheduler::new();
        let a = scheduler.spawn().unwrap();
        let b = scheduler.spawn().unwrap();
        let c = scheduler.spawn().unwrap();
        let mut table = ChannelTable::new();
        let first = table.create(a, b).unwrap();
        let second = table.create(a, c).unwrap();
        assert_eq!(scheduler.run_next(), Ok(Some(a)));
        assert_eq!(table.send(first, a, b"ONE"), Ok(()));
        assert_eq!(table.send(second, a, b"TWO"), Ok(()));
        assert_eq!(scheduler.exit_current(0xa), Ok(Some(b)));
        table.channel_mut(second).unwrap().receiver_wait = Some(c);
        assert_eq!(
            table.reap_exited_task(a, &mut scheduler),
            Err(Error::WaitState)
        );
        assert_eq!(table.queued(first), Ok(1));
        assert_eq!(table.queued(second), Ok(1));
        assert_eq!(table.live_channels(), 2);
        table.channel_mut(second).unwrap().receiver_wait = None;
        assert_eq!(
            table.reap_exited_task(a, &mut scheduler),
            Ok(ReapedChannels {
                ids: [first, second],
                count: 2,
                discarded_messages: 2,
                woken_peers: 0,
            })
        );
        assert_eq!(table.live_channels(), 0);
    }

    #[test]
    fn k301_channel_ids_never_reopen_a_stale_number() {
        let mut table = ChannelTable::new();
        let first = table.create(A, B).unwrap();
        assert_eq!(table.destroy(first, C), Err(Error::WrongProducer));
        assert_eq!(table.destroy(first, A), Ok(()));
        let second = table.create(A, B).unwrap();
        assert_eq!((first, second), (1, 2));
        assert_eq!(table.send(first, A, b"old"), Err(Error::InvalidChannel));
        assert_eq!(table.receive(first, B), Err(Error::InvalidChannel));
        assert_eq!(table.queued(second), Ok(0));
        assert_eq!(table.create(A, A), Err(Error::SameEndpoint));
        assert_eq!(table.live_channels(), 1);
    }

    #[test]
    fn k302_single_producer_consumer_preserves_fifo_and_ownership() {
        let mut table = ChannelTable::new();
        let id = table.create(A, B).unwrap();
        assert_eq!(table.send(id, C, b"foreign"), Err(Error::WrongProducer));
        assert_eq!(table.send(id, A, b"ONE"), Ok(()));
        assert_eq!(table.send(id, A, b"TWO"), Ok(()));
        assert_eq!(table.receive(id, C), Err(Error::WrongConsumer));
        assert_eq!(table.queued(id), Ok(2));
        assert_eq!(table.receive(id, B).unwrap().bytes(), b"ONE");
        assert_eq!(table.receive(id, B).unwrap().bytes(), b"TWO");
        assert_eq!(table.receive(id, B), Err(Error::QueueEmpty));
    }

    #[test]
    fn k303_message_limit_rejects_oversize_without_changing_queue() {
        let mut table = ChannelTable::new();
        let id = table.create(A, B).unwrap();
        assert_eq!(table.send(id, A, &[0x58; MAX_MESSAGE_BYTES]), Ok(()));
        assert_eq!(
            table.send(id, A, &[0x59; MAX_MESSAGE_BYTES + 1]),
            Err(Error::TooLarge)
        );
        assert_eq!(table.queued(id), Ok(1));
        assert_eq!(
            table.receive(id, B).unwrap().bytes(),
            &[0x58; MAX_MESSAGE_BYTES]
        );
    }

    #[test]
    fn k304_zero_length_message_is_a_real_fifo_entry() {
        let mut table = ChannelTable::new();
        let id = table.create(A, B).unwrap();
        assert_eq!(table.send(id, A, b""), Ok(()));
        assert_eq!(table.queued(id), Ok(1));
        let zero = table.receive(id, B).unwrap();
        assert!(zero.is_empty());
        assert_eq!(zero.len(), 0);
        assert_eq!(zero.bytes(), b"");
        assert_eq!(table.receive(id, B), Err(Error::QueueEmpty));
    }

    #[test]
    fn k305_missing_receiver_rejects_send_without_enqueue() {
        let mut table = ChannelTable::new();
        let id = table.create(A, B).unwrap();
        assert_eq!(table.close_receiver(id, C), Err(Error::WrongConsumer));
        assert_eq!(table.close_receiver(id, B), Ok(()));
        assert_eq!(table.send(id, A, b"X"), Err(Error::NoReceiver));
        assert_eq!(table.queued(id), Ok(0));
        assert_eq!(table.close_receiver(id, B), Err(Error::NoReceiver));
    }

    #[test]
    fn k320_timeline_tracks_order_and_exact_latency_extrema() {
        let mut timeline = MessageTimeline::new();
        assert_eq!(timeline.summary(), (0, 0, 0, 0));
        assert_eq!(timeline.record(1, 10, 16), Ok(6));
        assert_eq!(timeline.record(2, 17, 27), Ok(10));
        assert_eq!(timeline.record(3, 27, 31), Ok(4));
        assert_eq!(timeline.record(4, 40, 45), Ok(5));
        assert_eq!(timeline.record(5, 46, 54), Ok(8));
        assert_eq!(timeline.summary(), (5, 4, 10, 33));
        assert_eq!(timeline.record(6, 55, 60), Err(TimelineError::Capacity));
        assert_eq!(timeline.summary(), (5, 4, 10, 33));
    }

    #[test]
    fn k320_timeline_rejects_gap_reverse_and_overlap_without_partial_update() {
        let mut timeline = MessageTimeline::new();
        assert_eq!(timeline.record(2, 10, 20), Err(TimelineError::Order));
        assert_eq!(timeline.record(1, 20, 20), Err(TimelineError::Clock));
        assert_eq!(timeline.record(1, 21, 20), Err(TimelineError::Clock));
        assert_eq!(timeline.summary(), (0, 0, 0, 0));
        assert_eq!(timeline.record(1, 10, 20), Ok(10));
        assert_eq!(timeline.record(1, 21, 22), Err(TimelineError::Order));
        assert_eq!(timeline.record(2, 19, 22), Err(TimelineError::Clock));
        assert_eq!(timeline.summary(), (1, 10, 10, 10));
        assert_eq!(timeline.record(2, 20, 25), Ok(5));
        assert_eq!(timeline.summary(), (2, 5, 10, 15));
    }

    #[test]
    fn k319_four_slot_ring_survives_eight_full_empty_cycles() {
        let mut table = ChannelTable::new();
        let id = table.create(A, B).unwrap();
        for cycle in 0..8u8 {
            for offset in 0..4u8 {
                assert_eq!(table.send(id, A, &[cycle * 4 + offset]), Ok(()));
            }
            assert_eq!(table.queued(id), Ok(4));
            assert_eq!(table.send(id, A, b"X"), Err(Error::QueueFull));
            for offset in 0..4u8 {
                assert_eq!(table.receive(id, B).unwrap().bytes(), &[cycle * 4 + offset]);
            }
            assert_eq!(table.receive(id, B), Err(Error::QueueEmpty));
            assert_eq!(table.queued(id), Ok(0));
        }
        assert_eq!(table.destroy(id, A), Ok(()));
        assert_eq!(table.live_channels(), 0);
    }

    #[test]
    fn channel_capacity_and_id_overflow_are_failure_atomic() {
        let mut table = ChannelTable::new();
        let first = table.create(A, B).unwrap();
        let second = table.create(A, C).unwrap();
        assert_eq!(table.create(B, C), Err(Error::Full));
        assert_eq!(table.live_channels(), 2);
        assert_eq!(table.destroy(first, A), Ok(()));
        table.next_id = u64::MAX;
        assert_eq!(table.create(B, C), Err(Error::IdExhausted));
        assert_eq!(table.live_channels(), 1);
        assert_eq!(table.queued(second), Ok(0));
    }
}