fix(agentic): preserve late stream chunks before dialog completion

src/apps/desktop/src/lib.rs

@@ -887,9 +887,12 @@ fn start_event_loop_with_transport(
     tokio::spawn(async move {
         loop {
             event_queue.wait_for_events().await;
-            let batch = event_queue.dequeue_batch(10).await;
+            loop {
+                let batch = event_queue.dequeue_configured_batch().await;
+                if batch.is_empty() {
+                    break;
+                }
 
-            if !batch.is_empty() {
                 for envelope in batch {
                     // Route to internal subscribers (e.g. RemoteSessionStateTracker)
                     // sequentially so that text chunks are appended in order.

src/crates/core/src/agentic/events/queue.rs

@@ -131,6 +131,11 @@ impl EventQueue {
         batch
     }
 
+    /// Dequeue a batch using the queue's configured batch size.
+    pub async fn dequeue_configured_batch(&self) -> Vec<EventEnvelope> {
+        self.dequeue_batch(self.config.batch_size).await
+    }
+
     /// Clear all events for a session
     pub async fn clear_session(&self, session_id: &str) -> BitFunResult<()> {
         // Remove all events for this session from the queue

src/crates/core/src/agentic/execution/execution_engine.rs

@@ -1104,18 +1104,20 @@ impl ExecutionEngine {
         // Emit dialog turn completed event
         debug!("Preparing to send DialogTurnCompleted event");
 
-        self.emit_event(
-            AgenticEvent::DialogTurnCompleted {
-                session_id: context.session_id.clone(),
-                turn_id: context.dialog_turn_id.clone(),
-                total_rounds: round_index + 1,
-                total_tools,
-                duration_ms,
-                subagent_parent_info: event_subagent_parent_info,
-            },
-            EventPriority::High,
-        )
-        .await;
+        let _ = self
+            .event_queue
+            .enqueue(
+                AgenticEvent::DialogTurnCompleted {
+                    session_id: context.session_id.clone(),
+                    turn_id: context.dialog_turn_id.clone(),
+                    total_rounds: round_index + 1,
+                    total_tools,
+                    duration_ms,
+                    subagent_parent_info: event_subagent_parent_info,
+                },
+                None,
+            )
+            .await;
 
         debug!("DialogTurnCompleted event sent");
 

src/crates/core/src/agentic/execution/stream_processor.rs

@@ -491,7 +491,7 @@ impl StreamProcessor {
                                 },
                                 subagent_parent_info: ctx.event_subagent_parent_info.clone(),
                             },
-                            Some(EventPriority::Normal),
+                            None,
                         )
                         .await;
                 } else if ctx.tool_call_buffer.tool_name.is_empty() {
@@ -522,7 +522,7 @@ impl StreamProcessor {
                             },
                             subagent_parent_info: ctx.event_subagent_parent_info.clone(),
                         },
-                        Some(EventPriority::Normal),
+                        None,
                     )
                     .await;
             }
@@ -557,7 +557,7 @@ impl StreamProcessor {
                     text,
                     subagent_parent_info: ctx.event_subagent_parent_info.clone(),
                 },
-                Some(EventPriority::Normal),
+                None,
             )
             .await;
     }
@@ -582,7 +582,7 @@ impl StreamProcessor {
                     is_end: false,
                     subagent_parent_info: ctx.event_subagent_parent_info.clone(),
                 },
-                Some(EventPriority::Normal),
+                None,
             )
             .await;
     }

src/crates/core/src/agentic/tools/pipeline/state_manager.rs

@@ -4,7 +4,7 @@
 
 use super::types::ToolTask;
 use crate::agentic::core::ToolExecutionState;
-use crate::agentic::events::{AgenticEvent, EventPriority, EventQueue, ToolEventData};
+use crate::agentic::events::{AgenticEvent, EventQueue, ToolEventData};
 use dashmap::DashMap;
 use log::debug;
 use std::sync::Arc;
@@ -215,25 +215,6 @@ impl ToolStateManager {
             },
         };
 
