diff --git a/executor/src/runtime_work/events.rs b/executor/src/runtime_work/events.rs index 441c7a1e1e7..8733f3417f7 100644 --- a/executor/src/runtime_work/events.rs +++ b/executor/src/runtime_work/events.rs @@ -270,13 +270,16 @@ impl CodexNotificationEventMapper { self.remember_subagent_item(notification.params); return; } - emit_tool_start( + let emitted_tool = emit_tool_start( event_tx, device_id, local_task_id, request, notification.params, ); + if !emitted_tool { + self.finish_process_text_before_item(&emit_context, notification.params); + } } "item/tool/requestUserInput" => { emit_request_user_input( @@ -1073,6 +1076,35 @@ impl CodexNotificationEventMapper { self.process_text = None; } + fn finish_process_text_before_item( + &mut self, + emit_context: &EventEmitContext<'_>, + params: &Value, + ) { + let next_item_id = notification_item_id(params); + let Some(process_text) = self.process_text.as_ref() else { + return; + }; + if process_text.item_id == next_item_id { + return; + } + + emit_response_event( + emit_context.event_tx, + emit_context.device_id, + "response.block.updated", + emit_context.local_task_id, + emit_context.request, + json!({ + "block_id": process_text.id, + "updates": { + "status": "done", + } + }), + ); + self.reset_process_text(); + } + fn remember_subagent_item(&mut self, params: &Value) { if let Some(item_id) = notification_item_id(params) { self.subagent_item_ids.insert(item_id); @@ -1308,7 +1340,7 @@ fn emit_tool_start( local_task_id: &str, request: &ExecutionRequest, params: &Value, -) { +) -> bool { if let Some(block) = workbench_block_from_notification( params, &request.subtask_id, @@ -1324,7 +1356,9 @@ fn emit_tool_start( request, json!({"block": block}), ); + return true; } + false } fn emit_request_user_input( @@ -2461,6 +2495,72 @@ mod tests { assert!(event_rx.try_recv().is_err()); } + #[test] + fn completes_streamed_process_text_when_the_next_codex_item_starts() { + let (event_tx, mut event_rx) = broadcast::channel(4); + let request = ExecutionRequest { + task_id: "7".to_owned(), + subtask_id: "8".to_owned(), + ..ExecutionRequest::default() + }; + let mut mapper = CodexNotificationEventMapper::default(); + + for message in [ + json!({ + "method": "item/started", + "params": { + "item": { + "id": "msg-commentary", + "type": "agentMessage", + "phase": "commentary", + "text": "" + } + } + }), + json!({ + "method": "item/agentMessage/delta", + "params": { + "itemId": "msg-commentary", + "delta": "I will inspect." + } + }), + json!({ + "method": "item/started", + "params": { + "item": { + "id": "reasoning-1", + "type": "reasoning" + } + } + }), + ] { + mapper.map( + &Some(event_tx.clone()), + "device-1", + "local-1", + &request, + message, + ); + } + + let created = event_rx + .try_recv() + .expect("streamed process text should be emitted"); + let completed = event_rx + .try_recv() + .expect("the next item should settle the process text"); + + assert_eq!(created["event"], "response.block.created"); + assert_eq!(created["payload"]["data"]["block"]["status"], "streaming"); + assert_eq!(completed["event"], "response.block.updated"); + assert_eq!( + completed["payload"]["data"]["block_id"], + created["payload"]["data"]["block"]["id"] + ); + assert_eq!(completed["payload"]["data"]["updates"]["status"], "done"); + assert!(event_rx.try_recv().is_err()); + } + #[test] fn keeps_streamed_final_text_as_one_process_block_when_completed_as_commentary() { let (event_tx, mut event_rx) = broadcast::channel(4); @@ -3894,6 +3994,9 @@ mod tests { .try_recv() .expect("commentary process event should be emitted"); let tool = event_rx.try_recv().expect("tool event should be emitted"); + let process_completed = event_rx + .try_recv() + .expect("the final item should settle commentary"); let final_text = event_rx .try_recv() .expect("final process event should be emitted"); @@ -3909,6 +4012,15 @@ mod tests { tool["payload"]["data"]["block"]["tool_name"], "exec_command" ); + assert_eq!(process_completed["event"], "response.block.updated"); + assert_eq!( + process_completed["payload"]["data"]["block_id"], + process["payload"]["data"]["block"]["id"] + ); + assert_eq!( + process_completed["payload"]["data"]["updates"]["status"], + "done" + ); assert_eq!(final_text["event"], "response.block.created"); assert_eq!( final_text["payload"]["data"]["block"]["content"], diff --git a/wework/src/components/chat/MessageList.test.tsx b/wework/src/components/chat/MessageList.test.tsx index c7550c073e7..9f784057690 100644 --- a/wework/src/components/chat/MessageList.test.tsx +++ b/wework/src/components/chat/MessageList.test.tsx @@ -4665,6 +4665,35 @@ describe('MessageList', () => { expect(screen.queryByTestId('thinking-indicator')).not.toBeInTheDocument() }) + test('shows thinking after process text completes while waiting for the next item', () => { + const processBlock: ProcessingBlock = { + id: 'text-1', + subtaskId: 1, + type: 'text', + content: 'Let me inspect the repository first.', + status: 'done', + createdAt: 1770000000000, + } + + render( + + ) + + expect(screen.getByText('Let me inspect the repository first.')).toBeInTheDocument() + expect(screen.getByTestId('thinking-indicator')).toHaveTextContent('正在思考') + }) + test('collapses tool rows and shows trailing thinking once final text is visible', () => { const runningBlock: ProcessingBlock = { id: 'call-1', diff --git a/wework/src/components/chat/MessageList.tsx b/wework/src/components/chat/MessageList.tsx index 8cc64703cfa..68f87fdc6c5 100644 --- a/wework/src/components/chat/MessageList.tsx +++ b/wework/src/components/chat/MessageList.tsx @@ -2066,6 +2066,7 @@ function AssistantMessage({ isStreaming, hasProcessingDisplayBlock: hasProcessingDisplayBlock(displayBlocks), hasVisibleContent, + hasTrailingCompletedProcessText: hasTrailingCompletedProcessText(displayBlocks), }) const webSearchSources = isStreaming ? [] @@ -2416,16 +2417,26 @@ function hasProcessingDisplayBlock(blocks: ProcessingBlock[]): boolean { return buildProcessingDisplayRows(blocks).length > 0 } +function hasTrailingCompletedProcessText(blocks: ProcessingBlock[]): boolean { + const lastBlock = blocks.at(-1) + return lastBlock?.type === 'text' && (lastBlock.status === 'done' || lastBlock.status === 'error') +} + function shouldShowAssistantThinkingIndicator({ isStreaming, hasProcessingDisplayBlock, hasVisibleContent, + hasTrailingCompletedProcessText, }: { isStreaming: boolean hasProcessingDisplayBlock: boolean hasVisibleContent: boolean + hasTrailingCompletedProcessText: boolean }): boolean { - return isStreaming && (!hasProcessingDisplayBlock || hasVisibleContent) + return ( + isStreaming && + (!hasProcessingDisplayBlock || hasVisibleContent || hasTrailingCompletedProcessText) + ) } function AssistantErrorCard({