Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
116 changes: 114 additions & 2 deletions executor/src/runtime_work/events.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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,
Expand All @@ -1324,7 +1356,9 @@ fn emit_tool_start(
request,
json!({"block": block}),
);
return true;
}
false
}

fn emit_request_user_input(
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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");
Expand All @@ -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"],
Expand Down
29 changes: 29 additions & 0 deletions wework/src/components/chat/MessageList.test.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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(
<MessageList
messages={[
{
id: '2',
role: 'assistant',
content: '',
status: 'streaming',
createdAt: '2026-05-25T18:46:00.000+08:00',
blocks: [processBlock],
},
]}
/>
)

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',
Expand Down
13 changes: 12 additions & 1 deletion wework/src/components/chat/MessageList.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -2066,6 +2066,7 @@ function AssistantMessage({
isStreaming,
hasProcessingDisplayBlock: hasProcessingDisplayBlock(displayBlocks),
hasVisibleContent,
hasTrailingCompletedProcessText: hasTrailingCompletedProcessText(displayBlocks),
})
const webSearchSources = isStreaming
? []
Expand Down Expand Up @@ -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({
Expand Down
Loading