feat: coordinate live turn delivery

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<Self>, snapshot: &TurnSnapshot) -> Result<(), ChannelError>;
-    async fn abort(self: Box<Self>, 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
 

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<MessageBus>,
     channel_manager: ChannelManager,
     task_supervisor: TaskSupervisor,
+    write_locks: ConversationWriteLocks,
 }
 
 impl OutboundDispatcher {
@@ -27,11 +29,13 @@ impl OutboundDispatcher {
         bus: Arc<MessageBus>,
         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"]);
+    }
 }

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<String>,
+    pub metadata: HashMap<String, String>,
+}
+
+#[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<Box<dyn TurnSink>, 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

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;

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<Mutex<HashMap<String, Weak<AsyncMutex<()>>>>>,
+}
+
+impl ConversationWriteLocks {
+    pub fn for_target(&self, channel: &str, chat_id: &str) -> Arc<AsyncMutex<()>> {
+        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<Arc<[Duration]>>,
+    ) -> 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<dyn Channel + Send + Sync>,
+        target: TurnTarget,
+        presentation: PresentationPolicy,
+        snapshots: watch::Receiver<Arc<TurnSnapshot>>,
+    ) -> 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<dyn Channel + Send + Sync>,
+        target: TurnTarget,
+        presentation: PresentationPolicy,
+        snapshots: watch::Receiver<Arc<TurnSnapshot>>,
+    ) -> Result<oneshot::Receiver<Result<(), DeliveryError>>, 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<Arc<TurnSnapshot>>,
+        mut sink: Box<dyn TurnSink>,
+    ) -> 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<AsyncMutex<()>>,
+        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<Vec<TurnSnapshot>>,
+        terminal: TokioMutex<Vec<TurnSnapshot>>,
+        update_started: Notify,
+        release_update: Notify,
+        block_first_update: bool,
+        fail_updates: AtomicUsize,
+        fail_finish: AtomicUsize,
+    }
+
+    struct RecordingSink(Arc<SinkState>);
+
+    struct SinkChannel {
+        state: Arc<SinkState>,
+        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<MessageBus>) -> 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<Box<dyn TurnSink>, 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<SinkState>) -> Box<dyn TurnSink> {
+        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);
+    }
+}

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};

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 { .. }));
+    }
+}

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<crate::storage::Storage>,
     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

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;