From e037de88a260c3a3b70a4541cdd2ac2b9bb9f671 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Thu, 6 Aug 2026 10:01:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(agent,gateway):=20=E6=9B=BF=E6=8D=A2=20unre?= =?UTF-8?q?achable!()=20=E4=B8=BA=E9=98=B2=E5=BE=A1=E6=80=A7=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E8=BF=94=E5=9B=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 run_final_summary 和 send_with_retry 中的 unreachable!() 替换为 错误返回，避免未来重构导致循环体未返回时 panic 终止所有会话。 - agent_loop.rs: run_final_summary 兜底构造错误 AgentProcessResult， 含 emit_live_tool_call_message 推送，与 Err 不可重试路径一致 - outbound_dispatcher.rs: send_with_retry 兜底返回 ChannelError::SendError， 覆盖 RETRY_DELAYS_SECS 为空数组的边界情况 同时 outbound_dispatcher.rs 包含优先级队列重构：高优消息使用独立 high 队列和扩展重试策略，保证最终响应不被中间过程挤占丢弃。 --- src/agent/agent_loop.rs | 22 +- src/gateway/outbound_dispatcher.rs | 458 ++++++++++++++++++++++++++--- 2 files changed, 435 insertions(+), 45 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 1eb21b8..7940e4e 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -1818,7 +1818,27 @@ impl AgentLoop { } } - unreachable!("retry loop must return within its body") + // 防御性兜底：正常情况下 for 循环最后一次迭代会走 error 分支返回。 + // 若未来重构导致循环体未返回（如 max_retries 为 0 且循环范围变更）， + // 返回错误响应而非 panic，避免终止所有进行中的会话。 + // 与上方 Err 不可重试路径保持一致：包含 emit_live 推送，确保前端可见。 + tracing::error!( + provider = %self.provider.name(), + model = %self.provider.model_id(), + "run_final_summary retry loop exited without returning" + ); + let final_message = ChatMessage::assistant( + "Failed to generate final summary: retry loop exited unexpectedly.", + ); + emitted_messages.push(final_message.clone()); + self.emit_live_tool_call_message(final_message.clone()) + .await; + AgentProcessResult { + final_response: final_message, + emitted_messages: std::mem::take(emitted_messages), + compaction_performed: false, + engineering_compaction_applied: false, + } } /// 构建取消响应，包含已完成的迭代次数和已生成的消息数量。 diff --git a/src/gateway/outbound_dispatcher.rs b/src/gateway/outbound_dispatcher.rs index 756d124..d05430b 100644 --- a/src/gateway/outbound_dispatcher.rs +++ b/src/gateway/outbound_dispatcher.rs @@ -4,30 +4,54 @@ 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}; -/// 每个 channel 的发送队列容量。 +/// 低优先级队列容量（ToolCall / ToolResult / StreamDelta 等中间过程事件）。 /// -/// 略小于 MessageBus 的容量（100），确保 bus 的 `try_send` 丢消息 -/// 防线仍有效——channel 队列满时 dispatcher 立即丢弃该消息， -/// 不会阻塞路由循环影响其他 channel。 -const PER_CHANNEL_QUEUE_CAPACITY: usize = 64; +/// 略小于 MessageBus 的容量，确保 bus 的 `try_send` 丢消息防线仍有效—— +/// channel 队列满时 dispatcher 立即丢弃该消息，不阻塞路由循环影响其他 channel。 +const LOW_PRIORITY_QUEUE_CAPACITY: usize = 64; -/// 单个 channel 的发送重试间隔（秒）。 +/// 高优先级队列容量（仅 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/"; -/// 单个 channel 的发送上下文：独立 mpsc 队列 + sender task。 +/// 判断消息是否为高优先级（必达）。 /// -/// dispatcher 将消息 `try_send` 到 `tx`，`sender_task` 串行消费并调用 -/// `Channel::send`（含重试）。channel 之间完全隔离——某个 channel 的 -/// 慢发送或重试 sleep 不会阻塞其他 channel 的消息投递。 +/// `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 { - tx: mpsc::Sender, + high_tx: mpsc::Sender, + low_tx: mpsc::Sender, cancel: CancellationToken, } @@ -53,58 +77,82 @@ impl OutboundDispatcher { /// sender task 生命周期与 dispatcher 一致：dispatcher `run()` 退出时 /// 通过 cancel token 终止所有 sender task。 pub async fn register_channel(&self, name: &str, channel: Arc) { - let (tx, rx) = mpsc::channel::(PER_CHANNEL_QUEUE_CAPACITY); + let (high_tx, high_rx) = + mpsc::channel::(HIGH_PRIORITY_QUEUE_CAPACITY); + let (low_tx, low_rx) = + mpsc::channel::(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, rx, cancel_for_task).await; + Self::run_sender_task( + &channel_name, + channel, + high_rx, + low_rx, + cancel_for_task, + ) + .await; }); self.channels .write() .await - .insert(name.to_string(), ChannelSink { tx, cancel }); + .insert( + name.to_string(), + ChannelSink { + high_tx, + low_tx, + cancel, + }, + ); } - /// sender task：串行消费 channel 队列，调用 `Channel::send` 并重试。 + /// sender task：优先消费 high 队列（最终响应），再消费 low 队列（中间过程）， + /// 串行调用 `Channel::send` 并重试。 /// /// 重试 sleep 只阻塞当前 channel 的 task，不影响其他 channel。 async fn run_sender_task( channel_name: &str, channel: Arc, - mut rx: mpsc::Receiver, + mut high_rx: mpsc::Receiver, + mut low_rx: mpsc::Receiver, cancel: CancellationToken, ) { tracing::info!(channel = %channel_name, "Channel sender task started"); loop { - tokio::select! { - // 接收下一条待发送消息 - msg = rx.recv() => { - let Some(msg) = msg else { - tracing::info!(channel = %channel_name, "Channel queue closed, sender task stopping"); - break; - }; + // 优先消费 high 队列：保证最终响应优先发送，不被中间过程阻塞。 + // try_recv 非阻塞，有则立即处理，无则进入 select! 等待。 + if let Ok(msg) = high_rx.try_recv() { + Self::send_one(&*channel, channel_name, msg).await; + continue; + } - 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" - ); + 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; } - Err(error) => { - tracing::error!( - channel = %channel_name, - error = %error, - "Failed to send message after retries" - ); + } + } + // 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; } } } @@ -117,6 +165,54 @@ impl OutboundDispatcher { } } + /// 发送单条消息，处理重试结果日志。 + 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, + ) { + 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, + ) { + 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"); @@ -156,17 +252,26 @@ impl OutboundDispatcher { Some(sink) => { // try_send 保证 dispatcher 永不阻塞：队列满时立即丢弃该消息， // 不影响其他 channel 的投递。与 bus.publish_outbound 策略一致。 - match sink.tx.try_send(msg) { + // 高优先级（最终响应）投递到 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" ); } @@ -186,7 +291,10 @@ impl OutboundDispatcher { } } - /// 发送消息，失败时按 `[1, 2, 4]` 秒间隔重试。 + /// 发送消息，失败时按重试间隔重试。 + /// + /// 高优先级消息（最终响应）使用 `EXTENDED_RETRY_DELAYS_SECS`（5 次，最长 16 秒）， + /// 尽最大努力送达；低优先级消息（中间过程）使用 `RETRY_DELAYS_SECS`（3 次）。 /// /// `ChannelFull` 不可重试——队列满时重试只会浪费退避时间并阻塞 /// channel 队列消费。立即返回让 sender task 尽快处理下一条消息。 @@ -197,15 +305,21 @@ impl OutboundDispatcher { channel: &dyn Channel, msg: OutboundMessage, ) -> Result<(), ChannelError> { - for (attempt_index, delay) in RETRY_DELAYS_SECS.iter().enumerate() { + 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 < RETRY_DELAYS_SECS.len() - 1 => { + 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" ); @@ -215,7 +329,11 @@ impl OutboundDispatcher { } } - unreachable!() + // 防御性兜底：正常情况下循环最后一次迭代会 return Err(error)。 + // 若 delays 被改为空数组，循环体不执行，返回错误而非 panic。 + Err(ChannelError::SendError( + "send_with_retry exhausted with no retry attempts configured".into(), + )) } } @@ -231,6 +349,8 @@ mod tests { struct TestChannel { name: String, received: Arc, + /// 按发送顺序记录每条成功发送消息的 content，用于断言优先级顺序。 + received_contents: Arc>>, delay_ms: u64, fail_first_n: u32, /// 始终返回 `ChannelFull`，用于验证 `send_with_retry` 不重试队列满错误。 @@ -243,6 +363,7 @@ mod tests { 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, @@ -284,7 +405,7 @@ mod tests { Ok(()) } - async fn send(&self, _msg: OutboundMessage) -> Result<(), ChannelError> { + async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> { let count = self.call_count.fetch_add(1, Ordering::SeqCst); // 队列满：不可重试错误，用于验证 send_with_retry 立即返回 @@ -301,6 +422,9 @@ mod tests { } self.received.fetch_add(1, Ordering::SeqCst); + if let Ok(mut guard) = self.received_contents.lock() { + guard.push(msg.content.clone()); + } Ok(()) } } @@ -316,6 +440,54 @@ mod tests { ) } + /// 构造低优先级消息（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 的消息投递 @@ -528,4 +700,202 @@ mod tests { "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(); + } }