feat: stream turn snapshots to TUI and WebUI

src/channels/base.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use crate::bus::{BusError, InboundMessage, MessageBus, OutboundMessage};
+use crate::delivery::PresentationPolicy;
 use crate::session::TurnSnapshot;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -78,6 +79,10 @@ pub trait Channel: Send + Sync + 'static {
         LivePolicy::FinalOnly
     }
 
+    fn presentation_policy(&self) -> PresentationPolicy {
+        PresentationPolicy::external(matches!(self.live_policy(), LivePolicy::Snapshot { .. }))
+    }
+
     async fn open_turn(&self, _target: TurnTarget) -> Result<Box<dyn TurnSink>, ChannelError> {
         Err(ChannelError::Other(format!(
             "channel {} does not support turn delivery",

src/channels/cli_chat.rs

@@ -1,6 +1,7 @@
 use async_trait::async_trait;
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::{Mutex, mpsc};
 
 use crate::bus::{ControlMessage, InboundMessage, MessageBus, OutboundMessage};
@@ -8,9 +9,10 @@ use crate::gateway::uploads::UploadRegistry;
 use crate::protocol::{
     HistoryMessage, MessageAttachment, SlashCommandInfo, WsInbound, WsOutbound, parse_inbound,
 };
+use crate::session::TurnSnapshot;
 use crate::session::{SessionCommand, SessionEvent, UnifiedSessionId};
 
-use super::base::{Channel, ChannelError};
+use super::base::{Channel, ChannelError, LivePolicy, TurnSink, TurnTarget};
 
 // ============================================================================
 // Client - Connected CLI client
@@ -34,7 +36,7 @@ impl Client {
 
 pub struct CliChatChannel {
     bus: std::sync::Mutex<Option<Arc<MessageBus>>>,
-    clients: Mutex<HashMap<String, Arc<Client>>>,
+    clients: Arc<Mutex<HashMap<String, Arc<Client>>>>,
     uploads: UploadRegistry,
 }
 
@@ -52,7 +54,7 @@ impl CliChatChannel {
     pub fn with_upload_registry(uploads: UploadRegistry) -> Self {
         Self {
             bus: std::sync::Mutex::new(None),
-            clients: Mutex::new(HashMap::new()),
+            clients: Arc::new(Mutex::new(HashMap::new())),
             uploads,
         }
     }
@@ -451,6 +453,8 @@ impl CliChatChannel {
                                     seq: message.seq,
                                     role: message.role,
                                     content: message.content,
+                                    reasoning_content: message.reasoning_content,
+                                    completion_status: message.completion_status,
                                     created_at: message.created_at,
                                     tool_call_id: message.tool_call_id,
                                     tool_name: message.tool_name,
@@ -803,6 +807,54 @@ impl CliChatChannel {
     }
 }
 
+struct CliChatTurnSink {
+    clients: Arc<Mutex<HashMap<String, Arc<Client>>>>,
+    chat_id: String,
+}
+
+impl CliChatTurnSink {
+    async fn publish(&self, snapshot: &TurnSnapshot) {
+        let client = self.clients.lock().await.get(&self.chat_id).cloned();
+        let Some(client) = client else {
+            return;
+        };
+        if client
+            .sender
+            .send(WsOutbound::TurnUpdated {
+                snapshot: snapshot.clone(),
+            })
+            .await
+            .is_err()
+        {
+            let mut clients = self.clients.lock().await;
+            if clients
+                .get(&self.chat_id)
+                .is_some_and(|registered| Arc::ptr_eq(registered, &client))
+            {
+                clients.remove(&self.chat_id);
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl TurnSink for CliChatTurnSink {
+    async fn update(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
+        self.publish(snapshot).await;
+        Ok(())
+    }
+
+    async fn finish(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
+        self.publish(snapshot).await;
+        Ok(())
+    }
+
+    async fn abort(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
+        self.publish(snapshot).await;
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl Channel for CliChatChannel {
     fn name(&self) -> &str {
@@ -824,6 +876,23 @@ impl Channel for CliChatChannel {
         Ok(())
     }
 
+    fn live_policy(&self) -> LivePolicy {
+        LivePolicy::Snapshot {
+            min_interval: Duration::from_millis(33),
+        }
+    }
+
+    fn presentation_policy(&self) -> crate::delivery::PresentationPolicy {
+        crate::delivery::PresentationPolicy::interactive()
+    }
+
+    async fn open_turn(&self, target: TurnTarget) -> Result<Box<dyn TurnSink>, ChannelError> {
+        Ok(Box::new(CliChatTurnSink {
+            clients: self.clients.clone(),
+            chat_id: target.chat_id,
+        }))
+    }
+
     async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> {
         let client = self.clients.lock().await.get(&msg.chat_id).cloned();
         let Some(client) = client else {
@@ -1026,4 +1095,53 @@ mod tests {
             other => panic!("unexpected outbound: {other:?}"),
         }
     }
+
+    #[tokio::test]
+    async fn turn_sink_sends_the_same_snapshot_shape_for_running_and_terminal_states() {
+        let channel = CliChatChannel::new();
+        let (sender, mut receiver) = mpsc::channel(2);
+        let client = Arc::new(Client {
+            sender,
+            chat_id: "client".into(),
+            current_session_id: Mutex::new(None),
+        });
+        channel.clients.lock().await.insert("client".into(), client);
+        let mut sink = channel
+            .open_turn(TurnTarget {
+                channel: "cli_chat".into(),
+                chat_id: "client".into(),
+                session_id: "cli_chat:client:dialog".into(),
+                reply_to: None,
+                metadata: HashMap::new(),
+            })
+            .await
+            .unwrap();
+        let (controller, emitter, _) =
+            crate::session::TurnController::start("cli_chat:client:dialog", "message");
+        emitter
+            .emit(crate::agent::TurnEvent::TextDelta {
+                iteration: 0,
+                delta: "stream".into(),
+            })
+            .unwrap();
+        let running = controller.snapshot();
+        sink.update(&running).await.unwrap();
+        controller.complete(None);
+        let completed = controller.snapshot();
+        sink.finish(&completed).await.unwrap();
+
+        match receiver.recv().await.unwrap() {
+            WsOutbound::TurnUpdated { snapshot } => {
+                assert_eq!(snapshot.status, crate::session::TurnStatus::Running);
+            }
+            other => panic!("unexpected outbound: {other:?}"),
+        }
+        match receiver.recv().await.unwrap() {
+            WsOutbound::TurnUpdated { snapshot } => {
+                assert_eq!(snapshot.status, crate::session::TurnStatus::Completed);
+                assert!(snapshot.revision > running.revision);
+            }
+            other => panic!("unexpected outbound: {other:?}"),
+        }
+    }
 }

src/client/mod.rs

@@ -253,6 +253,28 @@ async fn run_app(
 
 async fn handle_ws_message(app: &mut App, outbound: WsOutbound) {
     match outbound {
+        WsOutbound::TurnUpdated { snapshot } => {
+            let terminal = snapshot.status != crate::session::TurnStatus::Running;
+            let session_id = snapshot.session_id.clone();
+            if terminal {
+                app.pending_responses = app.pending_responses.saturating_sub(1);
+            }
+            if app.apply_turn_snapshot(snapshot) {
+                if terminal {
+                    app.status_message = None;
+                    if app.current_session_id.as_deref() == Some(&session_id) {
+                        request_history(app, session_id).await;
+                    } else {
+                        app.status_message = Some("另一个会话已完成响应".to_string());
+                    }
+                    request_session_list(app).await;
+                } else {
+                    app.status_message = Some("正在生成回复…".to_string());
+                }
+            } else if terminal && app.current_session_id.as_deref() != Some(&session_id) {
+                app.status_message = Some("另一个会话已完成响应".to_string());
+            }
+        }
         WsOutbound::AssistantResponse {
             id,
             content,

src/client/tui/app.rs

@@ -1,6 +1,7 @@
 use crate::protocol::{
     HistoryMessage, MessageAttachment, SessionSummary, SlashCommandInfo, UploadDescriptor,
 };
+use crate::session::{TurnSnapshot, TurnStatus};
 use std::collections::VecDeque;
 use tokio_tungstenite::tungstenite::Message;
 
@@ -19,6 +20,8 @@ pub struct ChatMessage {
     pub id: String,
     pub role: MessageRole,
     pub content: String,
+    pub reasoning_content: Option<String>,
+    pub completion_status: crate::bus::CompletionStatus,
     pub attachments: Vec<MessageAttachment>,
 }
 
@@ -63,6 +66,7 @@ pub struct App {
     pub selected_session: usize,
     pub show_archived: bool,
     pub messages: VecDeque<ChatMessage>,
+    pub active_turn: Option<TurnSnapshot>,
     pub input: String,
     /// UTF-8 byte offset. It is always maintained at a character boundary.
     pub input_cursor_pos: usize,
@@ -95,6 +99,7 @@ impl App {
             selected_session: 0,
             show_archived: false,
             messages: VecDeque::new(),
+            active_turn: None,
             input: String::new(),
             input_cursor_pos: 0,
             focus: Focus::Input,
@@ -132,6 +137,8 @@ impl App {
             id,
             role,
             content,
+            reasoning_content: None,
+            completion_status: crate::bus::CompletionStatus::Completed,
             attachments,
         });
         while self.messages.len() > MAX_MESSAGES {
@@ -144,6 +151,10 @@ impl App {
         if self.current_session_id.as_deref() != Some(session_id) {
             return;
         }
+        let calibrates_terminal = self.active_turn.as_ref().is_some_and(|turn| {
+            turn.status != TurnStatus::Running
+                && messages.iter().any(|message| message.id == turn.message_id)
+        });
         self.messages = messages
             .into_iter()
             .filter_map(|message| {
@@ -157,6 +168,8 @@ impl App {
                     id: message.id,
                     role,
                     content: message.content,
+                    reasoning_content: message.reasoning_content,
+                    completion_status: message.completion_status,
                     attachments: message.attachments,
                 })
             })
@@ -166,6 +179,9 @@ impl App {
         }
         self.chat_scroll_from_bottom = 0;
         self.status_message = None;
+        if calibrates_terminal {
+            self.active_turn = None;
+        }
     }
 
     pub fn set_sessions(&mut self, sessions: Vec<SessionSummary>) {
@@ -185,6 +201,7 @@ impl App {
         if self.current_session_id != session_id {
             self.current_session_id = session_id;
             self.messages.clear();
+            self.active_turn = None;
             self.pending_uploads.clear();
             self.chat_scroll_from_bottom = 0;
         }
@@ -198,6 +215,21 @@ impl App {
         }
     }
 
+    pub fn apply_turn_snapshot(&mut self, snapshot: TurnSnapshot) -> bool {
+        if self.current_session_id.as_deref() != Some(&snapshot.session_id) {
+            return false;
+        }
+        if let Some(current) = &self.active_turn
+            && current.id == snapshot.id
+            && current.revision >= snapshot.revision
+        {
+            return false;
+        }
+        self.active_turn = Some(snapshot);
+        self.chat_scroll_from_bottom = 0;
+        true
+    }
+
     pub fn current_title(&self) -> &str {
         self.current_session_id
             .as_ref()
@@ -383,6 +415,21 @@ fn next_boundary(value: &str, offset: usize) -> Option<usize> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::session::{TurnId, TurnPhase, TurnState};
+
+    fn turn(revision: u64, status: TurnStatus) -> TurnSnapshot {
+        TurnState {
+            id: TurnId("turn".into()),
+            session_id: "current".into(),
+            message_id: "message".into(),
+            revision,
+            status,
+            phase: TurnPhase::Responding,
+            blocks: Vec::new(),
+            usage: None,
+            error: None,
+        }
+    }
 
     #[test]
     fn unicode_cursor_edits_only_at_character_boundaries() {
@@ -404,4 +451,57 @@ mod tests {
         app.set_history("old", Vec::new());
         assert_eq!(app.messages.len(), 1);
     }
+
+    #[test]
+    fn active_turn_ignores_stale_revisions_and_other_sessions() {
+        let mut app = App::new();
+        app.set_current_session(Some("current".into()));
+
+        assert!(app.apply_turn_snapshot(turn(2, TurnStatus::Running)));
+        assert!(!app.apply_turn_snapshot(turn(1, TurnStatus::Running)));
+        let mut other = turn(3, TurnStatus::Running);
+        other.session_id = "other".into();
+        assert!(!app.apply_turn_snapshot(other));
+        assert_eq!(app.active_turn.as_ref().unwrap().revision, 2);
+    }
+
+    #[test]
+    fn terminal_turn_remains_visible_until_history_calibrates_it() {
+        let mut app = App::new();
+        app.set_current_session(Some("current".into()));
+        app.apply_turn_snapshot(turn(3, TurnStatus::Completed));
+
+        assert!(app.active_turn.is_some());
+        app.set_history(
+            "current",
+            vec![HistoryMessage {
+                id: "message".into(),
+                seq: 1,
+                role: "assistant".into(),
+                content: "done".into(),
+                reasoning_content: None,
+                completion_status: crate::bus::CompletionStatus::Completed,
+                created_at: 1,
+                tool_call_id: None,
+                tool_name: None,
+                tool_calls: None,
+                attachments: Vec::new(),
+            }],
+        );
+        assert!(app.active_turn.is_none());
+    }
+
+    #[test]
+    fn failed_turn_without_a_durable_message_remains_visible_after_history_refresh() {
+        let mut app = App::new();
+        app.set_current_session(Some("current".into()));
+        app.apply_turn_snapshot(turn(3, TurnStatus::Failed));
+
+        app.set_history("current", Vec::new());
+
+        assert_eq!(
+            app.active_turn.as_ref().map(|turn| turn.status),
+            Some(TurnStatus::Failed)
+        );
+    }
 }

src/client/tui/components/chat_history.rs

@@ -1,4 +1,5 @@
 use crate::client::tui::app::{App, MessageRole};
+use crate::session::{ToolStatus, TurnBlock, TurnPhase, TurnStatus};
 use ratatui::{
     Frame,
     layout::Rect,
@@ -23,6 +24,15 @@ pub fn render(frame: &mut Frame, area: Rect, app: &App) {
             label,
             Style::default().fg(color).add_modifier(Modifier::BOLD),
         )));
+        if let Some(reasoning) = &message.reasoning_content {
+            lines.push(Line::from(Span::styled(
+                "思考过程",
+                Style::default()
+                    .fg(Color::DarkGray)
+                    .add_modifier(Modifier::ITALIC),
+            )));
+            push_wrapped(&mut lines, reasoning, content_width, Color::DarkGray);
+        }
         for source_line in message.content.lines() {
             let wrapped = textwrap::wrap(source_line, content_width);
             if wrapped.is_empty() {
@@ -35,6 +45,12 @@ pub fn render(frame: &mut Frame, area: Rect, app: &App) {
                 );
             }
         }
+        if message.completion_status != crate::bus::CompletionStatus::Completed {
+            lines.push(Line::from(Span::styled(
+                format!("[{}]", message.completion_status.as_str()),
+                Style::default().fg(Color::Yellow),
+            )));
+        }
         for attachment in &message.attachments {
             lines.push(Line::from(Span::styled(
                 format!("  [附件 {}] {}", attachment.index + 1, attachment.name),
@@ -43,7 +59,74 @@ pub fn render(frame: &mut Frame, area: Rect, app: &App) {
         }
         lines.push(Line::from(""));
     }
-    if app.pending_responses > 0 {
+    if let Some(turn) = &app.active_turn {
+        lines.push(Line::from(Span::styled(
+            "PicoBot",
+            Style::default()
+                .fg(Color::Green)
+                .add_modifier(Modifier::BOLD),
+        )));
+        for block in &turn.blocks {
+            match block {
+                TurnBlock::Reasoning { text, .. } => {
+                    lines.push(Line::from(Span::styled(
+                        "思考过程",
+                        Style::default()
+                            .fg(Color::DarkGray)
+                            .add_modifier(Modifier::ITALIC),
+                    )));
+                    push_wrapped(&mut lines, text, content_width, Color::DarkGray);
+                }
+                TurnBlock::Assistant { text, .. } => {
+                    push_wrapped(&mut lines, text, content_width, Color::Reset);
+                }
+                TurnBlock::Tool {
+                    name,
+                    status,
+                    preview,
+                    ..
+                } => {
+                    let status = match status {
+                        ToolStatus::Running => "执行中",
+                        ToolStatus::Completed => "已完成",
+                        ToolStatus::Failed => "失败",
+                    };
+                    lines.push(Line::from(Span::styled(
+                        format!("工具 · {name} · {status}"),
+                        Style::default().fg(Color::Magenta),
+                    )));
+                    if let Some(preview) = preview {
+                        push_wrapped(&mut lines, preview, content_width, Color::DarkGray);
+                    }
+                }
+            }
+        }
+        let phase = match turn.phase {
+            TurnPhase::Queued => "排队中",
+            TurnPhase::Reasoning => "思考中",
+            TurnPhase::Responding => "生成中",
+            TurnPhase::Acting => "调用工具中",
+            TurnPhase::Finalizing => "收尾中",
+        };
+        let status = match turn.status {
+            TurnStatus::Running => phase,
+            TurnStatus::Completed => "已完成",
+            TurnStatus::Cancelled => "已停止",
+            TurnStatus::Failed => "失败",
+        };
+        lines.push(Line::from(Span::styled(
+            format!("● {status}"),
+            Style::default().fg(if turn.status == TurnStatus::Failed {
+                Color::Red
+            } else {
+                Color::Cyan
+            }),
+        )));
+        if let Some(error) = &turn.error {
+            push_wrapped(&mut lines, error, content_width, Color::Red);
+        }
+        lines.push(Line::from(""));
+    } else if app.pending_responses > 0 {
         lines.push(Line::from(Span::styled(
             "● 正在思考…",
             Style::default().fg(Color::Cyan),
@@ -66,3 +149,16 @@ pub fn render(frame: &mut Frame, area: Rect, app: &App) {
         area,
     );
 }
+
+fn push_wrapped(lines: &mut Vec<Line<'static>>, text: &str, width: usize, color: Color) {
+    for source_line in text.lines() {
+        let wrapped = textwrap::wrap(source_line, width);
+        if wrapped.is_empty() {
+            lines.push(Line::from(""));
+        } else {
+            lines.extend(wrapped.into_iter().map(|line| {
+                Line::from(Span::styled(line.into_owned(), Style::default().fg(color)))
+            }));
+        }
+    }
+}

src/client/tui/ui.rs

@@ -189,4 +189,38 @@ mod tests {
         );
         terminal.draw(|frame| render_ui(frame, &app)).unwrap();
     }
+
+    #[test]
+    fn active_reasoning_text_and_tool_snapshot_renders_without_panic() {
+        let backend = TestBackend::new(96, 24);
+        let mut terminal = Terminal::new(backend).unwrap();
+        let mut app = App::new();
+        app.set_current_session(Some("session".into()));
+        let (controller, emitter, _) = crate::session::TurnController::start("session", "message");
+        emitter
+            .emit(crate::agent::TurnEvent::ReasoningDelta {
+                iteration: 0,
+                delta: "先检查状态".into(),
+            })
+            .unwrap();
+        emitter
+            .emit(crate::agent::TurnEvent::TextDelta {
+                iteration: 0,
+                delta: "正在处理".into(),
+            })
+            .unwrap();
+        emitter
+            .emit(crate::agent::TurnEvent::ToolStarted {
+                iteration: 0,
+                call: crate::providers::ToolCall {
+                    id: "call".into(),
+                    name: "bash".into(),
+                    arguments: serde_json::json!({"cmd": "pwd"}),
+                },
+            })
+            .unwrap();
+        app.apply_turn_snapshot((*controller.snapshot()).clone());
+
+        terminal.draw(|frame| render_ui(frame, &app)).unwrap();
+    }
 }

src/delivery/coordinator.rs

@@ -19,6 +19,7 @@ const FINAL_RETRY_DELAYS: &[Duration] = &[
 
 #[derive(Debug)]
 pub enum DeliveryError {
+    ChannelNotFound(String),
     OpenFailed(ChannelError),
     SnapshotStreamClosed,
     SupervisorStopping,
@@ -29,6 +30,7 @@ pub enum DeliveryError {
 impl std::fmt::Display for DeliveryError {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
+            Self::ChannelNotFound(channel) => write!(formatter, "channel not found: {channel}"),
             Self::OpenFailed(error) => write!(formatter, "failed to open turn sink: {error}"),
             Self::SnapshotStreamClosed => {
                 formatter.write_str("turn snapshot stream closed before a terminal state")
@@ -77,6 +79,13 @@ pub struct DeliveryCoordinator {
     final_retry_delays: Arc<[Duration]>,
 }
 
+pub(crate) struct SinkRoute {
+    pub channel: String,
+    pub chat_id: String,
+    pub live_policy: LivePolicy,
+    pub presentation: PresentationPolicy,
+}
+
 impl DeliveryCoordinator {
     pub fn new(write_locks: ConversationWriteLocks) -> Self {
         Self {
@@ -142,6 +151,47 @@ impl DeliveryCoordinator {
             let result = coordinator
                 .open_and_deliver(channel, target, presentation, snapshots)
                 .await;
+            if let Err(error) = &result {
+                tracing::error!(error = %error, "Turn delivery failed");
+            }
+            let _ = result_tx.send(result);
+        });
+        if !spawned {
+            return Err(DeliveryError::SupervisorStopping);
+        }
+        Ok(result_rx)
+    }
+
+    pub(crate) fn spawn_sink(
+        &self,
+        supervisor: &TaskSupervisor,
+        route: SinkRoute,
+        snapshots: watch::Receiver<Arc<TurnSnapshot>>,
+        sink: Box<dyn TurnSink>,
+    ) -> Result<oneshot::Receiver<Result<(), DeliveryError>>, DeliveryError> {
+        let SinkRoute {
+            channel,
+            chat_id,
+            live_policy,
+            presentation,
+        } = route;
+        let (result_tx, result_rx) = oneshot::channel();
+        let coordinator = self.clone();
+        let task_name = format!("turn-delivery:{channel}:{chat_id}");
+        let spawned = supervisor.spawn(task_name, async move {
+            let result = coordinator
+                .deliver(
+                    &channel,
+                    &chat_id,
+                    live_policy,
+                    presentation,
+                    snapshots,
+                    sink,
+                )
+                .await;
+            if let Err(error) = &result {
+                tracing::error!(channel, chat_id, error = %error, "Turn delivery failed");
+            }
             let _ = result_tx.send(result);
         });
         if !spawned {

src/delivery/mod.rs

@@ -1,5 +1,7 @@
 mod coordinator;
 mod policy;
+mod service;
 
 pub use coordinator::{ConversationWriteLocks, DeliveryCoordinator, DeliveryError};
 pub use policy::{PresentationPolicy, ReasoningVisibility, ToolVisibility, project_snapshot};
+pub use service::TurnDeliveryService;

src/delivery/service.rs

@@ -0,0 +1,64 @@
+use std::sync::Arc;
+
+use tokio::sync::watch;
+
+use crate::channels::{ChannelManager, TurnTarget};
+use crate::delivery::{DeliveryCoordinator, DeliveryError};
+use crate::session::TurnSnapshot;
+use crate::task_supervisor::TaskSupervisor;
+
+use super::coordinator::SinkRoute;
+
+/// Gateway-owned facade that resolves a target Channel and starts exactly one
+/// TurnSink lifecycle. Session workers depend on this abstraction rather than
+/// on Channel implementations or WebSocket/Feishu protocols.
+#[derive(Clone)]
+pub struct TurnDeliveryService {
+    coordinator: DeliveryCoordinator,
+    channels: ChannelManager,
+    supervisor: TaskSupervisor,
+}
+
+impl TurnDeliveryService {
+    pub fn new(
+        coordinator: DeliveryCoordinator,
+        channels: ChannelManager,
+        supervisor: TaskSupervisor,
+    ) -> Self {
+        Self {
+            coordinator,
+            channels,
+            supervisor,
+        }
+    }
+
+    pub async fn start(
+        &self,
+        target: TurnTarget,
+        snapshots: watch::Receiver<Arc<TurnSnapshot>>,
+    ) -> Result<(), DeliveryError> {
+        let channel = self
+            .channels
+            .get_channel(&target.channel)
+            .await
+            .ok_or_else(|| DeliveryError::ChannelNotFound(target.channel.clone()))?;
+        let live_policy = channel.live_policy();
+        let presentation = channel.presentation_policy();
+        let sink = channel
+            .open_turn(target.clone())
+            .await
+            .map_err(DeliveryError::OpenFailed)?;
+        let _result = self.coordinator.spawn_sink(
+            &self.supervisor,
+            SinkRoute {
+                channel: target.channel,
+                chat_id: target.chat_id,
+                live_policy,
+                presentation,
+            },
+            snapshots,
+            sink,
+        )?;
+        Ok(())
+    }
+}

src/gateway/mod.rs

@@ -12,12 +12,12 @@ 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::delivery::{ConversationWriteLocks, DeliveryCoordinator, TurnDeliveryService};
 use crate::logging;
 use crate::mcp;
 use crate::memory::MemoryManager;
 use crate::scheduler::Scheduler;
-use crate::session::SessionManager;
+use crate::session::{SessionManager, SessionManagerServices};
 use crate::task_supervisor::TaskSupervisor;
 
 pub struct GatewayState {
@@ -105,6 +105,20 @@ impl GatewayState {
         // Create MessageBus first (shared by SessionManager and ChannelManager)
         let bus = MessageBus::new(100);
 
+        // Channels are resolved by TurnDeliveryService, while Session workers
+        // depend only on that protocol-neutral delivery facade.
+        let cli_chat_channel = Arc::new(CliChatChannel::with_upload_registry(uploads.clone()));
+        let channel_manager = ChannelManager::with_bus(cli_chat_channel, bus.clone());
+        channel_manager
+            .init(&config, workspace_path.clone())
+            .await
+            .map_err(|e| format!("Failed to init channels: {}", e))?;
+        let turn_delivery = TurnDeliveryService::new(
+            delivery_coordinator.clone(),
+            channel_manager.clone(),
+            task_supervisor.clone(),
+        );
+
         let browser_config = if config.browser.enabled {
             Some(config.browser.clone())
         } else {
@@ -115,22 +129,17 @@ impl GatewayState {
         let session_manager = SessionManager::new(
             provider_config.clone(),
             storage.clone(),
-            bus.clone(),
+            SessionManagerServices::new(
-            memory_manager,
+                bus.clone(),
+                memory_manager,
+                task_supervisor.clone(),
+                turn_delivery,
+            ),
             browser_config,
             config.gateway.max_concurrent_background_tasks,
-            task_supervisor.clone(),
         )?;
         let session_manager = Arc::new(session_manager);
 
-        // Create ChannelManager and init channels
-        let cli_chat_channel = Arc::new(CliChatChannel::with_upload_registry(uploads.clone()));
-        let channel_manager = ChannelManager::with_bus(cli_chat_channel, bus);
-        channel_manager
-            .init(&config, workspace_path.clone())
-            .await
-            .map_err(|e| format!("Failed to init channels: {}", e))?;
-
         // Register send_message tool with available channel names
         let available_channels = channel_manager.list_channel_names().await;
         let valid_channels = available_channels.clone();

src/gateway/ws.rs

@@ -56,10 +56,12 @@ async fn handle_socket(
     let _ = sender
         .send(WsOutbound::SessionEstablished {
             session_id: session_id.clone(),
-            capabilities: if state.uploads.enabled() {
+            capabilities: {
-                vec!["file_transfer_v1".to_string()]
+                let mut capabilities = vec!["turn_snapshots_v1".to_string()];
-            } else {
+                if state.uploads.enabled() {
-                Vec::new()
+                    capabilities.push("file_transfer_v1".to_string());
+                }
+                capabilities
             },
         })
         .await;

src/protocol.rs

@@ -62,6 +62,10 @@ pub struct HistoryMessage {
     pub seq: i64,
     pub role: String,
     pub content: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub reasoning_content: Option<String>,
+    #[serde(default)]
+    pub completion_status: crate::bus::CompletionStatus,
     pub created_at: i64,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub tool_call_id: Option<String>,
@@ -140,6 +144,10 @@ pub enum WsInbound {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "type")]
 pub enum WsOutbound {
+    #[serde(rename = "turn_updated")]
+    TurnUpdated {
+        snapshot: crate::session::TurnSnapshot,
+    },
     #[serde(rename = "assistant_response")]
     AssistantResponse {
         id: String,
@@ -222,3 +230,51 @@ pub fn serialize_inbound(msg: &WsInbound) -> Result<String, serde_json::Error> {
 pub fn serialize_outbound(msg: &WsOutbound) -> Result<String, serde_json::Error> {
     serde_json::to_string(msg)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::session::{TurnId, TurnPhase, TurnState, TurnStatus};
+
+    #[test]
+    fn turn_updated_serializes_as_one_complete_snapshot_frame() {
+        let frame = WsOutbound::TurnUpdated {
+            snapshot: TurnState {
+                id: TurnId("turn-1".into()),
+                session_id: "cli_chat:client:dialog".into(),
+                message_id: "message-1".into(),
+                revision: 7,
+                status: TurnStatus::Running,
+                phase: TurnPhase::Responding,
+                blocks: Vec::new(),
+                usage: None,
+                error: None,
+            },
+        };
+
+        let json = serialize_outbound(&frame).unwrap();
+        let value: serde_json::Value = serde_json::from_str(&json).unwrap();
+        assert_eq!(value["type"], "turn_updated");
+        assert_eq!(value["snapshot"]["revision"], 7);
+        assert_eq!(value["snapshot"]["status"], "running");
+    }
+
+    #[test]
+    fn history_defaults_new_reasoning_fields_for_old_frames() {
+        let message: HistoryMessage = serde_json::from_value(serde_json::json!({
+            "id": "message",
+            "seq": 1,
+            "role": "assistant",
+            "content": "answer",
+            "created_at": 1,
+            "attachments": []
+        }))
+        .unwrap();
+
+        assert_eq!(message.reasoning_content, None);
+        assert_eq!(
+            message.completion_status,
+            crate::bus::CompletionStatus::Completed
+        );
+    }
+}

src/session/mod.rs

@@ -12,7 +12,7 @@ pub mod turn;
 pub use commands::SessionCommand;
 pub use error::SessionError;
 pub use events::{DialogInfo, SessionEvent};
-pub use session::{SLASH_COMMANDS, Session, SessionManager, SlashCommand};
+pub use session::{SLASH_COMMANDS, Session, SessionManager, SessionManagerServices, SlashCommand};
 pub use session_id::UnifiedSessionId;
 pub use turn::{
     BlockId, ToolStatus, TurnBlock, TurnController, TurnId, TurnPhase, TurnSnapshot, TurnState,

src/session/session.rs

@@ -44,6 +44,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;
 
 /// Check if an LLM error message indicates a context window overflow.
 fn is_context_overflow_error(msg: &str) -> bool {
@@ -241,6 +242,7 @@ struct AgentWorkerDeps {
     work_manager: Arc<crate::work::WorkManager>,
     skills_loader: Arc<SkillsLoader>,
     task_supervisor: crate::task_supervisor::TaskSupervisor,
+    turn_delivery: TurnDeliveryService,
 }
 
 impl Session {
@@ -1036,6 +1038,31 @@ pub struct SessionManager {
     work_manager: Arc<crate::work::WorkManager>,
     sub_agent_manager: Arc<crate::agent::SubAgentManager>,
     task_supervisor: crate::task_supervisor::TaskSupervisor,
+    turn_delivery: TurnDeliveryService,
+}
+
+/// Gateway-owned runtime services shared by all Session workers.
+pub struct SessionManagerServices {
+    bus: Arc<MessageBus>,
+    memory_manager: Arc<crate::memory::MemoryManager>,
+    task_supervisor: crate::task_supervisor::TaskSupervisor,
+    turn_delivery: TurnDeliveryService,
+}
+
+impl SessionManagerServices {
+    pub fn new(
+        bus: Arc<MessageBus>,
+        memory_manager: Arc<crate::memory::MemoryManager>,
+        task_supervisor: crate::task_supervisor::TaskSupervisor,
+        turn_delivery: TurnDeliveryService,
+    ) -> Self {
+        Self {
+            bus,
+            memory_manager,
+            task_supervisor,
+            turn_delivery,
+        }
+    }
 }
 
 struct SessionManagerInner {
@@ -1149,18 +1176,23 @@ impl SessionManager {
             work_manager: self.work_manager.clone(),
             skills_loader: self.skills_loader.clone(),
             task_supervisor: self.task_supervisor.clone(),
+            turn_delivery: self.turn_delivery.clone(),
         }
     }
 
     pub fn new(
         provider_config: LLMProviderConfig,
         storage: Arc<Storage>,
-        bus: Arc<MessageBus>,
+        services: SessionManagerServices,
-        memory_manager: Arc<crate::memory::MemoryManager>,
         browser_config: Option<BrowserConfig>,
         max_concurrent_background_tasks: usize,
-        task_supervisor: crate::task_supervisor::TaskSupervisor,
     ) -> Result<Self, AgentError> {
+        let SessionManagerServices {
+            bus,
+            memory_manager,
+            task_supervisor,
+            turn_delivery,
+        } = services;
         let mut skills_loader = SkillsLoader::new();
         skills_loader.load_skills();
         skills_loader.set_workspace_skills_dir(provider_config.workspace_dir.clone());
@@ -1247,6 +1279,7 @@ impl SessionManager {
             work_manager,
             sub_agent_manager,
             task_supervisor,
+            turn_delivery,
         })
     }
 
@@ -2309,6 +2342,7 @@ fn spawn_agent_worker(
         work_manager,
         skills_loader,
         task_supervisor,
+        turn_delivery,
     } = deps;
     let worker_supervisor = task_supervisor.clone();
     task_supervisor.spawn(format!("session-worker:{unified_str}"), async move {
@@ -2514,12 +2548,35 @@ fn spawn_agent_worker(
                     Session::append_runtime_context_to_user_message(last_msg, &runtime_context);
                 }
 
-                let (turn_controller, turn_emitter, _turn_receiver) = TurnController::start(
+                let (turn_controller, turn_emitter, turn_receiver) = TurnController::start(
                     unified_str.clone(),
                     uuid::Uuid::new_v4().to_string(),
                 );
                 let initial_turn = turn_controller.snapshot();
                 let active_turn_id = initial_turn.id.0.clone();
+                let live_delivery_started = match turn_delivery
+                    .start(
+                        crate::channels::TurnTarget {
+                            channel: task_chan.clone(),
+                            chat_id: task_cid.clone(),
+                            session_id: unified_str.clone(),
+                            reply_to: None,
+                            metadata: outbound_session_metadata(&unified_str),
+                        },
+                        turn_receiver,
+                    )
+                    .await
+                {
+                    Ok(()) => true,
+                    Err(error) => {
+                        tracing::debug!(
+                            channel = %task_chan,
+                            error = %error,
+                            "Live turn delivery unavailable; using ordinary final delivery"
+                        );
+                        false
+                    }
+                };
                 {
                     let mut guard = session.lock().await;
                     if guard.worker_generation != worker_gen || guard.state_version != base_version {
@@ -2604,7 +2661,9 @@ fn spawn_agent_worker(
                                             ),
                                             delivery: None,
                                         };
-                                        let _ = bus2.publish_outbound(err_outbound).await;
+                                        if !live_delivery_started {
+                                            let _ = bus2.publish_outbound(err_outbound).await;
+                                        }
                                         return;
                                     }
                                 };
@@ -2676,7 +2735,9 @@ fn spawn_agent_worker(
                                         metadata: outbound_session_metadata(&response_session_id),
                                         delivery: None,
                                     };
-                                    let _ = bus2.publish_outbound(err_outbound).await;
+                                    if !live_delivery_started {
+                                        let _ = bus2.publish_outbound(err_outbound).await;
+                                    }
                                     return;
                                 }
                             }
@@ -2698,7 +2759,9 @@ fn spawn_agent_worker(
                                 metadata: outbound_session_metadata(&response_session_id),
                                 delivery: None,
                             };
-                            let _ = bus2.publish_outbound(err_outbound).await;
+                            if !live_delivery_started {
+                                let _ = bus2.publish_outbound(err_outbound).await;
+                            }
                             return;
                         }
                     };
@@ -2747,7 +2810,9 @@ fn spawn_agent_worker(
                             metadata: outbound_session_metadata(&response_session_id),
                             delivery: None,
                         };
-                        let _ = bus2.publish_outbound(err_outbound).await;
+                        if !live_delivery_started {
+                            let _ = bus2.publish_outbound(err_outbound).await;
+                        }
                         return;
                     };
 
@@ -2755,16 +2820,18 @@ fn spawn_agent_worker(
                         tracing::warn!("failed to generate title: {}", e);
                     }
 
-                    let outbound = OutboundMessage {
+                    if !live_delivery_started {
-                        channel: chan2,
+                        let outbound = OutboundMessage {
-                        chat_id: cid2,
+                            channel: chan2,
-                        content: response,
+                            chat_id: cid2,
-                        reply_to: None,
+                            content: response,
-                        media: vec![],
+                            reply_to: None,
-                        metadata: outbound_session_metadata(&response_session_id),
+                            media: vec![],
-                        delivery: None,
+                            metadata: outbound_session_metadata(&response_session_id),
-                    };
+                            delivery: None,
-                    let _ = bus2.publish_outbound(outbound).await;
+                        };
+                        let _ = bus2.publish_outbound(outbound).await;
+                    }
                 };
 
                 tokio::select! {

tests/test_request_format.rs

@@ -135,6 +135,8 @@ fn test_bounded_session_history_protocol() {
             seq: 1,
             role: "user".to_string(),
             content: "你好".to_string(),
+            reasoning_content: None,
+            completion_status: picobot::bus::CompletionStatus::Completed,
             created_at: 123,
             tool_call_id: None,
             tool_name: None,
@@ -182,6 +184,8 @@ fn test_session_history_preserves_tool_call_metadata() {
             seq: 2,
             role: "assistant".to_string(),
             content: String::new(),
+            reasoning_content: Some("checking".to_string()),
+            completion_status: picobot::bus::CompletionStatus::Completed,
             created_at: 124,
             tool_call_id: None,
             tool_name: None,
@@ -197,6 +201,7 @@ fn test_session_history_preserves_tool_call_metadata() {
     let json = serde_json::to_string(&outbound).unwrap();
     assert!(json.contains(r#""tool_calls""#));
     assert!(json.contains(r#""read_file""#));
+    assert!(json.contains(r#""reasoning_content":"checking""#));
     let decoded: WsOutbound = serde_json::from_str(&json).unwrap();
     match decoded {
         WsOutbound::SessionHistory { messages, .. } => {

webui/src/lib/TurnView.svelte

@@ -0,0 +1,45 @@
+<script>
+  import Markdown from "./Markdown.svelte";
+
+  let { turn } = $props();
+
+  function phaseLabel(value) {
+    return ({ queued: "排队中", reasoning: "思考中", responding: "生成中", acting: "调用工具中", finalizing: "收尾中" })[value] || value;
+  }
+
+  function statusLabel(value) {
+    return ({ running: phaseLabel(turn.phase), completed: "已完成", cancelled: "已停止", failed: "失败" })[value] || value;
+  }
+
+  function toolStatus(value) {
+    return ({ running: "执行中", completed: "已完成", failed: "失败" })[value] || value;
+  }
+</script>
+
+<div class="message assistant active-turn">
+  <div class="avatar">P</div>
+  <div class="message-content">
+    {#each turn.blocks as block (block.id)}
+      {#if block.type === "reasoning"}
+        <details class="reasoning-block">
+          <summary><span class="pulse"></span>思考过程</summary>
+          <div class="reasoning-content"><Markdown content={block.text} /></div>
+        </details>
+      {:else if block.type === "assistant"}
+        <div class="bubble streaming"><Markdown content={block.text} /></div>
+      {:else if block.type === "tool"}
+        <details class="live-tool">
+          <summary><span>⌘</span><strong>{block.name}</strong><small>{toolStatus(block.status)}</small></summary>
+          <div class="live-tool-details">
+            {#if block.arguments !== null}<pre>{JSON.stringify(block.arguments, null, 2)}</pre>{/if}
+            {#if block.preview}<div class="tool-result"><Markdown content={block.preview} /></div>{/if}
+          </div>
+        </details>
+      {/if}
+    {/each}
+    <div class:failed={turn.status === "failed"} class="turn-status">
+      {#if turn.status === "running"}<span class="pulse"></span>{/if}{statusLabel(turn.status)}
+    </div>
+    {#if turn.error}<div class="turn-error">{turn.error}</div>{/if}
+  </div>
+</div>

webui/src/pages/ChatPage.svelte

@@ -4,6 +4,7 @@
   import { clientId, formatTime } from "../lib/api.js";
   import Markdown from "../lib/Markdown.svelte";
   import ToolCallCard from "../lib/ToolCallCard.svelte";
+  import TurnView from "../lib/TurnView.svelte";
 
   let { notify } = $props();
   let socket = $state(null);
@@ -17,6 +18,7 @@
   let selectedCommand = $state(0);
   let commandMenuDismissed = $state(false);
   let thinking = $state(false);
+  let activeTurn = $state(null);
   let pendingUploads = $state([]);
   let fileInput;
   let plansBySession = $state({});
@@ -75,7 +77,7 @@
 
   function handleFrame(frame) {
     switch (frame.type) {
-      case "session_established": currentId = frame.session_id; break;
+      case "session_established": currentId = frame.session_id; activeTurn = null; break;
       case "session_list":
         sessions = frame.sessions || [];
         if (frame.current_session_id) currentId = frame.current_session_id;
@@ -84,16 +86,19 @@
       case "session_created":
         currentId = frame.session_id;
         messages = [];
+        activeTurn = null;
         send({ type: "list_sessions", include_archived: false });
         break;
       case "session_loaded":
         currentId = frame.session_id;
+        activeTurn = null;
         send({ type: "get_session_history", session_id: currentId, limit: 1000 });
         send({ type: "get_session_plan", session_id: currentId });
         break;
       case "session_history":
         if (frame.session_id === currentId) {
           messages = frame.messages || [];
+          if (activeTurn?.status !== "running" && messages.some((message) => message.id === activeTurn?.message_id)) activeTurn = null;
           scrollToBottom();
         }
         break;
@@ -133,6 +138,19 @@
         }
         send({ type: "list_sessions", include_archived: false });
         break;
+      case "turn_updated": {
+        const next = frame.snapshot;
+        if (!next || next.session_id !== currentId) break;
+        if (activeTurn?.id === next.id && activeTurn.revision >= next.revision) break;
+        activeTurn = next;
+        thinking = next.status === "running";
+        scrollToBottom();
+        if (next.status !== "running") {
+          send({ type: "get_session_history", session_id: currentId, limit: 1000 });
+          send({ type: "list_sessions", include_archived: false });
+        }
+        break;
+      }
       case "system_notification":
         if (!frame.session_id || frame.session_id === currentId) appendMessage("assistant", frame.content);
         break;
@@ -156,6 +174,8 @@
     currentId = id;
     clearPendingUploads();
     messages = [];
+    activeTurn = null;
+    thinking = false;
     todoOpen = Boolean(unseenPlanSessions[id]);
     unseenPlanSessions[id] = false;
     send({ type: "load_session", session_id: id });
@@ -375,7 +395,7 @@
       </div>
     </div>
     <div class="messages" bind:this={messageBox}>
-      {#if messages.length === 0}
+      {#if messages.length === 0 && !activeTurn}
         <div class="empty"><div class="empty-logo">P</div><h2>今天想做些什么？</h2><p>消息与 CLI 客户端使用同一套会话、记忆和工具能力。</p></div>
       {/if}
       {#each messages as message (message.id)}
@@ -383,7 +403,16 @@
           <div class:user={message.role === "user"} class:assistant={message.role !== "user"} class:has-tools={message.tool_calls?.length} class="message">
             <div class="avatar">{message.role === "user" ? "你" : "P"}</div>
             <div class="message-content">
+              {#if message.reasoning_content}
+                <details class="reasoning-block historical">
+                  <summary>思考过程</summary>
+                  <div class="reasoning-content"><Markdown content={message.reasoning_content} /></div>
+                </details>
+              {/if}
               {#if message.content}<div class="bubble"><Markdown content={message.content} /></div>{/if}
+              {#if message.completion_status && message.completion_status !== "completed"}
+                <small class="completion-status">{message.completion_status === "cancelled" ? "已停止" : "回复中断"}</small>
+              {/if}
               {#if message.attachments?.length}
                 <div class="message-attachments">
                   {#each message.attachments as attachment (`${message.id}:${attachment.index}`)}
@@ -408,7 +437,8 @@
           </div>
         {/if}
       {/each}
-      {#if thinking}<div class="message assistant typing"><div class="avatar">P</div><div class="bubble"><span class="pulse"></span>正在思考…</div></div>{/if}
+      {#if activeTurn}<TurnView turn={activeTurn} />
+      {:else if thinking}<div class="message assistant typing"><div class="avatar">P</div><div class="bubble"><span class="pulse"></span>正在思考…</div></div>{/if}
     </div>
     <form class="composer" onsubmit={(event) => { event.preventDefault(); submit(); }} ondragover={(event) => event.preventDefault()} ondrop={dropFiles}>
       {#if commandSuggestions.length}

webui/src/styles.css

@@ -150,6 +150,21 @@ main { min-width: 0; height: 100vh; display: flex; flex-direction: column; }
 .attachment-card strong { font-size: 12px; }.attachment-card small { margin-top: 3px; color: var(--muted); font-size: 9px; }
 .attachment-card a { width: 28px; height: 28px; display: grid; place-items: center; border: 1px solid var(--line); border-radius: 7px; color: var(--accent); text-decoration: none; }
 .typing .bubble { color: var(--muted); }
+.active-turn .message-content { width: min(82%, 760px); }
+.streaming { border-color: var(--accent-border); }
+.reasoning-block, .live-tool { width: min(100%, 680px); overflow: hidden; border: 1px solid var(--line); border-radius: 10px; background: var(--panel); }
+.reasoning-block summary, .live-tool summary { padding: 9px 12px; color: var(--muted); font-size: 11px; cursor: pointer; user-select: none; }
+.reasoning-block summary::marker, .live-tool summary::marker { color: var(--accent); }
+.reasoning-content { padding: 10px 13px; border-top: 1px solid var(--line); color: var(--muted); font-size: 12px; }
+.reasoning-block.historical { border-style: dashed; }
+.live-tool summary { display: grid; grid-template-columns: 24px 1fr auto; align-items: center; gap: 8px; }
+.live-tool summary strong { color: var(--text); font: 600 12px/1.4 ui-monospace, SFMono-Regular, Consolas, monospace; }
+.live-tool summary small { color: var(--muted); }
+.live-tool-details { display: grid; gap: 8px; padding: 10px 12px; border-top: 1px solid var(--line); }
+.live-tool-details pre { max-height: 260px; margin: 0; padding: 10px; overflow: auto; border-radius: 8px; color: var(--text-soft); background: var(--code-bg); font: 11px/1.5 ui-monospace, SFMono-Regular, Consolas, monospace; white-space: pre-wrap; }
+.turn-status, .completion-status { color: var(--muted); font-size: 10px; }
+.turn-status.failed, .turn-error { color: var(--danger); }
+.turn-error { padding: 8px 10px; border-radius: 8px; background: var(--danger-soft); font-size: 11px; }
 .pulse { display: inline-block; width: 6px; height: 6px; margin-right: 8px; border-radius: 50%; background: var(--accent); animation: pulse 1.1s infinite; }
 @keyframes pulse { 50% { opacity: .25; transform: scale(.8); } }
 .markdown-body { min-width: 0; }