PicoBot/src/gateway/outbound_dispatcher.rs

use std::collections::HashMap;
use std::sync::Arc;

use tokio::sync::{RwLock, mpsc};
use tokio_util::sync::CancellationToken;

use crate::bus::message::OutboundEventKind;
use crate::bus::{MessageBus, OutboundMessage};
use crate::channels::base::{Channel, ChannelError};

/// 低优先级队列容量（ToolCall / ToolResult / StreamDelta 等中间过程事件）。
///
/// 略小于 MessageBus 的容量，确保 bus 的 `try_send` 丢消息防线仍有效——
/// channel 队列满时 dispatcher 立即丢弃该消息，不阻塞路由循环影响其他 channel。
const LOW_PRIORITY_QUEUE_CAPACITY: usize = 64;

/// 高优先级队列容量（仅 AssistantResponse 最终响应）。
///
/// 单次 agent 执行仅产生 1 条最终响应；32 足够缓冲多 topic 并发的最终响应，
/// 极罕见满。最终响应是 agent 与用户的契约，必达。
const HIGH_PRIORITY_QUEUE_CAPACITY: usize = 32;

/// 低优先级事件的发送重试间隔（秒）。
const RETRY_DELAYS_SECS: [u64; 3] = [1, 2, 4];

/// 高优先级事件（最终响应）的发送重试间隔（秒）——更多次、更长退避，
/// 尽最大努力送达最终响应。
const EXTENDED_RETRY_DELAYS_SECS: [u64; 5] = [1, 2, 4, 8, 16];

/// Prefix for virtual scheduler chat IDs that should not be sent to external channels.
const SCHEDULER_VIRTUAL_CHAT_ID_PREFIX: &str = "scheduler/";

/// 判断消息是否为高优先级（必达）。
///
/// `AssistantResponse`（最终响应）和 `ErrorNotification`（agent 异常终止的错误通知）
/// 都是 agent 与用户的终态契约，必须送达；其他事件（工具调用进度、流式增量、
/// 执行完成信号等）是中间过程，可丢。
fn is_high_priority(msg: &OutboundMessage) -> bool {
    matches!(
        msg.event_kind,
        OutboundEventKind::AssistantResponse | OutboundEventKind::ErrorNotification
    )
}

/// 单个 channel 的发送上下文：双优先级 mpsc 队列 + sender task。
///
/// dispatcher 按消息优先级 `try_send` 到 `high_tx`（最终响应，必达）或
/// `low_tx`（中间过程，可丢）；`sender_task` 优先消费 high 队列，再消费
/// low 队列，串行调用 `Channel::send`（含重试）。channel 之间完全隔离——
/// 某个 channel 的慢发送或重试 sleep 不会阻塞其他 channel 的消息投递。
#[derive(Clone)]
struct ChannelSink {
    high_tx: mpsc::Sender<OutboundMessage>,
    low_tx: mpsc::Sender<OutboundMessage>,
    cancel: CancellationToken,
}

/// Consumes outbound messages from MessageBus and dispatches them to channels.
///
/// 架构：dispatcher 主循环只负责路由（O(1) try_send），不参与发送。
/// 每个 channel 拥有独立的 sender task 和有界队列，实现 channel 级隔离。
pub struct OutboundDispatcher {
    bus: Arc<MessageBus>,
    channels: Arc<RwLock<HashMap<String, ChannelSink>>>,
}