-        // Determine priority based on tool event type
-        let priority = match &task.state {
-            // Critical state change: High priority (user needs to see immediately)
-            ToolExecutionState::Running { .. }                  // Start execution
-            | ToolExecutionState::AwaitingConfirmation { .. }   // Need confirmation
-            | ToolExecutionState::Completed { .. }              // Completed
-            | ToolExecutionState::Failed { .. }                 // Failed
-            => EventPriority::High,
-
-            // Cancel event: Critical priority (need immediate feedback)
-            ToolExecutionState::Cancelled { .. } => EventPriority::Critical,
-
-            // Progress state: Normal priority (avoid blocking critical events)
-            ToolExecutionState::Queued { .. }
-            | ToolExecutionState::Waiting { .. }
-            | ToolExecutionState::Streaming { .. }
-            => EventPriority::Normal,
-        };
-
         let event_subagent_parent_info = task.context.subagent_parent_info.map(|info| info.into());
         let event = AgenticEvent::ToolEvent {
             session_id: task.context.session_id,
@@ -242,7 +223,7 @@ impl ToolStateManager {
             subagent_parent_info: event_subagent_parent_info,
         };
 
-        let _ = self.event_queue.enqueue(event, Some(priority)).await;
+        let _ = self.event_queue.enqueue(event, None).await;
     }
 
     /// Get statistics

src/crates/events/src/agentic.rs

@@ -339,21 +339,46 @@ impl AgenticEvent {
 
             Self::SessionStateChanged { .. }
             | Self::SessionTitleGenerated { .. }
-            | Self::DialogTurnCompleted { .. }
             | Self::ContextCompressionFailed { .. } => AgenticEventPriority::High,
 
             Self::ImageAnalysisStarted { .. }
             | Self::ImageAnalysisCompleted { .. }
             | Self::TextChunk { .. }
             | Self::ThinkingChunk { .. }
-            | Self::ToolEvent { .. }
             | Self::ModelRoundStarted { .. }
             | Self::ModelRoundCompleted { .. }
             | Self::TokenUsageUpdated { .. }
+            | Self::DialogTurnCompleted { .. }
             | Self::ContextCompressionStarted { .. }
             | Self::ContextCompressionCompleted { .. } => AgenticEventPriority::Normal,
 
+            Self::ToolEvent { tool_event, .. } => tool_event.default_priority(),
+
             _ => AgenticEventPriority::Low,
         }
     }
 }
+
+impl ToolEventData {
+    /// Get the default priority for a specific tool event variant.
+    pub fn default_priority(&self) -> AgenticEventPriority {
+        match self {
+            Self::Cancelled { .. } => AgenticEventPriority::Critical,
+
+            Self::Started { .. }
+            | Self::Completed { .. }
+            | Self::Failed { .. }
+            | Self::ConfirmationNeeded { .. } => AgenticEventPriority::High,
+
+            Self::EarlyDetected { .. }
+            | Self::ParamsPartial { .. }
+            | Self::Queued { .. }
+            | Self::Waiting { .. }
+            | Self::Progress { .. }
+            | Self::Streaming { .. }
+            | Self::StreamChunk { .. }
+            | Self::Confirmed { .. }
+            | Self::Rejected { .. } => AgenticEventPriority::Normal,
+        }
+    }
+}

src/web-ui/src/app/components/NavPanel/sections/sessions/SessionsSection.tsx

@@ -99,7 +99,11 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
       const running = new Set<string>();
       for (const session of flowChatState.sessions.values()) {
         const machine = stateMachineManager.get(session.sessionId);
-        if (machine && machine.getCurrentState() === SessionExecutionState.PROCESSING) {
+        if (
+          machine &&
+          (machine.getCurrentState() === SessionExecutionState.PROCESSING ||
+            machine.getCurrentState() === SessionExecutionState.FINISHING)
+        ) {
           running.add(session.sessionId);
         }
       }

src/web-ui/src/app/layout/FloatingMiniChat.tsx

@@ -70,7 +70,10 @@ export const FloatingMiniChat: React.FC = () => {
     }
 
     const lastTurn = activeSession.dialogTurns[activeSession.dialogTurns.length - 1];
-    const isStreaming = lastTurn.status === 'processing' || lastTurn.status === 'image_analyzing';
+    const isStreaming =
+      lastTurn.status === 'processing' ||
+      lastTurn.status === 'finishing' ||
+      lastTurn.status === 'image_analyzing';
     return { isStreaming };
   }, [flowChatState]);
 

