fix: observe terminal turn delivery
This commit is contained in:
parent
65ef919714
commit
0c835db380
@ -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 终态与独立消息并发写入同一目标。
|
||||
|
||||
@ -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}"),
|
||||
}
|
||||
|
||||
@ -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};
|
||||
|
||||
@ -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<Result<(), DeliveryError>>,
|
||||
}
|
||||
|
||||
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<Arc<TurnSnapshot>>,
|
||||
) -> Result<(), DeliveryError> {
|
||||
) -> Result<TurnDeliveryHandle, DeliveryError> {
|
||||
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)));
|
||||
}
|
||||
}
|
||||
|
||||
@ -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<String> {
|
||||
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<String, String>,
|
||||
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<Mutex<Session>>,
|
||||
@ -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<Vec<String>>,
|
||||
}
|
||||
|
||||
#[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<MessageBus>) -> 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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user