alkcall/src/channels/manager.rs

//! `ChannelManager` — the shared state for a channels connection
//! (ADR-039). Holds `channel_id → ChannelState`, the `MuxHandle`, and
//! the per-connection opener ledger (ADR-047 §7). The manager is
//! ALPN-blind and auth-blind (ADR-039) — it does pure byte routing.
//!
//! The `ChannelManager` is `Clone` (cheap — `Arc` internally) so the
//! `ChannelsAdapter`, the open-op wrapper, and relay logic can all
//! hold a handle.

use std::collections::{HashMap, VecDeque};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;

use bytes::Bytes;
use parking_lot::Mutex;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::debug;

use crate::protocol::wire::CallError;
use serde_json::Value;

use super::mux::{MuxHandle, OpenerLedger};
use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP};

/// Which side of a channels connection this manager belongs to.
/// Determines the ID allocation range to prevent collisions when both
/// sides open channels on the same connection (ADR-047 §5 odd/even split).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChannelSide {
    /// The connect side (initiator) — allocates odd IDs (1, 3, 5, …).
    Connect,
    /// The accept side (responder) — allocates even IDs (2, 4, 6, …).
    Accept,
}

/// The default per-connection channel cap (ADR-040) — a per-connection
/// **memory bound** (limits one connection's reassembly-buffer cost),
/// NOT a DoS defense. The per-identity DoS defense is the
/// `ChannelLifecyclePolicy` (ADR-041, amended by ADR-047 §7).
pub const DEFAULT_MAX_CHANNELS: usize = 256;

/// The per-channel out-of-band control callback (ADR-052): received
/// from the open-op establisher (`Establishment::with_control`),
/// installed by the open-op wrapper under the channel's id, routed to
/// by `channel/control`, removed by every teardown path. The manager
/// stays blind to what the closure does (ADR-039).
pub type ChannelControl = Arc<
    dyn Fn(Value) -> futures::future::BoxFuture<'static, Result<Value, CallError>> + Send + Sync,
>;

/// The per-channel state — the reassembly sender (the demux feeds chunk
/// payloads into it; the handler reads them out via `MpscRecvStream`),
/// the handler task handle, and the ALPN (for observability).
struct ChannelState {
    /// The sender the demux uses to feed chunk payloads to the
    /// handler's read half. Dropping this signals EOF to the handler
    /// (REQ-CH-02).
    demux_sender: mpsc::Sender<Bytes>,
    /// The handler task — aborted on close / connection drop.
    handler_task: Option<JoinHandle<()>>,
    /// The ALPN the channel carries (observability; the manager is
    /// ALPN-blind operationally — it doesn't parse the ALPN's protocol).
    alpn: String,
}

/// Errors raised by `ChannelManager::open_channel` and friends.
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
pub enum ManagerError {
    #[error("too many channels: {count} (max {max})")]
    TooManyChannels { count: usize, max: usize },
    #[error("unknown channel: {0}")]
    UnknownChannel(u32),
    #[error("channel already exists: {0}")]
    ChannelExists(u32),
    #[error(
        "channel {channel_id} is outside the remote side's allocation range (expected {expected})"
    )]
    InvalidAdopt {
        channel_id: u32,
        expected: &'static str,
    },
}

/// The shared state for a channels connection. `Clone` (cheap — `Arc`
/// internally). The manager is ALPN-blind and auth-blind (ADR-039);
/// the per-identity cap and the opener ledger live in `channels-call`
/// (ADR-047 §7) — the ledger is here only because the manager is the
/// per-connection state that outlives individual channels.
///
/// The `OpenerLedger` is kept here (in the manager) rather than in
/// `channels-call` because the manager is the per-connection state
/// that owns channel lifetimes. The ledger records the `PeerId` of the
/// peer that opened each channel; `channels-call` reads it on
/// teardown to call `ChannelLifecyclePolicy::on_close` with the
/// opener (not the closer). Keeping the ledger here preserves the
/// auth-blindness of `channels-core` (the manager doesn't consult the
/// policy; it just records who opened what).
#[derive(Clone)]
pub struct ChannelManager {
    inner: Arc<Inner>,
}

struct Inner {
    channels: Mutex<HashMap<u32, ChannelState>>,
    next_id: AtomicU32,
    max_channels: usize,
    buffer_cap: usize,
    mux: MuxHandle,
    opener_ledger: OpenerLedger,
    remote_addr: Option<SocketAddr>,
    side: ChannelSide,
    dropped_unknown_chunks: AtomicU64,
    /// Chunks that arrived for a channel_id this side has not
    /// adopted yet (the open-op response carrying `channel_id` and the
    /// producer's first data-plane write race — the handler can start
    /// pumping before the connect side's `adopt_channel` runs). Each
    /// channel buffers up to `early_arrival_cap` payloads; `adopt_channel`
    /// drains them into the new receiver. `clear_all` drops the map.
    early_arrivals: Mutex<HashMap<u32, VecDeque<Bytes>>>,
    early_arrival_count: AtomicU64,
    /// Per-channel out-of-band control callbacks (ADR-052). Installed
    /// by the open-op wrapper from the establisher's
    /// `Establishment::with_control`; removed by every teardown path.
    control_handles: Mutex<HashMap<u32, ChannelControl>>,
}