impl OutboundDispatcher {
    pub fn new(bus: Arc<MessageBus>) -> Self {
        Self {
            bus,
            channels: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 注册 channel 并启动其独立 sender task。
    ///
    /// sender task 生命周期与 dispatcher 一致：dispatcher `run()` 退出时
    /// 通过 cancel token 终止所有 sender task。
    pub async fn register_channel(&self, name: &str, channel: Arc<dyn Channel + Send + Sync>) {
        let (high_tx, high_rx) =
            mpsc::channel::<OutboundMessage>(HIGH_PRIORITY_QUEUE_CAPACITY);
        let (low_tx, low_rx) =
            mpsc::channel::<OutboundMessage>(LOW_PRIORITY_QUEUE_CAPACITY);
        let cancel = CancellationToken::new();

        let channel_name = name.to_string();
        let cancel_for_task = cancel.clone();

        tokio::spawn(async move {
            Self::run_sender_task(
                &channel_name,
                channel,
                high_rx,
                low_rx,
                cancel_for_task,
            )
            .await;
        });

        self.channels
            .write()
            .await
            .insert(
                name.to_string(),
                ChannelSink {
                    high_tx,
                    low_tx,
                    cancel,
                },
            );
    }

    /// sender task：优先消费 high 队列（最终响应），再消费 low 队列（中间过程），
    /// 串行调用 `Channel::send` 并重试。
    ///
    /// 重试 sleep 只阻塞当前 channel 的 task，不影响其他 channel。
    async fn run_sender_task(
        channel_name: &str,
        channel: Arc<dyn Channel + Send + Sync>,
        mut high_rx: mpsc::Receiver<OutboundMessage>,
        mut low_rx: mpsc::Receiver<OutboundMessage>,
        cancel: CancellationToken,
    ) {
        tracing::info!(channel = %channel_name, "Channel sender task started");

        loop {
            // 优先消费 high 队列：保证最终响应优先发送，不被中间过程阻塞。
            // try_recv 非阻塞，有则立即处理，无则进入 select! 等待。
            if let Ok(msg) = high_rx.try_recv() {
                Self::send_one(&*channel, channel_name, msg).await;
                continue;
            }

            tokio::select! {
                // high 优先：一旦有最终响应立即处理
                msg = high_rx.recv() => {
                    match msg {
                        Some(msg) => Self::send_one(&*channel, channel_name, msg).await,
                        None => {
                            // high 关闭：仅消费 low 残留后退出
                            tracing::debug!(channel = %channel_name, "High-priority queue closed, draining low queue");
                            Self::drain_low(&*channel, channel_name, &mut low_rx).await;
                            break;
                        }
                    }
                }
                // low：high 空时消费中间过程
                msg = low_rx.recv() => {
                    match msg {
                        Some(msg) => Self::send_one(&*channel, channel_name, msg).await,
                        None => {
                            // low 关闭：仅消费 high 残留后退出
                            tracing::debug!(channel = %channel_name, "Low-priority queue closed, draining high queue");
                            Self::drain_high(&*channel, channel_name, &mut high_rx).await;
                            break;
                        }
                    }
                }
                // dispatcher 退出时取消所有 sender task
                _ = cancel.cancelled() => {
                    tracing::info!(channel = %channel_name, "Sender task cancelled, stopping");
                    break;
                }
            }
        }
    }

    /// 发送单条消息，处理重试结果日志。
    async fn send_one(
        channel: &dyn Channel,
        channel_name: &str,
        msg: OutboundMessage,
    ) {
        match Self::send_with_retry(channel, msg).await {
            Ok(()) => {}
            Err(ChannelError::ChannelFull) => {
                // 队列满是不可重试的——send_with_retry 已跳过重试。
                // 记 warn 而非 error：这是预期的背压丢弃。
                tracing::warn!(
                    channel = %channel_name,
                    "Message dropped: channel queue full"
                );
            }
            Err(error) => {
                tracing::error!(
                    channel = %channel_name,
                    error = %error,
                    "Failed to send message after retries"
                );
            }
        }
    }

    /// 排空 high 队列残留消息后返回。
    async fn drain_high(
        channel: &dyn Channel,
        channel_name: &str,
        high_rx: &mut mpsc::Receiver<OutboundMessage>,
    ) {
        while let Ok(msg) = high_rx.try_recv() {
            Self::send_one(channel, channel_name, msg).await;
        }
    }

    /// 排空 low 队列残留消息后返回。
    async fn drain_low(
        channel: &dyn Channel,
        channel_name: &str,
        low_rx: &mut mpsc::Receiver<OutboundMessage>,
    ) {
        while let Ok(msg) = low_rx.try_recv() {
            Self::send_one(channel, channel_name, msg).await;
        }
    }

    pub async fn run(&self) {
        tracing::info!("OutboundDispatcher started");

        loop {
            let msg = match self.bus.consume_outbound().await {
                Some(msg) => msg,
                None => {
                    tracing::info!("Outbound bus closed, stopping dispatcher");
                    break;
                }
            };

            #[cfg(debug_assertions)]
            tracing::debug!(
                channel = %msg.channel,
                chat_id = %msg.chat_id,
                content_len = msg.content.len(),
                "OutboundDispatcher received message"
            );

            // Skip messages with virtual scheduler chat IDs (e.g., "scheduler/job_id")
            // These are internal messages from SilentAgentTask that should not be sent externally
            if msg.chat_id.starts_with(SCHEDULER_VIRTUAL_CHAT_ID_PREFIX) {
                #[cfg(debug_assertions)]
                tracing::debug!(
                    channel = %msg.channel,
                    chat_id = %msg.chat_id,
                    "Skipping message with virtual scheduler chat_id"
                );
                continue;
            }

            let channel_name = msg.channel.clone();
            let sink = self.channels.read().await.get(&channel_name).cloned();

            match sink {
                Some(sink) => {
                    // try_send 保证 dispatcher 永不阻塞：队列满时立即丢弃该消息，
                    // 不影响其他 channel 的投递。与 bus.publish_outbound 策略一致。
                    // 高优先级（最终响应）投递到 high 队列——容量 32 且仅最终响应，
                    // 极罕见满；低优先级（中间过程）投递到 low 队列，满则丢弃。
                    let (queue, priority_label) = if is_high_priority(&msg) {
                        (&sink.high_tx, "high")
                    } else {
                        (&sink.low_tx, "low")
                    };
                    match queue.try_send(msg) {
                        Ok(()) => {}
                        Err(mpsc::error::TrySendError::Full(_)) => {
                            tracing::warn!(
                                channel = %channel_name,
                                priority = priority_label,
                                "Channel queue full, dropping message"
                            );
                        }
                        Err(mpsc::error::TrySendError::Closed(_)) => {
                            tracing::warn!(
                                channel = %channel_name,
                                priority = priority_label,
                                "Channel queue closed, dropping message"
                            );
                        }
                    }
                }
                None => {
                    tracing::warn!(channel = %channel_name, "No channel found for message");
                }
            }
        }

        // 通知所有 sender task 退出
        let sinks = self.channels.write().await;
        for (name, sink) in sinks.iter() {
            sink.cancel.cancel();
            tracing::debug!(channel = %name, "Cancelled sender task");
        }
    }

    /// 发送消息，失败时按重试间隔重试。
    ///
    /// 高优先级消息（最终响应）使用 `EXTENDED_RETRY_DELAYS_SECS`（5 次，最长 16 秒），
    /// 尽最大努力送达；低优先级消息（中间过程）使用 `RETRY_DELAYS_SECS`（3 次）。
    ///
    /// `ChannelFull` 不可重试——队列满时重试只会浪费退避时间并阻塞
    /// channel 队列消费。立即返回让 sender task 尽快处理下一条消息。
    ///
    /// 仅在单个 channel 的 sender task 内执行——重试 sleep 只阻塞
    /// 该 channel 的发送，不影响其他 channel。
    async fn send_with_retry(
        channel: &dyn Channel,
        msg: OutboundMessage,
    ) -> Result<(), ChannelError> {
        let delays: &[u64] = if is_high_priority(&msg) {
            &EXTENDED_RETRY_DELAYS_SECS
        } else {
            &RETRY_DELAYS_SECS
        };
        for (attempt_index, delay) in delays.iter().enumerate() {
            match channel.send(msg.clone()).await {
                Ok(()) => return Ok(()),
                // 队列满：不可重试，立即返回
                Err(ChannelError::ChannelFull) => return Err(ChannelError::ChannelFull),
                Err(error) if attempt_index < delays.len() - 1 => {
                    tracing::warn!(
                        attempt = attempt_index + 1,
                        delay = delay,
                        high_priority = is_high_priority(&msg),
                        error = %error,
                        "Send failed, retrying"
                    );
                    tokio::time::sleep(tokio::time::Duration::from_secs(*delay)).await;
                }
                Err(error) => return Err(error),
            }
        }

        // 防御性兜底：正常情况下循环最后一次迭代会 return Err(error)。
        // 若 delays 被改为空数组，循环体不执行，返回错误而非 panic。
        Err(ChannelError::SendError(
            "send_with_retry exhausted with no retry attempts configured".into(),
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bus::OutboundMessage;
    use async_trait::async_trait;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::time::Duration;

    /// 测试用 channel：记录所有收到的消息内容，可配置人为延迟和失败。
    struct TestChannel {
        name: String,
        received: Arc<AtomicU32>,
        /// 按发送顺序记录每条成功发送消息的 content，用于断言优先级顺序。
        received_contents: Arc<std::sync::Mutex<Vec<String>>>,
        delay_ms: u64,
        fail_first_n: u32,
        /// 始终返回 `ChannelFull`，用于验证 `send_with_retry` 不重试队列满错误。
        always_full: bool,
        call_count: Arc<AtomicU32>,
    }

    impl TestChannel {
        fn new(name: &str) -> Self {
            Self {
                name: name.to_string(),
                received: Arc::new(AtomicU32::new(0)),
                received_contents: Arc::new(std::sync::Mutex::new(Vec::new())),
                delay_ms: 0,
                fail_first_n: 0,
                always_full: false,
                call_count: Arc::new(AtomicU32::new(0)),
            }
        }

        fn with_delay(mut self, ms: u64) -> Self {
            self.delay_ms = ms;
            self
        }

        fn with_fail_first_n(mut self, n: u32) -> Self {
            self.fail_first_n = n;
            self
        }

        fn with_channel_full(mut self) -> Self {
            self.always_full = true;
            self
        }
    }

    #[async_trait]
    impl Channel for TestChannel {
        fn name(&self) -> &str {
            &self.name
        }

        fn is_running(&self) -> bool {
            true
        }

        async fn start(&self, _bus: Arc<MessageBus>) -> Result<(), ChannelError> {
            Ok(())
        }

        async fn stop(&self) -> Result<(), ChannelError> {
            Ok(())
        }

        async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> {
            let count = self.call_count.fetch_add(1, Ordering::SeqCst);

            // 队列满：不可重试错误，用于验证 send_with_retry 立即返回
            if self.always_full {
                return Err(ChannelError::ChannelFull);
            }

            if (count as u32) < self.fail_first_n {
                return Err(ChannelError::SendError("simulated failure".to_string()));
            }

            if self.delay_ms > 0 {
                tokio::time::sleep(Duration::from_millis(self.delay_ms)).await;
            }

            self.received.fetch_add(1, Ordering::SeqCst);
            if let Ok(mut guard) = self.received_contents.lock() {
                guard.push(msg.content.clone());
            }
            Ok(())
        }
    }

    fn make_message(channel: &str, chat_id: &str, content: &str) -> OutboundMessage {
        OutboundMessage::assistant(
            channel,
            chat_id,
            None,
            content,
            None,
            std::collections::HashMap::new(),
        )
    }

    /// 构造低优先级消息（ToolCall），用于验证 low 队列的丢弃与优先级行为。
    fn make_low_message(channel: &str, chat_id: &str, content: &str) -> OutboundMessage {
        OutboundMessage::tool_call(
            channel,
            chat_id,
            None,
            "msg-id",
            content,
            serde_json::json!({}),
            None,
            std::collections::HashMap::new(),
        )
    }

    /// 构造错误通知（agent 异常终止），用于验证 ErrorNotification 走高优队列。
    fn make_error_message(channel: &str, chat_id: &str, content: &str) -> OutboundMessage {
        OutboundMessage::error_notification(
            channel,
            chat_id,
            None,
            content,
            None,
            std::collections::HashMap::new(),
        )
    }

    #[test]
    fn test_is_high_priority_classifies_terminal_events() {
        // AssistantResponse 和 ErrorNotification 都是终态,必须走高优队列必达。
        let assistant = make_message("c", "chat", "final");
        let error = make_error_message("c", "chat", "agent failed");
        let tool_call = make_low_message("c", "chat", "calling tool");
        let tool_result = OutboundMessage::tool_result(
            "c", "chat", None, "id", "tool", "result", None,
            std::collections::HashMap::new(),
        );
        let exec_done = OutboundMessage::execution_completed(
            "c", "chat", None,
            std::collections::HashMap::new(),
        );

        assert!(is_high_priority(&assistant), "AssistantResponse should be high priority");
        assert!(is_high_priority(&error), "ErrorNotification should be high priority");
        assert!(!is_high_priority(&tool_call), "ToolCall should be low priority");
        assert!(!is_high_priority(&tool_result), "ToolResult should be low priority");
        assert!(!is_high_priority(&exec_done), "ExecutionCompleted should be low priority");
    }

    #[tokio::test]
    async fn test_fast_channel_not_blocked_by_slow_channel() {
        // 验证核心目标：channel A 慢发送不应阻塞 channel B 的消息投递
        let bus = MessageBus::new(16);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        let slow = Arc::new(TestChannel::new("slow").with_delay(500));
        let fast = Arc::new(TestChannel::new("fast"));

        let slow_received = slow.received.clone();
        let fast_received = fast.received.clone();

        dispatcher.register_channel("slow", slow).await;
        dispatcher.register_channel("fast", fast).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // 先发一条 slow（500ms 延迟），紧接着发一条 fast
        bus.publish_outbound(make_message("slow", "chat-1", "slow-msg")).await.unwrap();
        bus.publish_outbound(make_message("fast", "chat-2", "fast-msg")).await.unwrap();

        // 等待 fast 消息被投递（远早于 slow 完成）
        tokio::time::timeout(Duration::from_millis(200), async {
            while fast_received.load(Ordering::SeqCst) == 0 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("fast channel should receive message within 200ms, but was blocked by slow channel");

        // 等待 slow 消息完成
        tokio::time::timeout(Duration::from_secs(2), async {
            while slow_received.load(Ordering::SeqCst) == 0 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("slow channel should eventually receive message");

        assert_eq!(fast_received.load(Ordering::SeqCst), 1);
        assert_eq!(slow_received.load(Ordering::SeqCst), 1);

        // 关闭 bus 让 dispatcher 退出
        // dispatcher 持有 Arc<MessageBus> 克隆，drop(bus) 不会关闭 bus。
        // 直接 abort dispatcher 及其 sender task 即可清理。
        dispatcher_handle.abort();
    }

    #[tokio::test]
    async fn test_retry_does_not_block_other_channel() {
        // 验证：channel A 重试 sleep（1+2=3秒）期间，channel B 正常投递
        let bus = MessageBus::new(16);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        let flaky = Arc::new(
            TestChannel::new("flaky")
                .with_fail_first_n(2) // 前 2 次失败，触发 1+2 秒重试
                .with_delay(0),
        );
        let stable = Arc::new(TestChannel::new("stable"));

        let flaky_received = flaky.received.clone();
        let stable_received = stable.received.clone();

        dispatcher.register_channel("flaky", flaky).await;
        dispatcher.register_channel("stable", stable).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // 先发 flaky（会重试 3 秒），紧接着发 stable
        bus.publish_outbound(make_message("flaky", "chat-1", "flaky-msg")).await.unwrap();
        bus.publish_outbound(make_message("stable", "chat-2", "stable-msg")).await.unwrap();

        // stable 应在 200ms 内收到，远早于 flaky 的 3 秒重试完成
        tokio::time::timeout(Duration::from_millis(200), async {
            while stable_received.load(Ordering::SeqCst) == 0 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("stable channel should not be blocked by flaky channel's retry sleep");

        // 等待 flaky 重试成功（第 3 次尝试）
        tokio::time::timeout(Duration::from_secs(5), async {
            while flaky_received.load(Ordering::SeqCst) == 0 {
                tokio::time::sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("flaky channel should eventually succeed after retries");

        assert_eq!(stable_received.load(Ordering::SeqCst), 1);
        assert_eq!(flaky_received.load(Ordering::SeqCst), 1);

        // dispatcher 持有 Arc<MessageBus> 克隆，drop(bus) 不会关闭 bus。
        // 直接 abort dispatcher 及其 sender task 即可清理。
        dispatcher_handle.abort();
    }

    #[tokio::test]
    async fn test_scheduler_virtual_chat_id_skipped() {
        // 验证：scheduler/ 前缀的 chat_id 不被投递到任何 channel
        let bus = MessageBus::new(16);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        let channel = Arc::new(TestChannel::new("test"));
        let received = channel.received.clone();

        dispatcher.register_channel("test", channel).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // scheduler 虚拟消息应被跳过
        bus.publish_outbound(make_message("test", "scheduler/job-1", "internal"))
            .await
            .unwrap();
        // 正常消息应被投递
        bus.publish_outbound(make_message("test", "chat-1", "normal"))
            .await
            .unwrap();

        tokio::time::timeout(Duration::from_secs(2), async {
            while received.load(Ordering::SeqCst) == 0 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("normal message should be delivered");

        // 只收到 1 条（scheduler 虚拟消息被跳过）
        assert_eq!(received.load(Ordering::SeqCst), 1);

        // dispatcher 持有 Arc<MessageBus> 克隆，drop(bus) 不会关闭 bus。
        // 直接 abort dispatcher 及其 sender task 即可清理。
        dispatcher_handle.abort();
    }

    #[tokio::test]
    async fn test_unknown_channel_warns_and_continues() {
        // 验证：未知 channel 的消息被跳过，不影响后续消息投递
        let bus = MessageBus::new(16);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        let channel = Arc::new(TestChannel::new("known"));
        let received = channel.received.clone();

        dispatcher.register_channel("known", channel).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // 发往未知 channel 的消息
        bus.publish_outbound(make_message("unknown", "chat-1", "lost"))
            .await
            .unwrap();
        // 发往已知 channel 的消息
        bus.publish_outbound(make_message("known", "chat-2", "delivered"))
            .await
            .unwrap();

        tokio::time::timeout(Duration::from_secs(2), async {
            while received.load(Ordering::SeqCst) == 0 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("known channel should receive message despite preceding unknown channel message");

        assert_eq!(received.load(Ordering::SeqCst), 1);

        // dispatcher 持有 Arc<MessageBus> 克隆，drop(bus) 不会关闭 bus。
        // 直接 abort dispatcher 及其 sender task 即可清理。
        dispatcher_handle.abort();
    }

    #[tokio::test]
    async fn test_send_with_retry_no_retry_on_channel_full() {
        // 验证：send_with_retry 遇到 ChannelFull 应立即返回，不重试。
        // 若错误地重试，会 sleep 1+2+4=7 秒，测试将在超时阈值内失败。
        let channel = TestChannel::new("full").with_channel_full();
        let call_count = channel.call_count.clone();

        let msg = make_message("full", "chat-1", "dropped");

        // 500ms 阈值：远小于首次重试间隔 1s，足以区分"立即返回"与"至少一次重试"
        let result = tokio::time::timeout(Duration::from_millis(500), async {
            OutboundDispatcher::send_with_retry(&channel, msg).await
        })
        .await
        .expect("send_with_retry should return immediately on ChannelFull, not retry");

        // 必须返回 ChannelFull 错误
        assert!(
            matches!(result, Err(ChannelError::ChannelFull)),
            "expected ChannelFull error, got {:?}",
            result
        );

        // send 只被调用一次——证明没有重试
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            1,
            "send should be called exactly once when ChannelFull is returned"
        );
    }

    #[tokio::test]
    async fn test_high_priority_not_dropped_when_low_full() {
        // 验证:low 队列被大量低优消息填满时,高优(AssistantResponse)仍被发送。
        // 这是本次修复的核心目标——最终响应必达,不被中间过程挤占丢弃。
        let bus = MessageBus::new(256);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        // 慢 channel:50ms/条,确保 low 队列持续积压
        let channel = Arc::new(TestChannel::new("test").with_delay(50));
        let contents = channel.received_contents.clone();
        dispatcher.register_channel("test", channel).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // 投递大量低优消息(超过 low 容量 64),填满 low 队列
        for i in 0..80 {
            bus.publish_outbound(make_low_message("test", "chat-1", &format!("low-{i}")))
                .await
                .unwrap();
        }
        // 投递高优消息(最终响应)
        bus.publish_outbound(make_message("test", "chat-1", "HIGH-FINAL"))
            .await
            .unwrap();

        // 高优应在 3 秒内被发送(独立 high 队列 + sender 优先消费)
        tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                let sent = contents.lock().unwrap().clone();
                if sent.iter().any(|c| c == "HIGH-FINAL") {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("high priority message should be sent even when low queue is full");

        dispatcher_handle.abort();
    }

    #[tokio::test]
    async fn test_low_priority_dropped_when_full() {
        // 验证:low 队列满时,后续低优消息被丢弃(不进入 high 队列)。
        // 低优是中间过程,可丢——这是与高优必达的对比行为。
        let bus = MessageBus::new(256);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        // 慢 channel:50ms/条,确保 low 队列积压触发丢弃
        let channel = Arc::new(TestChannel::new("test").with_delay(50));
        let received = channel.received.clone();
        dispatcher.register_channel("test", channel).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // 投递远超 low 容量的低优消息(200 条),确保触发丢弃
        let total_sent: u32 = 200;
        for i in 0..total_sent {
            bus.publish_outbound(make_low_message("test", "chat-1", &format!("low-{i}")))
                .await
                .unwrap();
        }

        // 等 sender 消费完 low 队列内的消息(至少 LOW_PRIORITY_QUEUE_CAPACITY 条)
        tokio::time::timeout(Duration::from_secs(10), async {
            while received.load(Ordering::SeqCst) < LOW_PRIORITY_QUEUE_CAPACITY as u32 {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("should receive at least LOW_PRIORITY_QUEUE_CAPACITY messages");

        // 额外等待确认 sender 已消费完 low 队列残留(64 * 50ms ≈ 3.2s,给足 1s 余量)
        tokio::time::sleep(Duration::from_secs(1)).await;
        let final_received = received.load(Ordering::SeqCst);

        // 断言:有丢弃发生(received < 投递数),且 low 队列曾被填满(received >= 容量)。
        // 不断言精确数量——sender 与 dispatcher 的并发竞态会使进入 low 的条数略多于容量。
        assert!(
            final_received < total_sent,
            "some low-priority messages should be dropped when queue full, got {}/{}",
            final_received,
            total_sent
        );
        assert!(
            final_received >= LOW_PRIORITY_QUEUE_CAPACITY as u32,
            "should receive at least {} messages (queue was filled), got {}",
            LOW_PRIORITY_QUEUE_CAPACITY,
            final_received
        );

        dispatcher_handle.abort();
    }

    #[tokio::test]
    async fn test_high_priority_uses_extended_retry() {
        // 验证:高优消息(AssistantResponse)使用 EXTENDED_RETRY_DELAYS(5次)。
        // fail_first_n(3):前 3 次失败,第 4 次成功——证明高优在 3 次后仍继续重试。
        let channel = TestChannel::new("flaky").with_fail_first_n(3);
        let call_count = channel.call_count.clone();
        let msg = make_message("flaky", "chat-1", "final"); // 高优

        // EXTENDED_RETRY_DELAYS = [1,2,4,8,16];前 3 次失败后第 4 次成功,耗时 1+2+4=7s
        let result = tokio::time::timeout(Duration::from_secs(15), async {
            OutboundDispatcher::send_with_retry(&channel, msg).await
        })
        .await
        .expect("high priority should succeed within extended retry budget");

        assert!(result.is_ok(), "high priority should succeed after 4 attempts");
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            4,
            "high priority should attempt 4 times (extended retry)"
        );
    }

    #[tokio::test]
    async fn test_low_priority_gives_up_after_standard_retries() {
        // 验证:低优消息(ToolCall)使用 RETRY_DELAYS(3次),前 3 次失败后放弃。
        // 与高优的 5 次重试形成对比——低优是中间过程,放弃可接受。
        let channel = TestChannel::new("flaky").with_fail_first_n(3);
        let call_count = channel.call_count.clone();
        let msg = make_low_message("flaky", "chat-1", "tool"); // 低优

        // RETRY_DELAYS = [1,2,4];3 次都失败,耗时 1+2=3s
        let result = tokio::time::timeout(Duration::from_secs(10), async {
            OutboundDispatcher::send_with_retry(&channel, msg).await
        })
        .await
        .expect("low priority should give up within standard retry budget");

        assert!(result.is_err(), "low priority should fail after 3 attempts");
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            3,
            "low priority should attempt 3 times (standard retry)"
        );
    }

    #[tokio::test]
    async fn test_sender_consumes_high_first() {
        // 验证:high 队列优先于 low 队列被消费。
        // 投递多条低优后再投递高优,高优应在大部分低优之前被发送。
        let bus = MessageBus::new(64);
        let dispatcher = OutboundDispatcher::new(bus.clone());

        let channel = Arc::new(TestChannel::new("test").with_delay(30));
        let contents = channel.received_contents.clone();
        let received = channel.received.clone();
        dispatcher.register_channel("test", channel).await;

        let dispatcher_handle = tokio::spawn(async move {
            dispatcher.run().await;
        });

        // 投递 5 条低优
        for i in 0..5 {
            bus.publish_outbound(make_low_message("test", "chat-1", &format!("low-{i}")))
                .await
                .unwrap();
        }
        // 等待低优入 low 队列
        tokio::time::sleep(Duration::from_millis(10)).await;
        // 投递 1 条高优
        bus.publish_outbound(make_message("test", "chat-1", "HIGH"))
            .await
            .unwrap();

        // 等待 6 条全部发送完成
        tokio::time::timeout(Duration::from_secs(3), async {
            while received.load(Ordering::SeqCst) < 6 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("all 6 messages should be sent within 3s");

        let sent = contents.lock().unwrap().clone();
        let high_index = sent
            .iter()
            .position(|c| c == "HIGH")
            .expect("HIGH should be in sent list");
        // HIGH 应在前 3 条内:sender 完成当前低优后立即取 high,优先于剩余低优
        assert!(
            high_index < 3,
            "HIGH should be sent before most low-priority messages, got index {} in {:?}",
            high_index,
            sent
        );

        dispatcher_handle.abort();
    }
}