From 6230ac8e369fae358878e5a043479e33d30c5828 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Fri, 17 Jul 2026 16:53:21 +0800 Subject: [PATCH] feat: stream turn snapshots to TUI and WebUI --- src/channels/base.rs | 5 + src/channels/cli_chat.rs | 124 +++++++++++++++++++++- src/client/mod.rs | 22 ++++ src/client/tui/app.rs | 100 +++++++++++++++++ src/client/tui/components/chat_history.rs | 98 ++++++++++++++++- src/client/tui/ui.rs | 34 ++++++ src/delivery/coordinator.rs | 50 +++++++++ src/delivery/mod.rs | 2 + src/delivery/service.rs | 64 +++++++++++ src/gateway/mod.rs | 35 +++--- src/gateway/ws.rs | 10 +- src/protocol.rs | 56 ++++++++++ src/session/mod.rs | 2 +- src/session/session.rs | 103 ++++++++++++++---- tests/test_request_format.rs | 5 + webui/src/lib/TurnView.svelte | 45 ++++++++ webui/src/pages/ChatPage.svelte | 36 ++++++- webui/src/styles.css | 15 +++ 18 files changed, 763 insertions(+), 43 deletions(-) create mode 100644 src/delivery/service.rs create mode 100644 webui/src/lib/TurnView.svelte diff --git a/src/channels/base.rs b/src/channels/base.rs index 8fad062..aa460c6 100644 --- a/src/channels/base.rs +++ b/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, ChannelError> { Err(ChannelError::Other(format!( "channel {} does not support turn delivery", diff --git a/src/channels/cli_chat.rs b/src/channels/cli_chat.rs index 031cf4e..337eb31 100644 --- a/src/channels/cli_chat.rs +++ b/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>>, - clients: Mutex>>, + clients: Arc>>>, 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>>>, + 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, 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:?}"), + } + } } diff --git a/src/client/mod.rs b/src/client/mod.rs index c976c2f..50d3dcf 100644 --- a/src/client/mod.rs +++ b/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, diff --git a/src/client/tui/app.rs b/src/client/tui/app.rs index 613791e..642b4da 100644 --- a/src/client/tui/app.rs +++ b/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, + pub completion_status: crate::bus::CompletionStatus, pub attachments: Vec, } @@ -63,6 +66,7 @@ pub struct App { pub selected_session: usize, pub show_archived: bool, pub messages: VecDeque, + pub active_turn: Option, 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) { @@ -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 { #[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) + ); + } } diff --git a/src/client/tui/components/chat_history.rs b/src/client/tui/components/chat_history.rs index 38dbadf..3e48f1a 100644 --- a/src/client/tui/components/chat_history.rs +++ b/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>, 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))) + })); + } + } +} diff --git a/src/client/tui/ui.rs b/src/client/tui/ui.rs index 64e337b..da6a2a7 100644 --- a/src/client/tui/ui.rs +++ b/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(); + } } diff --git a/src/delivery/coordinator.rs b/src/delivery/coordinator.rs index 5769b68..86b2749 100644 --- a/src/delivery/coordinator.rs +++ b/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>, + sink: Box, + ) -> Result>, 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 { diff --git a/src/delivery/mod.rs b/src/delivery/mod.rs index a2c6d2c..10f594e 100644 --- a/src/delivery/mod.rs +++ b/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; diff --git a/src/delivery/service.rs b/src/delivery/service.rs new file mode 100644 index 0000000..5c3303c --- /dev/null +++ b/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>, + ) -> 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(()) + } +} diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 085db4a..c7e0f10 100644 --- a/src/gateway/mod.rs +++ b/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(), - memory_manager, + SessionManagerServices::new( + 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(); diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index 5831e72..064e38d 100644 --- a/src/gateway/ws.rs +++ b/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() { - vec!["file_transfer_v1".to_string()] - } else { - Vec::new() + capabilities: { + let mut capabilities = vec!["turn_snapshots_v1".to_string()]; + if state.uploads.enabled() { + capabilities.push("file_transfer_v1".to_string()); + } + capabilities }, }) .await; diff --git a/src/protocol.rs b/src/protocol.rs index 9a43174..f1221cd 100644 --- a/src/protocol.rs +++ b/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, + #[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, @@ -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 { pub fn serialize_outbound(msg: &WsOutbound) -> Result { 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 + ); + } +} diff --git a/src/session/mod.rs b/src/session/mod.rs index 9ea385d..7523334 100644 --- a/src/session/mod.rs +++ b/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, diff --git a/src/session/session.rs b/src/session/session.rs index e767e40..3ec08d7 100644 --- a/src/session/session.rs +++ b/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, skills_loader: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, + turn_delivery: TurnDeliveryService, } impl Session { @@ -1036,6 +1038,31 @@ pub struct SessionManager { work_manager: Arc, sub_agent_manager: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, + turn_delivery: TurnDeliveryService, +} + +/// Gateway-owned runtime services shared by all Session workers. +pub struct SessionManagerServices { + bus: Arc, + memory_manager: Arc, + task_supervisor: crate::task_supervisor::TaskSupervisor, + turn_delivery: TurnDeliveryService, +} + +impl SessionManagerServices { + pub fn new( + bus: Arc, + memory_manager: Arc, + 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, - bus: Arc, - memory_manager: Arc, + services: SessionManagerServices, browser_config: Option, max_concurrent_background_tasks: usize, - task_supervisor: crate::task_supervisor::TaskSupervisor, ) -> Result { + 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 { - channel: chan2, - chat_id: cid2, - content: response, - reply_to: None, - media: vec![], - metadata: outbound_session_metadata(&response_session_id), - delivery: None, - }; - let _ = bus2.publish_outbound(outbound).await; + if !live_delivery_started { + let outbound = OutboundMessage { + channel: chan2, + chat_id: cid2, + content: response, + reply_to: None, + media: vec![], + metadata: outbound_session_metadata(&response_session_id), + delivery: None, + }; + let _ = bus2.publish_outbound(outbound).await; + } }; tokio::select! { diff --git a/tests/test_request_format.rs b/tests/test_request_format.rs index a6a5be7..0deff88 100644 --- a/tests/test_request_format.rs +++ b/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, .. } => { diff --git a/webui/src/lib/TurnView.svelte b/webui/src/lib/TurnView.svelte new file mode 100644 index 0000000..6573573 --- /dev/null +++ b/webui/src/lib/TurnView.svelte @@ -0,0 +1,45 @@ + + +
+
P
+
+ {#each turn.blocks as block (block.id)} + {#if block.type === "reasoning"} +
+ 思考过程 +
+
+ {:else if block.type === "assistant"} +
+ {:else if block.type === "tool"} +
+ ⌘{block.name}{toolStatus(block.status)} +
+ {#if block.arguments !== null}
{JSON.stringify(block.arguments, null, 2)}
{/if} + {#if block.preview}
{/if} +
+
+ {/if} + {/each} +
+ {#if turn.status === "running"}{/if}{statusLabel(turn.status)} +
+ {#if turn.error}
{turn.error}
{/if} +
+
diff --git a/webui/src/pages/ChatPage.svelte b/webui/src/pages/ChatPage.svelte index 442eefd..c99dc71 100644 --- a/webui/src/pages/ChatPage.svelte +++ b/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 @@
- {#if messages.length === 0} + {#if messages.length === 0 && !activeTurn}

今天想做些什么？

消息与 CLI 客户端使用同一套会话、记忆和工具能力。

{/if} {#each messages as message (message.id)} @@ -383,7 +403,16 @@
{message.role === "user" ? "你" : "P"}
+ {#if message.reasoning_content} +
+ 思考过程 +
+
+ {/if} {#if message.content}
{/if} + {#if message.completion_status && message.completion_status !== "completed"} + {message.completion_status === "cancelled" ? "已停止" : "回复中断"} + {/if} {#if message.attachments?.length}
{#each message.attachments as attachment (`${message.id}:${attachment.index}`)} @@ -408,7 +437,8 @@
{/if} {/each} - {#if thinking}
P
正在思考…
{/if} + {#if activeTurn} + {:else if thinking}
P
正在思考…
{/if}
{ event.preventDefault(); submit(); }} ondragover={(event) => event.preventDefault()} ondrop={dropFiles}> {#if commandSuggestions.length} diff --git a/webui/src/styles.css b/webui/src/styles.css index 63265cd..7de62da 100644 --- a/webui/src/styles.css +++ b/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; }