/// The per-channel cap on parked early-arrival chunks (`route_payload`
/// parks chunks for not-yet-adopted channels instead of dropping them —
/// the open-op response/first-data race). A channel whose adopter never
/// arrives leaks its parked chunks until `clear_all`; the cap bounds
/// that leak per channel.
///
/// Consumer-facing bound (review 006 E-04): for a push-first producer
/// (dial-then-pump is the common tunnel shape), up to `EARLY_ARRIVAL_CAP`
/// chunks per channel can be parked before the consumer's
/// `adopt_channel` runs — and chunks past the cap drop silently at the
/// consumer (data loss presents as a truncated stream with clean
/// framing elsewhere; `dropped_unknown_chunks` /
/// `early_arrival_count` are the producer-side observability). A
/// tunnel crate's sizing (e.g. a UDP POC's MTU-vs-buffer math) should
/// treat 64 parked chunks as the observable bound and adopt promptly.
const EARLY_ARRIVAL_CAP: usize = 64;

impl ChannelManager {
    /// Construct a new manager with the given `MuxHandle`, max
    /// channels, buffer cap, and side. The `remote_addr` is
    /// informational (NAT/proxy). The `side` determines the ID
    /// allocation range (odd for connect, even for accept) to prevent
    /// collisions when both sides open channels (ADR-047 §5 odd/even
    /// split).
    pub fn new(
        mux: MuxHandle,
        max_channels: usize,
        buffer_cap: usize,
        remote_addr: Option<SocketAddr>,
        side: ChannelSide,
    ) -> Self {
        let start_id = match side {
            ChannelSide::Connect => 1u32,
            ChannelSide::Accept => 2u32,
        };
        Self {
            inner: Arc::new(Inner {
                channels: Mutex::new(HashMap::new()),
                next_id: AtomicU32::new(start_id),
                max_channels,
                buffer_cap,
                mux,
                opener_ledger: OpenerLedger::new(),
                remote_addr,
                side,
                dropped_unknown_chunks: AtomicU64::new(0),
                early_arrivals: Mutex::new(HashMap::new()),
                early_arrival_count: AtomicU64::new(0),
                control_handles: Mutex::new(HashMap::new()),
            }),
        }
    }

    /// Construct with default settings (256 channels, 1 MiB buffer).
    /// Uses `ChannelSide::Accept` — the accept side is the typical
    /// constructor for the `ChannelsAdapter`. The connect side
    /// (`ChannelClient`) should use `ChannelSide::Connect`.
    pub fn with_defaults(mux: MuxHandle, remote_addr: Option<SocketAddr>) -> Self {
        Self::new(
            mux,
            DEFAULT_MAX_CHANNELS,
            DEFAULT_BUFFER_CAP,
            remote_addr,
            ChannelSide::Accept,
        )
    }

    /// The side this manager belongs to.
    pub fn side(&self) -> ChannelSide {
        self.inner.side
    }

    /// The remote address (informational — NAT/proxy).
    pub fn remote_addr(&self) -> Option<SocketAddr> {
        self.inner.remote_addr
    }

    /// The per-connection opener ledger (ADR-047 §7). `channels-call`
    /// reads this on teardown to call `ChannelLifecyclePolicy::on_close`
    /// with the opener (not the closer).
    pub fn opener_ledger(&self) -> &OpenerLedger {
        &self.inner.opener_ledger
    }

    /// The number of open channels (for tests and observability).
    pub fn open_count(&self) -> usize {
        self.inner.channels.lock().len()
    }

    /// The number of chunks the demux dropped because the `channel_id`
    /// was unknown (REQ-CH-04 — lenient handling: the demux drops and
    /// continues rather than tearing down the connection). A non-zero
    /// value indicates either a peer sending chunks for channels this
    /// side hasn't opened/adopted yet, or a desync that the lenient
    /// drop is masking. Observability only — the counter does not
    /// drive any control flow.
    pub fn dropped_unknown_chunks(&self) -> u64 {
        self.inner.dropped_unknown_chunks.load(Ordering::Relaxed)
    }

    /// The number of chunks parked in the early-arrival buffer since
    /// the connection started (monotonic — parked chunks handed to the
    /// adopter are not subtracted). Together with
    /// [`Self::dropped_unknown_chunks`] this bounds the two observable
    /// halves of the open-op-response/first-data race: how many of a
    /// push-first producer's first chunks were buffered
    /// (`early_arrival_count`) versus lost to the per-channel cap
    /// (`dropped_unknown_chunks` — see `EARLY_ARRIVAL_CAP`, review 006
    /// E-04). Observability only — neither counter drives control flow.
    pub fn early_arrival_count(&self) -> u64 {
        self.inner.early_arrival_count.load(Ordering::Relaxed)
    }

