From 55da3204f8864f057eddb089de83e3969e3a7a51 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Fri, 17 Jul 2026 15:56:52 +0800 Subject: [PATCH] feat: coordinate live turn delivery --- docs/STREAMING_TURN_DESIGN.md | 6 +- src/bus/dispatcher.rs | 66 +++- src/channels/base.rs | 43 ++- src/channels/mod.rs | 2 +- src/delivery/coordinator.rs | 548 ++++++++++++++++++++++++++++++++++ src/delivery/mod.rs | 5 + src/delivery/policy.rs | 138 +++++++++ src/gateway/mod.rs | 5 + src/lib.rs | 1 + 9 files changed, 796 insertions(+), 18 deletions(-) create mode 100644 src/delivery/coordinator.rs create mode 100644 src/delivery/mod.rs create mode 100644 src/delivery/policy.rs diff --git a/docs/STREAMING_TURN_DESIGN.md b/docs/STREAMING_TURN_DESIGN.md index 91c2159..0d75218 100644 --- a/docs/STREAMING_TURN_DESIGN.md +++ b/docs/STREAMING_TURN_DESIGN.md @@ -474,12 +474,12 @@ pub trait Channel: Send + Sync + 'static { #[async_trait] pub trait TurnSink: Send { async fn update(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; - async fn finish(self: Box, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; - async fn abort(self: Box, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; + async fn finish(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; + async fn abort(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; } ``` -`TurnSink` 的每个调用都接收完整、过滤后的快照。Sink 不拼接 token。 +`TurnSink` 的每个调用都接收完整、过滤后的快照。Sink 不拼接 token。终态方法保留 `&mut self`，使协调器可以在瞬态错误或超时后重试同一个、仍持有远端消息 ID 的 sink；终态成功或重试耗尽后由协调器销毁 sink。 ### 10.2 cli_chat diff --git a/src/bus/dispatcher.rs b/src/bus/dispatcher.rs index 35927c9..edb830e 100644 --- a/src/bus/dispatcher.rs +++ b/src/bus/dispatcher.rs @@ -7,6 +7,7 @@ use tokio::sync::mpsc; use crate::bus::{MessageBus, OutboundMessage}; use crate::channels::ChannelManager; use crate::channels::base::{Channel, ChannelError}; +use crate::delivery::ConversationWriteLocks; use crate::task_supervisor::TaskSupervisor; const LANE_CAPACITY: usize = 64; @@ -20,6 +21,7 @@ pub struct OutboundDispatcher { bus: Arc, channel_manager: ChannelManager, task_supervisor: TaskSupervisor, + write_locks: ConversationWriteLocks, } impl OutboundDispatcher { @@ -27,11 +29,13 @@ impl OutboundDispatcher { bus: Arc, channel_manager: ChannelManager, task_supervisor: TaskSupervisor, + write_locks: ConversationWriteLocks, ) -> Self { Self { bus, channel_manager, task_supervisor, + write_locks, } } @@ -120,6 +124,7 @@ impl OutboundDispatcher { channel_name: String, chat_id: String, ) -> bool { + let target_lock = self.write_locks.for_target(&channel_name, &chat_id); self.task_supervisor.spawn( format!("outbound-lane:{channel_name}:{chat_id}"), async move { @@ -128,7 +133,7 @@ impl OutboundDispatcher { Ok(Some(msg)) => msg, Ok(None) | Err(_) => break, }; - let result = Self::send_with_retry(&*channel, &msg).await; + let result = Self::send_with_retry(&*channel, &msg, &target_lock).await; if let Err(error) = &result { tracing::error!( channel = %channel_name, @@ -146,7 +151,9 @@ impl OutboundDispatcher { async fn send_with_retry( channel: &dyn Channel, msg: &OutboundMessage, + target_lock: &tokio::sync::Mutex<()>, ) -> Result<(), ChannelError> { + let _guard = target_lock.lock().await; const DELAYS: &[u64] = &[1, 2, 4]; for (attempt, &delay) in DELAYS.iter().enumerate() { @@ -263,7 +270,12 @@ mod tests { manager.register_channel("recording", channel.clone()).await; let supervisor = TaskSupervisor::new(); - let dispatcher = OutboundDispatcher::new(bus.clone(), manager, supervisor.clone()); + let dispatcher = OutboundDispatcher::new( + bus.clone(), + manager, + supervisor.clone(), + ConversationWriteLocks::default(), + ); let task = tokio::spawn(async move { dispatcher.run().await }); bus.publish_outbound(outbound("slow", "slow-1")) .await @@ -301,7 +313,12 @@ mod tests { bus.clone(), ); let supervisor = TaskSupervisor::new(); - let dispatcher = OutboundDispatcher::new(bus.clone(), manager, supervisor.clone()); + let dispatcher = OutboundDispatcher::new( + bus.clone(), + manager, + supervisor.clone(), + ConversationWriteLocks::default(), + ); let task = tokio::spawn(async move { dispatcher.run().await }); let mut message = outbound("missing", "not delivered"); @@ -326,7 +343,12 @@ mod tests { }); manager.register_channel("recording", channel.clone()).await; let supervisor = TaskSupervisor::new(); - let dispatcher = OutboundDispatcher::new(bus.clone(), manager, supervisor.clone()); + let dispatcher = OutboundDispatcher::new( + bus.clone(), + manager, + supervisor.clone(), + ConversationWriteLocks::default(), + ); let task = tokio::spawn(async move { dispatcher.run().await }); bus.deliver_outbound(outbound("confirmed", "delivered")) @@ -344,11 +366,41 @@ mod tests { attempts: AtomicUsize::new(0), }; - let error = OutboundDispatcher::send_with_retry(&channel, &outbound("invalid", "message")) - .await - .unwrap_err(); + let target_lock = tokio::sync::Mutex::new(()); + let error = OutboundDispatcher::send_with_retry( + &channel, + &outbound("invalid", "message"), + &target_lock, + ) + .await + .unwrap_err(); assert!(matches!(error, ChannelError::Other(_))); assert_eq!(channel.attempts.load(Ordering::SeqCst), 1); } + + #[tokio::test] + async fn shared_write_lock_orders_dispatcher_with_live_turn_writes() { + let channel = RecordingChannel { + sent: Mutex::new(Vec::new()), + notify: Notify::new(), + }; + let write_locks = ConversationWriteLocks::default(); + let target_lock = write_locks.for_target("recording", "same-chat"); + let live_write = target_lock.lock().await; + let message = outbound("same-chat", "after-live-update"); + + let send = OutboundDispatcher::send_with_retry(&channel, &message, &target_lock); + tokio::pin!(send); + assert!( + tokio::time::timeout(Duration::from_millis(10), &mut send) + .await + .is_err() + ); + assert!(channel.sent.lock().await.is_empty()); + + drop(live_write); + send.await.unwrap(); + assert_eq!(channel.sent.lock().await.as_slice(), &["after-live-update"]); + } } diff --git a/src/channels/base.rs b/src/channels/base.rs index 73e164b..8fad062 100644 --- a/src/channels/base.rs +++ b/src/channels/base.rs @@ -1,7 +1,32 @@ use async_trait::async_trait; +use std::collections::HashMap; use std::sync::Arc; +use std::time::Duration; use crate::bus::{BusError, InboundMessage, MessageBus, OutboundMessage}; +use crate::session::TurnSnapshot; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LivePolicy { + FinalOnly, + Snapshot { min_interval: Duration }, +} + +#[derive(Debug, Clone)] +pub struct TurnTarget { + pub channel: String, + pub chat_id: String, + pub session_id: String, + pub reply_to: Option, + pub metadata: HashMap, +} + +#[async_trait] +pub trait TurnSink: Send { + async fn update(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; + async fn finish(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; + async fn abort(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError>; +} #[derive(Debug)] pub enum ChannelError { @@ -49,16 +74,20 @@ pub trait Channel: Send + Sync + 'static { /// Stop the channel async fn stop(&self) -> Result<(), ChannelError>; + fn live_policy(&self) -> LivePolicy { + LivePolicy::FinalOnly + } + + async fn open_turn(&self, _target: TurnTarget) -> Result, ChannelError> { + Err(ChannelError::Other(format!( + "channel {} does not support turn delivery", + self.name() + ))) + } + /// Send a message to the channel (called by OutboundDispatcher) async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError>; - /// Send a streaming delta (optional, for channels that support it) - async fn send_delta(&self, chat_id: &str, delta: &str) -> Result<(), ChannelError> { - let _ = chat_id; - let _ = delta; - Ok(()) - } - /// Check if a sender is allowed to use this channel fn is_allowed(&self, _sender_id: &str) -> bool { true diff --git a/src/channels/mod.rs b/src/channels/mod.rs index 9331bad..2eadad9 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -4,7 +4,7 @@ pub mod feishu; pub mod manager; pub mod slash_command; -pub use base::{Channel, ChannelError}; +pub use base::{Channel, ChannelError, LivePolicy, TurnSink, TurnTarget}; pub use cli_chat::CliChatChannel; pub use feishu::FeishuChannel; pub use manager::ChannelManager; diff --git a/src/delivery/coordinator.rs b/src/delivery/coordinator.rs new file mode 100644 index 0000000..5769b68 --- /dev/null +++ b/src/delivery/coordinator.rs @@ -0,0 +1,548 @@ +use std::collections::HashMap; +use std::sync::{Arc, Mutex, Weak}; +use std::time::Duration; + +use tokio::sync::{Mutex as AsyncMutex, oneshot, watch}; +use tokio::time::{Instant, sleep_until, timeout}; + +use crate::channels::{Channel, ChannelError, LivePolicy, TurnSink, TurnTarget}; +use crate::delivery::{PresentationPolicy, project_snapshot}; +use crate::session::{TurnSnapshot, TurnStatus}; +use crate::task_supervisor::TaskSupervisor; + +const SINK_CALL_TIMEOUT: Duration = Duration::from_secs(30); +const FINAL_RETRY_DELAYS: &[Duration] = &[ + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(4), +]; + +#[derive(Debug)] +pub enum DeliveryError { + OpenFailed(ChannelError), + SnapshotStreamClosed, + SupervisorStopping, + FinalTimedOut, + FinalFailed(ChannelError), +} + +impl std::fmt::Display for DeliveryError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::OpenFailed(error) => write!(formatter, "failed to open turn sink: {error}"), + Self::SnapshotStreamClosed => { + formatter.write_str("turn snapshot stream closed before a terminal state") + } + Self::SupervisorStopping => { + formatter.write_str("cannot start turn delivery while Gateway is stopping") + } + Self::FinalTimedOut => formatter.write_str("final turn delivery timed out"), + Self::FinalFailed(error) => write!(formatter, "final turn delivery failed: {error}"), + } + } +} + +impl std::error::Error for DeliveryError {} + +/// Shared ordering boundary for writes to one `(channel, chat_id)` target. +/// +/// The registry stores weak references so inactive conversations disappear +/// without a cleanup task. Callers hold the returned lock only around one +/// external write, never for the lifetime of a Turn. +#[derive(Clone, Default)] +pub struct ConversationWriteLocks { + locks: Arc>>>>, +} + +impl ConversationWriteLocks { + pub fn for_target(&self, channel: &str, chat_id: &str) -> Arc> { + let key = format!("{channel}\0{chat_id}"); + let mut locks = self + .locks + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(existing) = locks.get(&key).and_then(Weak::upgrade) { + return existing; + } + let lock = Arc::new(AsyncMutex::new(())); + locks.insert(key, Arc::downgrade(&lock)); + lock + } +} + +#[derive(Clone)] +pub struct DeliveryCoordinator { + write_locks: ConversationWriteLocks, + sink_call_timeout: Duration, + final_retry_delays: Arc<[Duration]>, +} + +impl DeliveryCoordinator { + pub fn new(write_locks: ConversationWriteLocks) -> Self { + Self { + write_locks, + sink_call_timeout: SINK_CALL_TIMEOUT, + final_retry_delays: FINAL_RETRY_DELAYS.into(), + } + } + + #[cfg(test)] + fn for_test( + sink_call_timeout: Duration, + final_retry_delays: impl Into>, + ) -> Self { + Self { + write_locks: ConversationWriteLocks::default(), + sink_call_timeout, + final_retry_delays: final_retry_delays.into(), + } + } + + pub fn write_locks(&self) -> ConversationWriteLocks { + self.write_locks.clone() + } + + pub async fn open_and_deliver( + &self, + channel: Arc, + target: TurnTarget, + presentation: PresentationPolicy, + snapshots: watch::Receiver>, + ) -> Result<(), DeliveryError> { + let live_policy = channel.live_policy(); + let sink = channel + .open_turn(target.clone()) + .await + .map_err(DeliveryError::OpenFailed)?; + self.deliver( + &target.channel, + &target.chat_id, + live_policy, + presentation, + snapshots, + sink, + ) + .await + } + + /// Start one sink lifecycle under the Gateway's task owner and return a + /// bounded completion report to the caller. + pub fn spawn( + &self, + supervisor: &TaskSupervisor, + channel: Arc, + target: TurnTarget, + presentation: PresentationPolicy, + snapshots: watch::Receiver>, + ) -> Result>, DeliveryError> { + let (result_tx, result_rx) = oneshot::channel(); + let coordinator = self.clone(); + let task_name = format!("turn-delivery:{}:{}", target.channel, target.chat_id); + let spawned = supervisor.spawn(task_name, async move { + let result = coordinator + .open_and_deliver(channel, target, presentation, snapshots) + .await; + let _ = result_tx.send(result); + }); + if !spawned { + return Err(DeliveryError::SupervisorStopping); + } + Ok(result_rx) + } + + pub async fn deliver( + &self, + channel: &str, + chat_id: &str, + live_policy: LivePolicy, + presentation: PresentationPolicy, + mut snapshots: watch::Receiver>, + mut sink: Box, + ) -> Result<(), DeliveryError> { + let target_lock = self.write_locks.for_target(channel, chat_id); + let min_interval = match live_policy { + LivePolicy::FinalOnly => None, + LivePolicy::Snapshot { min_interval } if presentation.live => Some(min_interval), + LivePolicy::Snapshot { .. } => None, + }; + let mut next_update_at = Instant::now(); + + loop { + let snapshot = snapshots.borrow_and_update().clone(); + if snapshot.status != TurnStatus::Running { + let projected = project_snapshot(&snapshot, presentation); + return self + .deliver_terminal(&target_lock, &mut *sink, &projected) + .await; + } + + if let Some(interval) = min_interval { + while Instant::now() < next_update_at { + tokio::select! { + changed = snapshots.changed() => { + changed.map_err(|_| DeliveryError::SnapshotStreamClosed)?; + let latest = snapshots.borrow_and_update().clone(); + if latest.status != TurnStatus::Running { + let projected = project_snapshot(&latest, presentation); + return self.deliver_terminal(&target_lock, &mut *sink, &projected).await; + } + } + () = sleep_until(next_update_at) => break, + } + } + + let latest = snapshots.borrow_and_update().clone(); + if latest.status != TurnStatus::Running { + let projected = project_snapshot(&latest, presentation); + return self + .deliver_terminal(&target_lock, &mut *sink, &projected) + .await; + } + let projected = project_snapshot(&latest, presentation); + let _guard = target_lock.lock().await; + match timeout(self.sink_call_timeout, sink.update(&projected)).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + tracing::warn!(error = %error, revision = projected.revision, "Live turn update failed; waiting for a newer snapshot"); + } + Err(_) => { + tracing::warn!( + revision = projected.revision, + "Live turn update timed out; waiting for a newer snapshot" + ); + } + } + next_update_at = Instant::now() + interval; + } + + snapshots + .changed() + .await + .map_err(|_| DeliveryError::SnapshotStreamClosed)?; + } + } + + async fn deliver_terminal( + &self, + target_lock: &Arc>, + sink: &mut dyn TurnSink, + snapshot: &TurnSnapshot, + ) -> Result<(), DeliveryError> { + let attempts = self.final_retry_delays.len() + 1; + for attempt in 0..attempts { + let _guard = target_lock.lock().await; + let result = if snapshot.status == TurnStatus::Completed { + timeout(self.sink_call_timeout, sink.finish(snapshot)).await + } else { + timeout(self.sink_call_timeout, sink.abort(snapshot)).await + }; + drop(_guard); + + match result { + Ok(Ok(())) => return Ok(()), + Ok(Err(error)) + if error.is_transient() && attempt < self.final_retry_delays.len() => + { + sleep_until(Instant::now() + self.final_retry_delays[attempt]).await; + } + Ok(Err(error)) => return Err(DeliveryError::FinalFailed(error)), + Err(_) if attempt < self.final_retry_delays.len() => { + sleep_until(Instant::now() + self.final_retry_delays[attempt]).await; + } + Err(_) => return Err(DeliveryError::FinalTimedOut), + } + } + unreachable!() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::{Mutex as TokioMutex, Notify}; + + use crate::agent::TurnEvent; + use crate::bus::{MessageBus, OutboundMessage}; + use crate::channels::{Channel, TurnSink}; + use crate::session::{TurnBlock, TurnController}; + + #[derive(Default)] + struct SinkState { + updates: TokioMutex>, + terminal: TokioMutex>, + update_started: Notify, + release_update: Notify, + block_first_update: bool, + fail_updates: AtomicUsize, + fail_finish: AtomicUsize, + } + + struct RecordingSink(Arc); + + struct SinkChannel { + state: Arc, + opened: AtomicUsize, + } + + #[async_trait] + impl Channel for SinkChannel { + fn name(&self) -> &str { + "sink-channel" + } + + fn is_running(&self) -> bool { + true + } + + async fn start(&self, _bus: Arc) -> Result<(), ChannelError> { + Ok(()) + } + + async fn stop(&self) -> Result<(), ChannelError> { + Ok(()) + } + + fn live_policy(&self) -> LivePolicy { + LivePolicy::FinalOnly + } + + async fn open_turn(&self, _target: TurnTarget) -> Result, ChannelError> { + self.opened.fetch_add(1, Ordering::SeqCst); + Ok(sink(self.state.clone())) + } + + async fn send(&self, _msg: OutboundMessage) -> Result<(), ChannelError> { + Ok(()) + } + } + + #[async_trait] + impl TurnSink for RecordingSink { + async fn update(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> { + self.0.update_started.notify_waiters(); + if self.0.block_first_update && self.0.updates.lock().await.is_empty() { + self.0.release_update.notified().await; + } + if self + .0 + .fail_updates + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return Err(ChannelError::SendError("update".into())); + } + self.0.updates.lock().await.push(snapshot.clone()); + Ok(()) + } + + async fn finish(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> { + if self + .0 + .fail_finish + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return Err(ChannelError::SendError("finish".into())); + } + self.0.terminal.lock().await.push(snapshot.clone()); + Ok(()) + } + + async fn abort(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> { + self.0.terminal.lock().await.push(snapshot.clone()); + Ok(()) + } + } + + fn sink(state: Arc) -> Box { + Box::new(RecordingSink(state)) + } + + #[tokio::test] + async fn slow_sink_observes_latest_snapshot_and_terminal_bypasses_throttle() { + let state = Arc::new(SinkState { + block_first_update: true, + ..SinkState::default() + }); + let (controller, emitter, receiver) = TurnController::start("session", "message"); + let coordinator = DeliveryCoordinator::for_test(Duration::from_secs(30), []); + let task = tokio::spawn({ + let state = state.clone(); + async move { + coordinator + .deliver( + "cli_chat", + "chat", + LivePolicy::Snapshot { + min_interval: Duration::from_secs(10), + }, + PresentationPolicy::interactive(), + receiver, + sink(state), + ) + .await + } + }); + + state.update_started.notified().await; + emitter + .emit(TurnEvent::TextDelta { + iteration: 0, + delta: "a".into(), + }) + .unwrap(); + emitter + .emit(TurnEvent::TextDelta { + iteration: 0, + delta: "b".into(), + }) + .unwrap(); + state.release_update.notify_waiters(); + tokio::task::yield_now().await; + controller.complete(None); + + assert!(task.await.unwrap().is_ok()); + let terminal = state.terminal.lock().await; + assert_eq!(terminal.len(), 1); + assert_eq!(terminal[0].status, TurnStatus::Completed); + assert!( + matches!(&terminal[0].blocks[0], TurnBlock::Assistant { text, .. } if text == "ab") + ); + } + + #[tokio::test] + async fn hidden_reasoning_is_removed_before_sink_and_failed_update_recovers() { + let state = Arc::new(SinkState { + fail_updates: AtomicUsize::new(1), + ..SinkState::default() + }); + let (controller, emitter, receiver) = TurnController::start("session", "message"); + let coordinator = DeliveryCoordinator::for_test(Duration::from_secs(30), []); + let task = tokio::spawn({ + let state = state.clone(); + async move { + coordinator + .deliver( + "feishu", + "chat", + LivePolicy::Snapshot { + min_interval: Duration::ZERO, + }, + PresentationPolicy::external(true), + receiver, + sink(state), + ) + .await + } + }); + + emitter + .emit(TurnEvent::ReasoningDelta { + iteration: 0, + delta: "secret".into(), + }) + .unwrap(); + tokio::task::yield_now().await; + emitter + .emit(TurnEvent::TextDelta { + iteration: 0, + delta: "public".into(), + }) + .unwrap(); + tokio::task::yield_now().await; + controller.complete(None); + + assert!(task.await.unwrap().is_ok()); + let terminal = state.terminal.lock().await; + assert!( + terminal[0] + .blocks + .iter() + .all(|block| !matches!(block, TurnBlock::Reasoning { .. })) + ); + } + + #[tokio::test] + async fn final_only_skips_updates_and_retries_transient_finish() { + let state = Arc::new(SinkState { + fail_finish: AtomicUsize::new(2), + ..SinkState::default() + }); + let (controller, emitter, receiver) = TurnController::start("session", "message"); + let coordinator = DeliveryCoordinator::for_test( + Duration::from_secs(30), + [Duration::from_millis(1), Duration::from_millis(2)], + ); + let task = tokio::spawn({ + let state = state.clone(); + async move { + coordinator + .deliver( + "channel", + "chat", + LivePolicy::FinalOnly, + PresentationPolicy::unattended(), + receiver, + sink(state), + ) + .await + } + }); + + emitter + .emit(TurnEvent::TextDelta { + iteration: 0, + delta: "done".into(), + }) + .unwrap(); + controller.complete(None); + + assert!(task.await.unwrap().is_ok()); + assert!(state.updates.lock().await.is_empty()); + assert_eq!(state.terminal.lock().await.len(), 1); + } + + #[tokio::test] + async fn open_and_deliver_owns_sink_creation_and_terminal_lifecycle() { + let state = Arc::new(SinkState::default()); + let channel = Arc::new(SinkChannel { + state: state.clone(), + opened: AtomicUsize::new(0), + }); + let (controller, emitter, receiver) = TurnController::start("session", "message"); + let coordinator = DeliveryCoordinator::for_test(Duration::from_secs(30), []); + let target = TurnTarget { + channel: "sink-channel".into(), + chat_id: "chat".into(), + session_id: "session".into(), + reply_to: None, + metadata: HashMap::new(), + }; + let task = tokio::spawn({ + let channel = channel.clone(); + async move { + coordinator + .open_and_deliver(channel, target, PresentationPolicy::unattended(), receiver) + .await + } + }); + + emitter + .emit(TurnEvent::TextDelta { + iteration: 0, + delta: "done".into(), + }) + .unwrap(); + controller.complete(None); + + assert!(task.await.unwrap().is_ok()); + assert_eq!(channel.opened.load(Ordering::SeqCst), 1); + assert_eq!(state.terminal.lock().await.len(), 1); + } +} diff --git a/src/delivery/mod.rs b/src/delivery/mod.rs new file mode 100644 index 0000000..a2c6d2c --- /dev/null +++ b/src/delivery/mod.rs @@ -0,0 +1,5 @@ +mod coordinator; +mod policy; + +pub use coordinator::{ConversationWriteLocks, DeliveryCoordinator, DeliveryError}; +pub use policy::{PresentationPolicy, ReasoningVisibility, ToolVisibility, project_snapshot}; diff --git a/src/delivery/policy.rs b/src/delivery/policy.rs new file mode 100644 index 0000000..14b082b --- /dev/null +++ b/src/delivery/policy.rs @@ -0,0 +1,138 @@ +use crate::session::{TurnBlock, TurnSnapshot}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReasoningVisibility { + Hidden, + Collapsed, + Expanded, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ToolVisibility { + Hidden, + Compact, + Detailed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PresentationPolicy { + pub live: bool, + pub reasoning: ReasoningVisibility, + pub tools: ToolVisibility, +} + +impl PresentationPolicy { + pub const fn interactive() -> Self { + Self { + live: true, + reasoning: ReasoningVisibility::Collapsed, + tools: ToolVisibility::Detailed, + } + } + + pub const fn external(live: bool) -> Self { + Self { + live, + reasoning: ReasoningVisibility::Hidden, + tools: ToolVisibility::Compact, + } + } + + pub const fn unattended() -> Self { + Self { + live: false, + reasoning: ReasoningVisibility::Hidden, + tools: ToolVisibility::Hidden, + } + } +} + +/// Produce the immutable view that is allowed to leave the Gateway core. +/// +/// Presentation filtering deliberately clones the snapshot. Conversation +/// history and the authoritative TurnController state remain untouched. +pub fn project_snapshot(snapshot: &TurnSnapshot, policy: PresentationPolicy) -> TurnSnapshot { + let mut projected = snapshot.clone(); + projected.blocks.retain_mut(|block| match block { + TurnBlock::Reasoning { .. } => policy.reasoning != ReasoningVisibility::Hidden, + TurnBlock::Assistant { .. } => true, + TurnBlock::Tool { + arguments, preview, .. + } => match policy.tools { + ToolVisibility::Hidden => false, + ToolVisibility::Compact => { + *arguments = serde_json::Value::Null; + *preview = None; + true + } + ToolVisibility::Detailed => true, + }, + }); + projected +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::session::{BlockId, ToolStatus, TurnId, TurnPhase, TurnState, TurnStatus}; + + fn snapshot() -> TurnSnapshot { + TurnState { + id: TurnId("turn".into()), + session_id: "session".into(), + message_id: "message".into(), + revision: 3, + status: TurnStatus::Running, + phase: TurnPhase::Acting, + blocks: vec![ + TurnBlock::Reasoning { + id: BlockId("reasoning".into()), + iteration: 0, + text: "private chain".into(), + }, + TurnBlock::Assistant { + id: BlockId("text".into()), + iteration: 0, + text: "visible".into(), + }, + TurnBlock::Tool { + id: "tool".into(), + iteration: 0, + name: "bash".into(), + arguments: serde_json::json!({"token": "secret"}), + status: ToolStatus::Completed, + preview: Some("sensitive output".into()), + }, + ], + usage: None, + error: None, + } + } + + #[test] + fn external_projection_removes_reasoning_and_tool_details_without_mutating_source() { + let source = snapshot(); + let projected = project_snapshot(&source, PresentationPolicy::external(true)); + + assert_eq!(projected.blocks.len(), 2); + assert!(matches!(projected.blocks[0], TurnBlock::Assistant { .. })); + assert!(matches!( + &projected.blocks[1], + TurnBlock::Tool { + arguments: serde_json::Value::Null, + preview: None, + .. + } + )); + assert_eq!(source.blocks.len(), 3); + assert!(matches!(source.blocks[0], TurnBlock::Reasoning { .. })); + } + + #[test] + fn unattended_projection_keeps_only_assistant_blocks() { + let projected = project_snapshot(&snapshot(), PresentationPolicy::unattended()); + + assert_eq!(projected.blocks.len(), 1); + assert!(matches!(projected.blocks[0], TurnBlock::Assistant { .. })); + } +} diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 9d40aa1..085db4a 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -12,6 +12,7 @@ use crate::bus::{ControlMessage, MessageBus, OutboundDispatcher}; use crate::channels::base::ChannelError; use crate::channels::{ChannelManager, CliChatChannel}; use crate::config::{Config, ensure_workspace_dir, expand_path}; +use crate::delivery::{ConversationWriteLocks, DeliveryCoordinator}; use crate::logging; use crate::mcp; use crate::memory::MemoryManager; @@ -27,6 +28,7 @@ pub struct GatewayState { pub channel_manager: ChannelManager, pub storage: Arc, pub task_supervisor: TaskSupervisor, + pub delivery_coordinator: DeliveryCoordinator, pub connection_shutdown: tokio_util::sync::CancellationToken, pub auth: auth::AuthManager, pub uploads: uploads::UploadRegistry, @@ -37,6 +39,7 @@ impl GatewayState { let config_path = crate::config::resolve_default_config_path(); let config = Config::load_default()?; let task_supervisor = TaskSupervisor::new(); + let delivery_coordinator = DeliveryCoordinator::new(ConversationWriteLocks::default()); let connection_shutdown = tokio_util::sync::CancellationToken::new(); let auth = auth::AuthManager::load( config.gateway.require_pairing, @@ -202,6 +205,7 @@ impl GatewayState { channel_manager, storage, task_supervisor, + delivery_coordinator, connection_shutdown, auth, uploads, @@ -335,6 +339,7 @@ impl GatewayState { bus_for_outbound, self.channel_manager.clone(), self.task_supervisor.clone(), + self.delivery_coordinator.write_locks(), ); self.task_supervisor diff --git a/src/lib.rs b/src/lib.rs index 3aea9f1..9412e40 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,7 @@ pub mod bus; pub mod channels; pub mod client; pub mod config; +pub mod delivery; pub mod gateway; pub mod logging; pub mod mcp;