Skip to content

feat(agui): support tool execution progress and fix stream delta loss #1240

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jujn wants to merge 3 commits into from
Closed

feat(agui): support tool execution progress and fix stream delta loss #1240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8cbfce8
feat(agui): support tool execution progress and fix stream delta loss
jujn Apr 17, 2026
7e5c70b
fix: copilot review
jujn Apr 18, 2026
499321f
fix: conflict
jujn Apr 29, 2026
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
31 changes: 31 additions & 0 deletions ...cope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAdapterConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ public class AguiAdapterConfig {
private final boolean enableReasoning;
private final Duration runTimeout;
private final String defaultAgentId;
private final boolean enableToolProgressStream;

private AguiAdapterConfig(Builder builder) {
this.toolMergeMode = builder.toolMergeMode;
Expand All @@ -40,6 +41,7 @@ private AguiAdapterConfig(Builder builder) {
this.enableReasoning = builder.enableReasoning;
this.runTimeout = builder.runTimeout;
this.defaultAgentId = builder.defaultAgentId;
this.enableToolProgressStream = builder.enableToolProgressStream;
}

/**
Expand Down Expand Up @@ -100,6 +102,18 @@ public String getDefaultAgentId() {
return defaultAgentId;
}

/**
* Check if tool execution progress should be streamed.
*
* <p>When enabled, intermediate tool execution events will be converted into
* reasoning messages (Thinking UI) to provide real-time feedback to the frontend.
*
* @return true if tool progress streaming is enabled
*/
public boolean isEnableToolProgressStream() {
return enableToolProgressStream;
}

/**
* Creates a new builder for AguiAdapterConfig.
*
Expand Down Expand Up @@ -129,6 +143,7 @@ public static class Builder {
private boolean enableReasoning = false;
private Duration runTimeout = Duration.ofMinutes(10);
private String defaultAgentId;
private boolean enableToolProgressStream = false;

/**
* Set the tool merge mode.
Expand Down Expand Up @@ -200,6 +215,22 @@ public Builder defaultAgentId(String defaultAgentId) {
return this;
}

/**
* Set whether to enable tool execution progress stream.
*
* <p>When enabled, intermediate progress chunks generated during tool execution
* (e.g., download progress, code execution stdout) will be converted into
* {@code ReasoningMessage} events. This leverages the frontend's "Thinking" UI
* to display real-time feedback without modifying the AG-UI protocol.
*
* @param enableToolProgressStream true to convert progress events to thinking blocks
* @return This builder
*/
public Builder enableToolProgressStream(boolean enableToolProgressStream) {
this.enableToolProgressStream = enableToolProgressStream;
return this;
}

/**
* Build the configuration.
*
Expand Down
164 changes: 99 additions & 65 deletions ...scope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAgentAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ public Flux<AguiEvent> run(RunAgentInput input) {
StreamOptions.builder()
.eventTypes(EventType.ALL)
.incremental(true)
.includeActingChunk(config.isEnableToolProgressStream())
.build();

// Track state for event conversion
Expand Down Expand Up @@ -147,44 +148,30 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
EventType type = event.getType();

if (type == EventType.REASONING || type == EventType.SUMMARY) {
String messageId = msg.getId();

// Handle reasoning/summary events - convert to text messages and tool calls
for (ContentBlock block : msg.getContent()) {
if (block instanceof TextBlock textBlock) {
String text = textBlock.getText();
if (text != null && !text.isEmpty()) {
String messageId = msg.getId();

// Start message if not started
if (!state.hasStartedMessage(messageId)) {
if (!state.hasStartedTextMessage(messageId)) {
events.add(
new AguiEvent.TextMessageStart(
state.threadId, state.runId, messageId, "assistant"));
state.startMessage(messageId);
}

if (!event.isLast()) {
// In incremental mode, text is already the delta
events.add(
new AguiEvent.TextMessageContent(
state.threadId, state.runId, messageId, text));
} else {
// End message if this is the last event
if (!state.hasEndedMessage(messageId)) {
events.add(
new AguiEvent.TextMessageEnd(
state.threadId, state.runId, messageId));
state.endMessage(messageId);
}
state.startTextMessage(messageId);
}
events.add(
new AguiEvent.TextMessageContent(
state.threadId, state.runId, messageId, text));
Comment on lines 158 to +167

Copilot AI Apr 17, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Text deltas are now always emitted even when event.isLast() is true. If a TextMessageEnd was already emitted earlier for the same messageId (e.g., because a ToolUseBlock ended the active text message), this can result in TextMessageContent being sent after TextMessageEnd with no subsequent restart/end, which is an invalid lifecycle for many consumers. Consider suppressing content emission for ended messages, or introducing a continuation strategy (e.g., generate a new messageId for post-tool text) so that content is never emitted after an end event.

Copilot uses AI. Check for mistakes.

@jujn jujn Apr 29, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps we can use the idea of ag-ui-4j to introduce an event stream and subscriber distribution pattern, abstracting events into explicit lifecycle callbacks. Thus eliminating the fragile isLast() guess. I will study it later and create a new PR to refactor it.

}
} else if (block instanceof ThinkingBlock thinkingBlock) {
// Handle thinking blocks - convert to REASONING_* events (only if enabled)
// According to AG-UI Reasoning draft: https://docs.ag-ui.com/drafts/reasoning
if (config.isEnableReasoning()) {
String thinking = thinkingBlock.getThinking();
if (thinking != null && !thinking.isEmpty()) {
String messageId = msg.getId();

// Start reasoning message if not started
if (!state.hasStartedReasoningMessage(messageId)) {
events.add(
Expand All @@ -195,19 +182,9 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
"reasoning"));
state.startReasoningMessage(messageId);
}

if (!event.isLast()) {
// In incremental mode, thinking is already the delta
events.add(
new AguiEvent.ReasoningMessageContent(
state.threadId, state.runId, messageId, thinking));
} else {
// End reasoning message if this is the last event
events.add(
new AguiEvent.ReasoningMessageEnd(
state.threadId, state.runId, messageId));
state.endReasoningMessage(messageId);
}
events.add(
new AguiEvent.ReasoningMessageContent(
state.threadId, state.runId, messageId, thinking));
}
}
// If reasoning is disabled, ThinkingBlock content is ignored (backward
Expand All @@ -219,7 +196,7 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
events.add(
new AguiEvent.TextMessageEnd(
state.threadId, state.runId, activeMessageId));
state.endMessage(activeMessageId);
state.endTextMessage(activeMessageId);
}

// End any active reasoning message before starting tool call
Expand Down Expand Up @@ -248,7 +225,7 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
}

// Emit tool call args if enabled
if (config.isEmitToolCallArgs() && !event.isLast()) {
if (config.isEmitToolCallArgs()) {
String args = toolUse.getContent();
if (args != null && !args.isEmpty()) {
events.add(
Expand All @@ -258,7 +235,26 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
}
}
}
} else if (type == EventType.TOOL_RESULT && event.isLast()) {

// Independent lifecycle check: prevents dropping the final data chunk
// and safely handles empty termination packets.
if (event.isLast()) {
if (state.hasStartedTextMessage(messageId)
&& !state.hasEndedTextMessage(messageId)) {
events.add(
new AguiEvent.TextMessageEnd(state.threadId, state.runId, messageId));
state.endTextMessage(messageId);
}
if (config.isEnableReasoning()
&& state.hasStartedReasoningMessage(messageId)
&& !state.hasEndedReasoningMessage(messageId)) {
events.add(
new AguiEvent.ReasoningMessageEnd(
state.threadId, state.runId, messageId));
state.endReasoningMessage(messageId);
}
}
} else if (type == EventType.TOOL_RESULT) {
// Handle tool results
for (ContentBlock block : msg.getContent()) {
if (block instanceof ToolResultBlock toolResult) {
Expand All @@ -267,7 +263,7 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
toolCallId = UUID.randomUUID().toString();
}

String result = extractToolResultText(toolResult);
String resultText = extractToolResultText(toolResult);

boolean hasStarted = state.hasStartedToolCall(toolCallId);
if (!hasStarted) {
Expand All @@ -281,18 +277,56 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
state.startToolCall(toolCallId);
}
Comment thread
jujn marked this conversation as resolved.

// Ensure ToolCallEnd is emitted to close arguments phase
events.add(new AguiEvent.ToolCallEnd(state.threadId, state.runId, toolCallId));
if (!event.isLast() && resultText != null && !resultText.isEmpty()) {
// Handle the progress of tool execution (only if enabled)
// (borrowing from the Reasoning block)
if (config.isEnableToolProgressStream()) {
String progressMsgId = "tool-progress-" + toolCallId;

events.add(
new AguiEvent.ToolCallResult(
state.threadId,
state.runId,
toolCallId,
result,
"tool",
msg.getId()));
state.endToolCall(toolCallId);
if (!state.hasStartedReasoningMessage(progressMsgId)) {
events.add(
new AguiEvent.ReasoningMessageStart(
state.threadId,
state.runId,
progressMsgId,
"assistant"));
state.startReasoningMessage(progressMsgId);
}

events.add(
new AguiEvent.ReasoningMessageContent(
state.threadId,
state.runId,
progressMsgId,
resultText));
}
}

// Decouple termination logic from content extraction
// to ensure the final progress chunk is emitted before closing.
if (event.isLast()) {
String thinkingMsgId = "tool-progress-" + toolCallId;
if (state.hasStartedReasoningMessage(thinkingMsgId)
&& !state.hasEndedReasoningMessage(thinkingMsgId)) {
events.add(
new AguiEvent.ReasoningMessageEnd(
state.threadId, state.runId, thinkingMsgId));
state.endReasoningMessage(thinkingMsgId);
}

// Send real tool call end and result events
events.add(
new AguiEvent.ToolCallEnd(state.threadId, state.runId, toolCallId));
events.add(
new AguiEvent.ToolCallResult(
state.threadId,
state.runId,
toolCallId,
resultText,
"tool",
msg.getId()));
state.endToolCall(toolCallId);
}
}
}
}
Expand All @@ -309,9 +343,9 @@ private List<AguiEvent> convertEvent(Event event, EventConversionState state) {
private Flux<AguiEvent> finishRun(EventConversionState state) {
List<AguiEvent> events = new ArrayList<>();

// End any messages that weren't properly ended
for (String messageId : state.getStartedMessages()) {
if (!state.hasEndedMessage(messageId)) {
// End any text messages that weren't properly ended
for (String messageId : state.getStartedTextMessages()) {
if (!state.hasEndedTextMessage(messageId)) {
events.add(new AguiEvent.TextMessageEnd(state.threadId, state.runId, messageId));
}
}
Expand Down Expand Up @@ -385,8 +419,8 @@ private String serializeToolArgs(Map<String, Object> input) {
private static class EventConversionState {
final String threadId;
final String runId;
private final Set<String> startedMessages = new LinkedHashSet<>();
private final Set<String> endedMessages = new LinkedHashSet<>();
private final Set<String> startedTextMessages = new LinkedHashSet<>();
private final Set<String> endedTextMessages = new LinkedHashSet<>();
private final Set<String> startedToolCalls = new LinkedHashSet<>();
private final Set<String> endedToolCalls = new LinkedHashSet<>();
private final Set<String> startedReasoningMessages = new LinkedHashSet<>();
Expand All @@ -399,36 +433,36 @@ private static class EventConversionState {
this.runId = runId;
}

boolean hasStartedMessage(String messageId) {
return startedMessages.contains(messageId);
boolean hasStartedTextMessage(String messageId) {
return startedTextMessages.contains(messageId);
}

void startMessage(String messageId) {
startedMessages.add(messageId);
void startTextMessage(String messageId) {
startedTextMessages.add(messageId);
currentTextMessageId = messageId;
}

void endMessage(String messageId) {
endedMessages.add(messageId);
void endTextMessage(String messageId) {
endedTextMessages.add(messageId);
if (Objects.equals(messageId, currentTextMessageId)) {
currentTextMessageId = null;
}
}

boolean hasEndedMessage(String messageId) {
return endedMessages.contains(messageId);
boolean hasEndedTextMessage(String messageId) {
return endedTextMessages.contains(messageId);
}

String getCurrentTextMessageId() {
return currentTextMessageId;
}

boolean hasActiveTextMessage() {
return currentTextMessageId != null && !hasEndedMessage(currentTextMessageId);
return currentTextMessageId != null && !hasEndedTextMessage(currentTextMessageId);
}

Set<String> getStartedMessages() {
return startedMessages;
Set<String> getStartedTextMessages() {
return startedTextMessages;
}

boolean hasStartedToolCall(String toolCallId) {
Expand Down
Loading
Loading