    /// The mux handle — for registering new channels' write halves.
    pub fn mux(&self) -> &MuxHandle {
        &self.inner.mux
    }

    /// Allocate a `channel_id` and install the channel state. Returns
    /// the `channel_id` and the `MpscSendStream` (the handler's write
    /// half — the mux frames the bytes onto the transport) plus the
    /// `MpscRecvStream` (the handler's read half — the demux feeds
    /// chunk payloads into it).
    ///
    /// `alpn` is the ALPN the channel carries (observability).
    /// `opener_peer_id` is the `PeerId` of the peer that opened the
    /// channel (recorded in the opener ledger, ADR-047 §7).
    ///
    /// **Allocation: ADR-047 §5 — odd/even split.** The connect side
    /// allocates odd IDs (1, 3, 5, …); the accept side allocates even
    /// IDs (2, 4, 6, …). `next_id` steps by 2 so both sides can open
    /// channels on the same connection without collision.
    ///
    /// **TOCTOU-safe:** the `max_channels` check and a reservation
    /// increment happen under one lock acquisition. The async
    /// `mux.register` happens outside the lock; on re-acquire the
    /// reservation is decremented and the insert is checked for
    /// collision. If the channel was torn down between reserve and
    /// insert, the reservation is still decremented (the teardown path
    /// already removed the entry).
    pub async fn open_channel(
        &self,
        alpn: impl Into<String>,
        opener_peer_id: impl Into<String>,
        handler_task: Option<JoinHandle<()>>,
    ) -> Result<(u32, MpscSendStream, MpscRecvStream), ManagerError> {
        let alpn = alpn.into();
        let opener = opener_peer_id.into();

        let channel_id = {
            let channels = self.inner.channels.lock();
            if channels.len() >= self.inner.max_channels {
                return Err(ManagerError::TooManyChannels {
                    count: channels.len(),
                    max: self.inner.max_channels,
                });
            }
            self.inner.next_id.fetch_add(2, Ordering::Relaxed)
        };

        let send = self
            .inner
            .mux
            .register(channel_id)
            .await
            .map_err(|_| ManagerError::ChannelExists(channel_id))?;

        let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);

        let state = ChannelState {
            demux_sender: demux_sender.clone(),
            handler_task,
            alpn,
        };
        {
            let mut channels = self.inner.channels.lock();
            if channels.len() >= self.inner.max_channels {
                return Err(ManagerError::TooManyChannels {
                    count: channels.len(),
                    max: self.inner.max_channels,
                });
            }
            if channels.contains_key(&channel_id) {
                return Err(ManagerError::ChannelExists(channel_id));
            }
            channels.insert(channel_id, state);
        }

        self.inner.opener_ledger.record(channel_id, opener);