src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.tsx

@@ -143,7 +143,10 @@ export const BtwSessionPanel: React.FC<BtwSessionPanelProps> = ({
   const lastModelRound = lastDialogTurn?.modelRounds[lastDialogTurn.modelRounds.length - 1];
   const lastItem = lastModelRound?.items[lastModelRound.items.length - 1];
   const lastItemContent = lastItem && 'content' in lastItem ? String((lastItem as any).content || '') : '';
-  const isTurnProcessing = lastDialogTurn?.status === 'processing' || lastDialogTurn?.status === 'image_analyzing';
+  const isTurnProcessing =
+    lastDialogTurn?.status === 'processing' ||
+    lastDialogTurn?.status === 'finishing' ||
+    lastDialogTurn?.status === 'image_analyzing';
   const [isContentGrowing, setIsContentGrowing] = useState(true);
   const lastContentRef = useRef(lastItemContent);
   const contentTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);

src/web-ui/src/flow_chat/components/modern/VirtualMessageList.tsx

@@ -1447,6 +1447,7 @@ export const VirtualMessageList = forwardRef<VirtualMessageListRef>((_, ref) =>
 
     if (
       lastDialogTurn.status === 'processing' ||
+      lastDialogTurn.status === 'finishing' ||
       lastDialogTurn.status === 'image_analyzing'
     ) {
       return true;
@@ -1781,8 +1782,10 @@ export const VirtualMessageList = forwardRef<VirtualMessageListRef>((_, ref) =>
       : undefined;
 
     const content = lastItem && 'content' in lastItem ? (lastItem as any).content : '';
-    const isTurnProcessing = lastDialogTurn?.status === 'processing' ||
-                              lastDialogTurn?.status === 'image_analyzing';
+    const isTurnProcessing =
+      lastDialogTurn?.status === 'processing' ||
+      lastDialogTurn?.status === 'finishing' ||
+      lastDialogTurn?.status === 'image_analyzing';
 
     return { lastItem, lastDialogTurn, content, isTurnProcessing };
   }, [activeSession]);

src/web-ui/src/flow_chat/components/toolbar-mode/ToolbarMode.tsx

@@ -118,7 +118,10 @@ export const ToolbarMode: React.FC = () => {
 
     const lastTurn = activeSession.dialogTurns[activeSession.dialogTurns.length - 1];
 
-    const isStreaming = lastTurn.status === 'processing' || lastTurn.status === 'image_analyzing';
+    const isStreaming =
+      lastTurn.status === 'processing' ||
+      lastTurn.status === 'finishing' ||
+      lastTurn.status === 'image_analyzing';
 
     if (!isStreaming || !lastTurn.modelRounds || lastTurn.modelRounds.length === 0) {
       return { isStreaming, toolName: null, content: null };

src/web-ui/src/flow_chat/services/FlowChatManager.ts

@@ -54,6 +54,7 @@ export class FlowChatManager {
       eventBatcher: new EventBatcher({
         onFlush: (events) => this.processBatchedEvents(events)
       }),
+      pendingTurnCompletions: new Map(),
       contentBuffers: new Map(),
       activeTextItems: new Map(),
       saveDebouncers: new Map(),
@@ -351,7 +352,11 @@ export class FlowChatManager {
     const session = this.context.flowChatStore.getState().sessions.get(parentSessionId);
     const turn = session?.dialogTurns.find(t => t.id === dialogTurnId);
     if (!turn) return;
-    if (turn.status !== 'processing' && turn.status !== 'image_analyzing') {
+    if (
+      turn.status !== 'processing' &&
+      turn.status !== 'finishing' &&
+      turn.status !== 'image_analyzing'
+    ) {
       // Only inject into an actively streaming turn; otherwise we'd create dangling streaming items.
       return;
     }