From 0c835db3805992b22ca73c5f06f96290f5207806 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Sun, 19 Jul 2026 13:23:01 +0800 Subject: [PATCH] fix: observe terminal turn delivery --- docs/ARCHITECTURE.md | 1 + src/delivery/coordinator.rs | 4 + src/delivery/mod.rs | 2 +- src/delivery/service.rs | 48 ++++++++++- src/session/session.rs | 155 +++++++++++++++++++++++++++++++++++- 5 files changed, 201 insertions(+), 9 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index febbae7..f789c88 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -135,6 +135,7 @@ sequenceDiagram - `TurnDeliveryService` 根据 Channel 创建本轮独占的 `TurnSink`；sink 私有保存远端消息 ID 和清理资源。 - `PresentationPolicy` 在快照离开 Gateway 核心前过滤内容。TUI/WebUI 展示独立 reasoning 和详细工具状态；外部 Channel 不接收 reasoning，只接收紧凑工具状态；无人值守投递只保留正文。 - `LivePolicy::Snapshot` 按渠道间隔发送最新运行态；`FinalOnly` 忽略运行态，只处理终态。终态绕过节流并只对明确的瞬态错误重试。 +- `TurnDeliveryService` 返回可等待的终态句柄；sink 生命周期启动不等于终态已送达。Session 在终态重试最终失败时通过普通出站路径兜底一次。 - `cli_chat` 将同一 `turn_updated` 快照发给 TUI 和 WebUI。客户端只保留当前 session 中 revision 更新的 `active_turn`，终态随后由持久化历史校准。 - 飞书默认 `FinalOnly`；开启 `live_updates` 后，第一个可见快照创建卡片，后续编辑同一卡片，终态编辑失败则发送完整结果兜底。reaction 清理在 finish、abort 和 Gateway shutdown 中幂等执行。 - DeliveryCoordinator 与 OutboundDispatcher 共享 `(channel, chat_id)` 写锁，避免活动 Turn 终态与独立消息并发写入同一目标。 diff --git a/src/delivery/coordinator.rs b/src/delivery/coordinator.rs index 789f123..4bb7e8f 100644 --- a/src/delivery/coordinator.rs +++ b/src/delivery/coordinator.rs @@ -23,6 +23,7 @@ pub enum DeliveryError { OpenFailed(ChannelError), SnapshotStreamClosed, SupervisorStopping, + CompletionLost, FinalTimedOut, FinalFailed(ChannelError), } @@ -38,6 +39,9 @@ impl std::fmt::Display for DeliveryError { Self::SupervisorStopping => { formatter.write_str("cannot start turn delivery while Gateway is stopping") } + Self::CompletionLost => { + formatter.write_str("turn delivery task stopped without reporting completion") + } Self::FinalTimedOut => formatter.write_str("final turn delivery timed out"), Self::FinalFailed(error) => write!(formatter, "final turn delivery failed: {error}"), } diff --git a/src/delivery/mod.rs b/src/delivery/mod.rs index 10f594e..04522fc 100644 --- a/src/delivery/mod.rs +++ b/src/delivery/mod.rs @@ -4,4 +4,4 @@ mod service; pub use coordinator::{ConversationWriteLocks, DeliveryCoordinator, DeliveryError}; pub use policy::{PresentationPolicy, ReasoningVisibility, ToolVisibility, project_snapshot}; -pub use service::TurnDeliveryService; +pub use service::{TurnDeliveryHandle, TurnDeliveryService}; diff --git a/src/delivery/service.rs b/src/delivery/service.rs index 5c3303c..22eb8a5 100644 --- a/src/delivery/service.rs +++ b/src/delivery/service.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use tokio::sync::watch; +use tokio::sync::{oneshot, watch}; use crate::channels::{ChannelManager, TurnTarget}; use crate::delivery::{DeliveryCoordinator, DeliveryError}; @@ -19,6 +19,23 @@ pub struct TurnDeliveryService { supervisor: TaskSupervisor, } +/// Completion handle for one TurnSink lifecycle. +/// +/// Creating a sink only proves that delivery started. Callers consume this +/// handle after publishing a terminal snapshot to learn whether the terminal +/// write actually reached the Channel. +pub struct TurnDeliveryHandle { + pub(crate) completion: oneshot::Receiver>, +} + +impl TurnDeliveryHandle { + pub async fn wait(self) -> Result<(), DeliveryError> { + self.completion + .await + .unwrap_or(Err(DeliveryError::CompletionLost)) + } +} + impl TurnDeliveryService { pub fn new( coordinator: DeliveryCoordinator, @@ -36,7 +53,7 @@ impl TurnDeliveryService { &self, target: TurnTarget, snapshots: watch::Receiver>, - ) -> Result<(), DeliveryError> { + ) -> Result { let channel = self .channels .get_channel(&target.channel) @@ -48,7 +65,7 @@ impl TurnDeliveryService { .open_turn(target.clone()) .await .map_err(DeliveryError::OpenFailed)?; - let _result = self.coordinator.spawn_sink( + let completion = self.coordinator.spawn_sink( &self.supervisor, SinkRoute { channel: target.channel, @@ -59,6 +76,29 @@ impl TurnDeliveryService { snapshots, sink, )?; - Ok(()) + Ok(TurnDeliveryHandle { completion }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn handle_reports_terminal_delivery_failure() { + let (sender, completion) = oneshot::channel(); + sender.send(Err(DeliveryError::FinalTimedOut)).unwrap(); + + let result = TurnDeliveryHandle { completion }.wait().await; + assert!(matches!(result, Err(DeliveryError::FinalTimedOut))); + } + + #[tokio::test] + async fn handle_reports_lost_delivery_task() { + let (sender, completion) = oneshot::channel(); + drop(sender); + + let result = TurnDeliveryHandle { completion }.wait().await; + assert!(matches!(result, Err(DeliveryError::CompletionLost))); } } diff --git a/src/session/session.rs b/src/session/session.rs index 11bac93..c2b7bf2 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -53,7 +53,7 @@ use crate::agent::{AgentError, AgentLoop, AgentTurnContext, ContextCompressor, T use crate::channels::slash_command::parse_slash_command; use crate::config::BrowserConfig; use crate::config::LLMProviderConfig; -use crate::delivery::TurnDeliveryService; +use crate::delivery::{TurnDeliveryHandle, TurnDeliveryService}; /// Check if an LLM error message indicates a context window overflow. fn is_context_overflow_error(msg: &str) -> bool { @@ -102,6 +102,46 @@ fn partial_assistant_message( Some(message) } +fn terminal_fallback_content(snapshot: &TurnSnapshot) -> Option { + partial_assistant_message(snapshot, CompletionStatus::Interrupted) + .map(|message| message.content) + .or_else(|| { + (snapshot.status == super::turn::TurnStatus::Failed) + .then(|| "The response could not be delivered. Please try again.".to_string()) + }) +} + +async fn deliver_terminal_fallback( + handle: TurnDeliveryHandle, + bus: &MessageBus, + channel: &str, + chat_id: &str, + session_id: &str, + forwarded_metadata: &HashMap, + controller: &TurnController, +) { + let Err(error) = handle.wait().await else { + return; + }; + tracing::error!(channel, chat_id, error = %error, "Turn sink terminal delivery failed; using ordinary outbound fallback"); + let snapshot = controller.snapshot(); + let Some(content) = terminal_fallback_content(&snapshot) else { + return; + }; + let outbound = OutboundMessage { + channel: channel.to_string(), + chat_id: chat_id.to_string(), + content, + reply_to: None, + media: vec![], + metadata: outbound_turn_metadata(session_id, forwarded_metadata), + delivery: None, + }; + if let Err(fallback_error) = bus.deliver_outbound(outbound).await { + tracing::error!(channel, chat_id, error = %fallback_error, "Ordinary terminal fallback delivery failed"); + } +} + async fn fail_turn_with_partial( controller: &TurnController, session: &Arc>, @@ -125,6 +165,39 @@ async fn fail_turn_with_partial( mod cancelled_partial_tests { use super::*; use crate::agent::TurnEvent; + use crate::bus::{MessageBus, OutboundDispatcher}; + use crate::channels::{Channel, ChannelError, ChannelManager, CliChatChannel}; + use crate::delivery::{ConversationWriteLocks, DeliveryError}; + use crate::task_supervisor::TaskSupervisor; + + #[derive(Default)] + struct RecordingChannel { + messages: Mutex>, + } + + #[async_trait::async_trait] + impl Channel for RecordingChannel { + fn name(&self) -> &str { + "recording" + } + + fn is_running(&self) -> bool { + true + } + + async fn start(&self, _bus: Arc) -> Result<(), ChannelError> { + Ok(()) + } + + async fn stop(&self) -> Result<(), ChannelError> { + Ok(()) + } + + async fn send(&self, message: OutboundMessage) -> Result<(), ChannelError> { + self.messages.lock().await.push(message.content); + Ok(()) + } + } #[test] fn turn_metadata_preserves_channel_cleanup_fields() { @@ -184,6 +257,66 @@ mod cancelled_partial_tests { assert_eq!(message.completion_status, CompletionStatus::Cancelled); } + #[test] + fn failed_terminal_without_visible_text_has_safe_fallback() { + let (controller, _emitter, _) = TurnController::start("session", "message-id"); + controller.fail("provider response contained a secret"); + + assert_eq!( + terminal_fallback_content(&controller.snapshot()).as_deref(), + Some("The response could not be delivered. Please try again.") + ); + } + + #[tokio::test] + async fn asynchronous_terminal_failure_uses_one_ordinary_fallback() { + let bus = MessageBus::new(8); + let cli = Arc::new(CliChatChannel::new()); + let channels = ChannelManager::with_bus(cli, bus.clone()); + let channel = Arc::new(RecordingChannel::default()); + channels + .register_channel("recording", channel.clone()) + .await; + let supervisor = TaskSupervisor::new(); + let dispatcher = OutboundDispatcher::new( + bus.clone(), + channels, + supervisor.clone(), + ConversationWriteLocks::default(), + ); + let dispatcher_task = tokio::spawn(async move { dispatcher.run().await }); + + let (controller, emitter, _) = TurnController::start("session", "message-id"); + emitter + .emit(TurnEvent::TextDelta { + iteration: 0, + delta: "completed response".into(), + }) + .unwrap(); + controller.complete(None); + let (sender, completion) = oneshot::channel(); + sender.send(Err(DeliveryError::FinalTimedOut)).unwrap(); + + deliver_terminal_fallback( + TurnDeliveryHandle { completion }, + &bus, + "recording", + "chat", + "session", + &HashMap::new(), + &controller, + ) + .await; + + assert_eq!( + channel.messages.lock().await.as_slice(), + &["completed response"] + ); + dispatcher_task.abort(); + let _ = dispatcher_task.await; + supervisor.shutdown(std::time::Duration::from_secs(1)).await; + } + #[test] fn reasoning_only_cancel_does_not_create_assistant_history() { let (controller, emitter, _) = TurnController::start("session", "message-id"); @@ -2591,7 +2724,7 @@ fn spawn_agent_worker( ); let initial_turn = turn_controller.snapshot(); let active_turn_id = initial_turn.id.0.clone(); - let live_delivery_started = match turn_delivery + let delivery_handle = match turn_delivery .start( crate::channels::TurnTarget { channel: task_chan.clone(), @@ -2604,16 +2737,17 @@ fn spawn_agent_worker( ) .await { - Ok(()) => true, + Ok(handle) => Some(handle), Err(error) => { tracing::debug!( channel = %task_chan, error = %error, "Live turn delivery unavailable; using ordinary final delivery" ); - false + None } }; + let live_delivery_started = delivery_handle.is_some(); { let mut guard = session.lock().await; if guard.worker_generation != worker_gen || guard.state_version != base_version { @@ -2912,6 +3046,19 @@ fn spawn_agent_worker( } } + if let Some(handle) = delivery_handle { + deliver_terminal_fallback( + handle, + &bus, + &task_chan, + &task_cid, + &unified_str, + &task_metadata, + &turn_controller, + ) + .await; + } + // Clean up let mut guard = session.lock().await; if guard