        Ok((channel_id, send, recv))
    }

    /// Adopt a channel whose `channel_id` was allocated by the remote
    /// side (ADR-047 §5 odd/even split). The non-allocating side calls
    /// this after receiving a `channel_id` in an open-op response to
    /// install local routing state: a mux write half (so the local
    /// handler can write to the channel) and a demux read half (so
    /// incoming chunks for this `channel_id` are routed to the local
    /// handler).
    ///
    /// The `alpn` is the ALPN the channel carries (observability).
    /// The opener ledger is NOT updated — the remote side is the
    /// opener; this side is the adopter.
    ///
    /// Returns `ChannelExists` if the `channel_id` is already in use
    /// (collision — the remote side allocated an ID this side already
    /// uses). Returns `TooManyChannels` if the per-connection cap is
    /// reached.
    pub async fn adopt_channel(
        &self,
        channel_id: u32,
        alpn: impl Into<String>,
        handler_task: Option<JoinHandle<()>>,
    ) -> Result<(MpscSendStream, MpscRecvStream), ManagerError> {
        // The adopt range is the remote side's allocation range (the
        // odd/even split, ADR-047 §5): the connect side allocates odd,
        // so the accept side adopts odd; the accept side allocates
        // even, so the connect side adopts even. Adopting inside the
        // local range would collide with this side's own monotonic
        // allocator shortly after. The duplicate check comes first —
        // adopting an id this side already holds returns
        // `ChannelExists` and must never touch the live channel (the
        // 0.8.1 regression pin).
        let expected = match self.inner.side {
            ChannelSide::Connect => "even (the accept side's allocated ids; ADR-047 §5)",
            ChannelSide::Accept => "odd (the connect side's allocated ids; ADR-047 §5)",
        };
        if self.inner.channels.lock().contains_key(&channel_id) {
            return Err(ManagerError::ChannelExists(channel_id));
        }
        let parity_ok = match self.inner.side {
            ChannelSide::Connect => channel_id.is_multiple_of(2),
            ChannelSide::Accept => !channel_id.is_multiple_of(2),
        };
        if !parity_ok {
            return Err(ManagerError::InvalidAdopt {
                channel_id,
                expected,
            });
        }
        let alpn = alpn.into();

        let send = self
            .inner
            .mux
            .register(channel_id)
            .await
            .map_err(|_| ManagerError::ChannelExists(channel_id))?;

        let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);

        let state = ChannelState {
            demux_sender: demux_sender.clone(),
            handler_task,
            alpn,
        };
        {
            let mut channels = self.inner.channels.lock();
            if channels.len() >= self.inner.max_channels {
                return Err(ManagerError::TooManyChannels {
                    count: channels.len(),
                    max: self.inner.max_channels,
                });
            }
            if channels.contains_key(&channel_id) {
                return Err(ManagerError::ChannelExists(channel_id));
            }
            channels.insert(channel_id, state);
        }

        // Route any chunks that arrived between the open-op response and
        // this adopt (the producer may have started pumping already).
        self.drain_early_arrivals(channel_id, &demux_sender).await;

        Ok((send, recv))
    }

    /// Install the handler task for a channel after it has been
    /// opened via [`ChannelManager::open_channel`] with
    /// `handler_task: None`. The open-op wrapper (ADR-047 §3) allocates the channel first (to
    /// get the `MpscSendStream`/`MpscRecvStream` for the channel's
    /// `Connection`), then spawns the ALPN's `OpenHandler` on that
    /// `Connection`, then records the resulting `JoinHandle` here so
    /// `channel/close` / connection drop can abort it.
    ///
    /// Returns `UnknownChannel` if the channel was closed between
    /// `open_channel` and this call (a race — the connection dropped
    /// mid-open). The caller should treat this as the open having
    /// been superseded by teardown.
    pub fn set_handler_task(
        &self,
        channel_id: u32,
        task: JoinHandle<()>,
    ) -> Result<(), ManagerError> {
        let mut channels = self.inner.channels.lock();
        match channels.get_mut(&channel_id) {
            Some(state) => {
                state.handler_task = Some(task);
                Ok(())
            }
            None => Err(ManagerError::UnknownChannel(channel_id)),
        }
    }

    /// Install the out-of-band control callback for a channel
    /// (ADR-052). Called by the open-op wrapper from the establisher's
    /// `Establishment::with_control`. Replaces an existing callback for
    /// the id (the establisher runs once per channel — replacement is
    /// not an expected path; the last call wins).
    pub fn set_control_handle(&self, channel_id: u32, control: ChannelControl) {
        self.inner
            .control_handles
            .lock()
            .insert(channel_id, control);
    }

    /// The control callback for a channel, for routing (the
    /// `channel/control` handler). `None` when the channel's ALPN
    /// registered no control surface (the op answers
    /// `channel:no_control_handle`).
    pub fn control_handle(&self, channel_id: u32) -> Option<ChannelControl> {
        self.inner.control_handles.lock().get(&channel_id).cloned()
    }

    fn remove_control_handle(&self, channel_id: u32) {
        self.inner.control_handles.lock().remove(&channel_id);
    }

    /// Install channel 0 (pre-negotiated as `alk/call`, ADR-036).
    /// Channel 0 is special only in that it's pre-allocated (by
    /// `channels-call`); the `ChannelsAdapter` hands the resulting
    /// `Connection` to the `CallAdapter`. The `channel_id` is 0.
    ///
    /// Returns the `MpscSendStream` (the call adapter's write half) and
    /// the `MpscRecvStream` (the call adapter's read half).
    pub async fn install_channel_zero(
        &self,
        handler_task: Option<JoinHandle<()>>,
    ) -> Result<(MpscSendStream, MpscRecvStream), ManagerError> {
        let send = self
            .inner
            .mux
            .register(super::wire::CHANNEL_ID_ZERO)
            .await
            .map_err(|_| ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO))?;
        let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);
        let state = ChannelState {
            demux_sender,
            handler_task,
            alpn: "alk/call".to_string(),
        };
        {
            let mut channels = self.inner.channels.lock();
            if channels
                .insert(super::wire::CHANNEL_ID_ZERO, state)
                .is_some()
            {
                return Err(ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO));
            }
        }
        // Channel 0 is opened by the accept side itself; the opener
        // ledger records the local peer (or is left empty for channel
        // 0 — the cap doesn't apply to the pre-negotiated control
        // channel). We skip the ledger for channel 0.
        Ok((send, recv))
    }

    /// Route a chunk payload to the reassembled stream for
    /// `channel_id` (the demux's per-chunk route). A zero-length
    /// payload is the EOF sentinel — the reassembled stream interprets
    /// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload
    /// in the early-arrival buffer (up to `EARLY_ARRIVAL_CAP` payloads
    /// per channel) — the open-op response carrying `channel_id` and
    /// the producer's first data-plane write race, and a push-first
    /// producer (a TTY backend's banner, a sub protocol's greeting)
    /// must not lose its first chunks. `adopt_channel` drains the
    /// parked payloads into the new receiver. Chunks beyond the cap
    /// drop with a debug log (the adopted-side REQ-CH-04 leniency for
    /// genuinely unknown channels).
    ///
    /// Awaits the bounded channel sender — if the handler's read half
    /// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless
    /// bounded-buffer backpressure). The demux loop is the only caller;
    /// stalling it stalls all channels on this connection, which is the
    /// intended behavior (the transport is the shared resource).
    pub async fn route_payload(&self, channel_id: u32, payload: Bytes) {
        let sender = {
            let channels = self.inner.channels.lock();
            channels.get(&channel_id).map(|s| s.demux_sender.clone())
        };
        match sender {
            Some(sender) => {
                if sender.send(payload).await.is_err() {
                    debug!(
                        channel_id,
                        "demux: channel receiver dropped, dropping chunk"
                    );
                }
            }
            None => {
                self.park_early_arrival(channel_id, payload);
            }
        }
    }

    /// Park an early-arrival chunk for `channel_id` (the adopter has
    /// not run yet). FIFO per channel, bounded by `EARLY_ARRIVAL_CAP`;
    /// beyond the cap the chunk is dropped with a debug log and the
    /// dropped counter increments (observability only).
    fn park_early_arrival(&self, channel_id: u32, payload: Bytes) {
        let mut arrivals = self.inner.early_arrivals.lock();
        let queue = arrivals.entry(channel_id).or_default();
        if queue.len() >= EARLY_ARRIVAL_CAP {
            drop(arrivals);
            self.inner
                .dropped_unknown_chunks
                .fetch_add(1, Ordering::Relaxed);
            debug!(
                channel_id,
                "demux: unknown channel_id, early-arrival buffer full, dropping chunk"
            );
            return;
        }
        queue.push_back(payload);
        self.inner
            .early_arrival_count
            .fetch_add(1, Ordering::Relaxed);
        debug!(
            channel_id,
            "demux: unknown channel_id, parking chunk (early arrival, not yet adopted)"
        );
    }

    /// Drain the early-arrival buffer for `channel_id` into `recv`'s
    /// sender (called by `adopt_channel` after the channel state is
    /// installed). Returns the drained payload count. Payloads past the
    /// receiver's buffer capacity await normally (backpressure).
    async fn drain_early_arrivals(
        &self,
        channel_id: u32,
        demux_sender: &mpsc::Sender<Bytes>,
    ) -> usize {
        let payloads = self
            .inner
            .early_arrivals
            .lock()
            .remove(&channel_id)
            .unwrap_or_default();
        let count = payloads.len();
        for payload in payloads {
            if demux_sender.send(payload).await.is_err() {
                debug!(
                    channel_id,
                    "demux: channel receiver dropped while draining early arrivals"
                );
                break;
            }
        }
        count
    }

    /// Take the sender for `channel_id` — used on close / teardown to
    /// drop the sender (which signals EOF to the handler, REQ-CH-02).
    /// Returns the handler task (if any) so the caller can abort it
    /// after the drain completes. Does NOT remove the opener ledger
    /// entry — the caller (channels-call) does that atomically with
    /// the policy decrement (ADR-047 §7).
    pub fn teardown_channel(
        &self,
        channel_id: u32,
    ) -> Result<Option<JoinHandle<()>>, ManagerError> {
        let mut channels = self.inner.channels.lock();
        match channels.remove(&channel_id) {
            Some(state) => {
                // Dropping `state.demux_sender` signals EOF to the
                // handler's read half (REQ-CH-02).
                drop(state.demux_sender);
                self.remove_control_handle(channel_id);
                Ok(state.handler_task)
            }
            None => Err(ManagerError::UnknownChannel(channel_id)),
        }
    }

    /// Clear the entire channel map — used on transport EOF
    /// (REQ-CH-02). Drops all senders (every handler's read half sees
    /// EOF) and aborts all handler tasks. Returns the list of
    /// `(channel_id, opener_peer_id)` for `channels-call` to decrement
    /// the per-identity policy (ADR-047 §7).
    pub fn clear_all(&self) -> Vec<(u32, String)> {
        let mut channels = self.inner.channels.lock();
        let drained: Vec<(u32, ChannelState)> = channels.drain().collect();
        for (_, state) in &drained {
            drop(state.demux_sender.clone());
            if let Some(task) = &state.handler_task {
                task.abort();
            }
        }
        self.inner.control_handles.lock().clear();
        // The early-arrival buffers die with the connection too.
        self.inner.early_arrivals.lock().clear();
        // The opener ledger has the opener PeerIds.
        self.inner.opener_ledger.drain()
    }

    /// `true` if `channel_id` is currently open.
    pub fn has_channel(&self, channel_id: u32) -> bool {
        self.inner.channels.lock().contains_key(&channel_id)
    }

    /// The ALPN for `channel_id` (observability).
    pub fn channel_alpn(&self, channel_id: u32) -> Option<String> {
        self.inner
            .channels
            .lock()
            .get(&channel_id)
            .map(|s| s.alpn.clone())
    }

    /// The `channel_id`s of all currently-open channels (observability
    /// and snapshot iterators). The returned `Vec` is a point-in-time
    /// snapshot — channels may open or close concurrently with the
    /// caller iterating it. Used by `channel/resources/subscribe`'s
    /// initial snapshot so the loop is O(open channels) rather than
    /// O(max id) (the previous `0..u32::MAX` iteration could spin
    /// millions of times on sparse ids after churn).
    pub fn channel_ids(&self) -> Vec<u32> {
        self.inner.channels.lock().keys().copied().collect()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::channels::mux::MuxRunner;
    use tokio::io::duplex;

    /// Construct a `ChannelManager` with a live mux runner. The runner
    /// is spawned so `register` calls succeed; the task is aborted when
    /// the test ends.
    async fn make_manager_with_runner() -> ChannelManager {
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        ChannelManager::with_defaults(handle, None)
    }

    #[tokio::test]
    async fn open_channel_returns_unique_ids() {
        let manager = make_manager_with_runner().await;
        let (id1, _send1, _recv1) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open 1");
        let (id2, _send2, _recv2) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open 2");
        assert_ne!(id1, id2, "channel IDs are unique");
        assert_eq!(manager.open_count(), 2);
    }

    #[tokio::test]
    async fn open_channel_records_opener_in_ledger() {
        let manager = make_manager_with_runner().await;
        let (id, _send, _recv) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open");
        assert_eq!(
            manager.opener_ledger().take(id),
            Some("alice".to_string()),
            "opener recorded in ledger"
        );
    }

    #[tokio::test]
    async fn teardown_channel_removes_state() {
        let manager = make_manager_with_runner().await;
        let (id, _send, _recv) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open");
        assert!(manager.has_channel(id));
        let task = manager.teardown_channel(id).expect("teardown");
        assert!(task.is_none(), "no handler task was registered");
        assert!(!manager.has_channel(id), "channel removed");
    }

    #[tokio::test]
    async fn channel_ids_returns_open_channel_ids() {
        let manager = make_manager_with_runner().await;
        assert!(manager.channel_ids().is_empty(), "no channels open yet");
        let (id1, _send1, _recv1) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open 1");
        let (id2, _send2, _recv2) = manager
            .open_channel("alk/tunnel", "bob", None)
            .await
            .expect("open 2");
        let mut ids = manager.channel_ids();
        ids.sort();
        assert_eq!(ids, vec![id1, id2], "channel_ids returns all open ids");
        manager.teardown_channel(id1).expect("teardown 1");
        let ids = manager.channel_ids();
        assert_eq!(ids, vec![id2], "channel_ids reflects teardown");
    }

    #[tokio::test]
    async fn route_payload_to_open_channel_succeeds() {
        let manager = make_manager_with_runner().await;
        let (id, _send, mut recv) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open");
        manager
            .route_payload(id, Bytes::from_static(b"hello"))
            .await;
        use tokio::io::AsyncReadExt;
        let mut buf = [0u8; 5];
        recv.read_exact(&mut buf).await.expect("read");
        assert_eq!(&buf, b"hello");
    }

    #[tokio::test]
    async fn route_payload_to_unknown_channel_is_lenient() {
        let manager = make_manager_with_runner().await;
        manager
            .route_payload(999, Bytes::from_static(b"data"))
            .await;
    }

    /// Early arrivals for a not-yet-adopted channel are parked, not
    /// dropped: `adopt_channel` drains them into the new receiver. This
    /// closes the open-op-response/first-data race (a push-first
    /// producer's first chunks must not be lost).
    #[tokio::test]
    async fn route_payload_to_unknown_channel_parks_until_adopt() {
        let manager = make_manager_with_runner().await;
        manager.route_payload(7, Bytes::from_static(b"first")).await;
        manager
            .route_payload(7, Bytes::from_static(b"second"))
            .await;

        let (_send, mut recv) = manager
            .adopt_channel(7, "alk/tty", None)
            .await
            .expect("adopt");

        use tokio::io::AsyncReadExt;
        let mut buf = [0u8; 11];
        recv.read_exact(&mut buf).await.expect("read first+second");
        assert_eq!(&buf, b"firstsecond");
    }

    #[tokio::test]
    async fn route_payload_to_unknown_channel_increments_dropped_counter() {
        let manager = make_manager_with_runner().await;
        assert_eq!(manager.dropped_unknown_chunks(), 0);
        // Parked arrivals don't increment the counter...
        manager
            .route_payload(998, Bytes::from_static(b"kept"))
            .await;
        assert_eq!(
            manager.dropped_unknown_chunks(),
            0,
            "parked early arrivals are not drops"
        );
        // ...but chunks beyond the per-channel cap do (the adopted-side
        // REQ-CH-04 leniency for genuinely unknown channels).
        for i in 0..(super::EARLY_ARRIVAL_CAP + 1) {
            let n = i;
            let payload = vec![b'x'; n % 9 + 1];
            manager.route_payload(999, Bytes::from(payload)).await;
        }
        assert_eq!(
            manager.dropped_unknown_chunks(),
            1,
            "dropped_unknown_chunks counter increments when the early-arrival buffer overflows"
        );
    }

    /// `early_arrival_count` observes parked chunks (review 006 N-2 —
    /// the counter was write-only before the accessor). Monotonic by
    /// design: it counts chunks parked since connection start; draining
    /// on adopt does not subtract (paired with `dropped_unknown_chunks`
    /// it bounds the open-response/first-data race's two halves).
    #[tokio::test]
    async fn early_arrival_count_tracks_parked_chunks_monotonically() {
        let manager = make_manager_with_runner().await;
        assert_eq!(manager.early_arrival_count(), 0);
        for _ in 0..3 {
            manager
                .route_payload(7, Bytes::from_static(b"parked"))
                .await;
        }
        assert_eq!(
            manager.early_arrival_count(),
            3,
            "each parked chunk increments the counter"
        );
        // Adoption drains the buffer into the receiver but does not
        // decrement (the accessor is an observability counter, not a
        // live-depth gauge).
        let (_send, mut recv) = manager
            .adopt_channel(7, "alk/tty", None)
            .await
            .expect("adopt");
        assert_eq!(
            manager.early_arrival_count(),
            3,
            "adopt-drain does not decrement the monotonic counter"
        );
        use tokio::io::AsyncReadExt;
        let mut buf = [0u8; 6 * 3];
        recv.read_exact(&mut buf).await.expect("read parked chunks");
    }

    #[tokio::test]
    async fn route_payload_to_known_channel_does_not_increment_dropped_counter() {
        let manager = make_manager_with_runner().await;
        let (id, _send, mut recv) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open");
        manager
            .route_payload(id, Bytes::from_static(b"hello"))
            .await;
        assert_eq!(
            manager.dropped_unknown_chunks(),
            0,
            "known channel does not increment the dropped counter"
        );
        use tokio::io::AsyncReadExt;
        let mut buf = [0u8; 5];
        recv.read_exact(&mut buf).await.expect("read");
    }

    #[tokio::test]
    async fn teardown_unknown_channel_returns_error() {
        let manager = make_manager_with_runner().await;
        match manager.teardown_channel(999) {
            Err(ManagerError::UnknownChannel(999)) => {}
            other => panic!("expected UnknownChannel, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn clear_all_returns_opener_ids() {
        let manager = make_manager_with_runner().await;
        manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open");
        manager
            .open_channel("alk/tty", "bob", None)
            .await
            .expect("open");
        let drained = manager.clear_all();
        assert_eq!(drained.len(), 2);
        assert_eq!(manager.open_count(), 0);
    }

    #[test]
    fn manager_error_unknown_channel_display() {
        let e = ManagerError::UnknownChannel(7);
        assert!(format!("{e}").contains("7"));
    }

    #[test]
    fn manager_error_too_many_channels_display() {
        let e = ManagerError::TooManyChannels {
            count: 256,
            max: 256,
        };
        assert!(format!("{e}").contains("256"));
    }

    #[tokio::test]
    async fn opener_ledger_accessible_from_manager() {
        let manager = make_manager_with_runner().await;
        manager.opener_ledger().record(1, "alice".to_string());
        assert_eq!(manager.opener_ledger().take(1), Some("alice".to_string()));
    }

    #[test]
    fn default_max_channels_is_256() {
        assert_eq!(DEFAULT_MAX_CHANNELS, 256);
    }

    #[test]
    fn connect_side_starts_at_1_accept_side_starts_at_2() {
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, _runner) = MuxRunner::new(Box::new(writer));
        let connect = ChannelManager::new(handle.clone(), 256, 64, None, ChannelSide::Connect);
        let accept = ChannelManager::new(handle, 256, 64, None, ChannelSide::Accept);
        assert_eq!(
            connect.inner.next_id.load(Ordering::Relaxed),
            1,
            "connect starts at 1"
        );
        assert_eq!(
            accept.inner.next_id.load(Ordering::Relaxed),
            2,
            "accept starts at 2"
        );
    }

    #[tokio::test]
    async fn odd_even_split_no_collision() {
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (connect_handle, connect_runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = connect_runner.run().await;
        });
        let connect =
            ChannelManager::new(connect_handle.clone(), 256, 64, None, ChannelSide::Connect);
        let accept = ChannelManager::new(connect_handle, 256, 64, None, ChannelSide::Accept);

        let (id_c, _, _) = connect
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("connect open");
        let (id_a, _, _) = accept
            .open_channel("alk/tty", "bob", None)
            .await
            .expect("accept open");

        assert!(id_c % 2 == 1, "connect ID {id_c} is odd");
        assert!(id_a % 2 == 0, "accept ID {id_a} is even");
        assert_ne!(id_c, id_a, "no collision");
    }

    #[tokio::test]
    async fn adopt_channel_installs_routing() {
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);

        let (send, mut recv) = manager
            .adopt_channel(7, "alk/tty", None)
            .await
            .expect("adopt");

        manager.route_payload(7, Bytes::from_static(b"hello")).await;
        use tokio::io::AsyncReadExt;
        let mut buf = [0u8; 5];
        recv.read_exact(&mut buf).await.expect("read");
        assert_eq!(&buf, b"hello");

        drop(send);
    }

    #[tokio::test]
    async fn adopt_channel_duplicate_id_returns_channel_exists() {
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);

        manager
            .adopt_channel(7, "alk/tty", None)
            .await
            .expect("first adopt");
        match manager.adopt_channel(7, "alk/tty", None).await {
            Err(ManagerError::ChannelExists(7)) => {}
            Err(other) => panic!("expected ChannelExists, got {other}"),
            Ok(_) => panic!("expected ChannelExists, got Ok"),
        }
    }

    /// Regression (found by the `manager_routing` fuzz target): a
    /// duplicate adopt must not destroy the live channel. The pre-fix
    /// insert path (`HashMap::insert(...).is_some()`) replaced the
    /// existing `ChannelState` and dropped its `demux_sender` while
    /// returning `Err(ChannelExists)` — the live channel's readers saw
    /// a spurious EOF and subsequently routed chunks were lost.
    #[tokio::test]
    async fn adopt_channel_duplicate_id_leaves_live_channel_intact() {
        use tokio::io::AsyncReadExt;

        let (_client, server) = duplex(4096);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);

        let (_send, mut recv) = manager
            .adopt_channel(7, "alk/tty", None)
            .await
            .expect("first adopt");

        // Duplicate adopt must fail without touching the live channel.
        match manager.adopt_channel(7, "alk/tty", None).await {
            Err(ManagerError::ChannelExists(7)) => {}
            Ok(_) => panic!("expected ChannelExists, got Ok"),
            Err(other) => panic!("expected ChannelExists, got {other}"),
        }

        // The live channel still routes: both chunks arrive.
        manager.route_payload(7, Bytes::from_static(b"first")).await;
        manager
            .route_payload(7, Bytes::from_static(b"second"))
            .await;
        let mut buf = [0u8; 11];
        recv.read_exact(&mut buf).await.expect("read both chunks");
        assert_eq!(
            &buf, b"firstsecond",
            "duplicate adopt must not drop the live demux_sender"
        );

        // The live channel still sees EOF only from a real sentinel.
        manager.route_payload(7, Bytes::new()).await;
        let n = recv.read(&mut buf).await.expect("clean EOF");
        assert_eq!(n, 0, "the sentinel is the only EOF source");
    }

    /// Twin regression for `open_channel`: a duplicate open (via the
    /// same insert-replace path) must not destroy the live channel.
    #[tokio::test]
    async fn open_channel_duplicate_id_leaves_live_channel_intact() {
        use tokio::io::AsyncReadExt;

        let (_client, server) = duplex(4096);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);

        let (id, _send, mut recv) = manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("first open");

        // A second open for the same id happens when the remote side
        // allocated an id this side already uses (adopt/open overlap —
        // the exact odd/even collision adopt_channel documents). Force
        // the collision through the manager's own map path.
        let duplicate = manager.open_channel("alk/tty", "bob", None).await;
        let _ = duplicate; // distinct id — no collision via open_channel alone

        // Drive the collision directly: open_channel for the same id
        // requires mux pump reuse, so exercise the map-level invariant
        // via adopt (the reachable duplicate path).
        match manager.adopt_channel(id, "alk/tty", None).await {
            Err(ManagerError::ChannelExists(_)) => {}
            Ok(_) => panic!("expected ChannelExists, got Ok"),
            Err(other) => panic!("expected ChannelExists, got {other}"),
        }

        manager
            .route_payload(id, Bytes::from_static(b"alive"))
            .await;
        let mut buf = [0u8; 5];
        recv.read_exact(&mut buf)
            .await
            .expect("read after collision");
        assert_eq!(
            &buf, b"alive",
            "collision must not destroy the live channel"
        );
    }

    #[tokio::test]
    async fn open_channel_too_many_channels_rejected() {
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::new(handle, 2, 64, None, ChannelSide::Accept);

        manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open 1");
        manager
            .open_channel("alk/tty", "bob", None)
            .await
            .expect("open 2");
        match manager.open_channel("alk/tty", "carol", None).await {
            Err(ManagerError::TooManyChannels { count, max }) => {
                assert_eq!(count, 2);
                assert_eq!(max, 2);
            }
            Err(other) => panic!("expected TooManyChannels, got {other}"),
            Ok(_) => panic!("expected TooManyChannels, got Ok"),
        }
    }
}