From a30384b9e75064528750b16662558d499ddc6602 Mon Sep 17 00:00:00 2001 From: wzq-xzwj <34700058+wzq-xzwj@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:15:56 +0800 Subject: [PATCH] fix(agui): migrate adapter to v2 streamEvents and bridge HITL confirmations (#2437) The AG-UI adapter consumed the deprecated v1 `Flux` stream, which only surfaces coarse REASONING/SUMMARY/TOOL_RESULT events and cannot carry human-in-the-loop signals. As a result, tool-confirmation events (RequireUserConfirmEvent) were silently dropped: over AG-UI, HITL was completely non-functional -- a paused run either hung or ran to a bare completion, and the client was never told a confirmation was required, nor could it resume an approved call. This change: * Adds `EventStreamingAgent`, a small capability interface in `io.agentscope.core.agent` declaring `Flux streamEvents(List, RuntimeContext)`. Both `ReActAgent` and `HarnessAgent` implement it, so the adapter can consume the fine-grained v2 stream uniformly via `instanceof EventStreamingAgent`, independent of the concrete agent type, and still fall back to the deprecated v1 path for other `Agent` implementations. * Migrates `AguiAgentAdapter` to consume the v2 `Flux` stream and map each typed event (text/thinking/tool-call/tool-result blocks) to the corresponding AG-UI events. * Bridges HITL in both directions: - Outbound: when a run pauses on RequireUserConfirmEvent, RUN_FINISHED now carries a RunFinishedInterruptOutcome with one Interrupt per pending tool call (reason "tool_confirmation"), so the client learns a confirmation is required instead of seeing a bare completion. - Inbound: confirmation results supplied via `forwardedProps["agentscope_confirm_results"]` are parsed into `List` and attached to the resumed message under `Msg.METADATA_CONFIRM_RESULTS`, letting a paused ReActAgent apply them and continue. The reconstructed ToolUseBlock also carries a raw-args JSON string in `content`, because the tool executor validates `ToolUseBlock.getContent()` (not the parsed input map) and applying a ConfirmResult fully replaces the stored block. Adds AguiAgentAdapterV2Test (11 tests) covering the v2 mapping path and both HITL directions. Verified end-to-end against a local Harness + MCP setup: an echo tool call pauses with an interrupt outcome and is not executed, and a follow-up resume with confirmation results executes it and returns the real result. Closes #2437 --- .../java/io/agentscope/core/ReActAgent.java | 3 +- .../core/agent/EventStreamingAgent.java | 47 ++ .../core/agui/adapter/AguiAgentAdapter.java | 495 +++++++++++++++++- .../agui/adapter/AguiAgentAdapterV2Test.java | 400 ++++++++++++++ .../harness/agent/HarnessAgent.java | 3 +- 5 files changed, 928 insertions(+), 20 deletions(-) create mode 100644 agentscope-core/src/main/java/io/agentscope/core/agent/EventStreamingAgent.java create mode 100644 agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAgentAdapterV2Test.java diff --git a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java index 003bbfeb15..c266d69696 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java @@ -19,6 +19,7 @@ import io.agentscope.core.agent.Agent; import io.agentscope.core.agent.AgentBase; import io.agentscope.core.agent.Event; +import io.agentscope.core.agent.EventStreamingAgent; import io.agentscope.core.agent.RuntimeContext; import io.agentscope.core.agent.StreamOptions; import io.agentscope.core.agent.SubagentEventBus; @@ -203,7 +204,7 @@ * {@link io.agentscope.core.state.AgentStateStore} are all safe to share across instances. */ @SuppressWarnings("deprecation") -public class ReActAgent extends AgentBase implements AutoCloseable { +public class ReActAgent extends AgentBase implements EventStreamingAgent, AutoCloseable { private static final Logger log = LoggerFactory.getLogger(ReActAgent.class); private static final GracefulShutdownManager shutdownManager = diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/EventStreamingAgent.java b/agentscope-core/src/main/java/io/agentscope/core/agent/EventStreamingAgent.java new file mode 100644 index 0000000000..173fae8658 --- /dev/null +++ b/agentscope-core/src/main/java/io/agentscope/core/agent/EventStreamingAgent.java @@ -0,0 +1,47 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.agent; + +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.message.Msg; +import java.util.List; +import reactor.core.publisher.Flux; + +/** + * Capability interface for agents that can emit the fine-grained 2.0 {@link AgentEvent} stream. + * + *

This is the modern replacement for the deprecated v1 {@link StreamableAgent} {@code + * stream(...)} API, which only surfaced coarse {@code REASONING} / {@code SUMMARY} / {@code + * TOOL_RESULT} events and, critically, could not carry human-in-the-loop signals such as {@code + * RequireUserConfirmEvent} / {@code RequestStopEvent}. + * + *

Both {@code io.agentscope.core.ReActAgent} and {@code io.agentscope.harness.agent.HarnessAgent} + * (which wraps a {@code ReActAgent} delegate) implement this interface. Consumers such as the AG-UI + * adapter can branch on {@code agent instanceof EventStreamingAgent} to consume the v2 stream + * uniformly, regardless of the concrete agent type, while still falling back to the deprecated v1 + * path for custom {@link Agent} implementations that do not support event streaming. + */ +public interface EventStreamingAgent { + + /** + * Stream fine-grained {@link AgentEvent}s covering the full agent invocation lifecycle. + * + * @param msgs input messages + * @param context runtime context to propagate into the call + * @return event stream covering the full agent invocation lifecycle + */ + Flux streamEvents(List msgs, RuntimeContext context); +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAgentAdapter.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAgentAdapter.java index ba3aa2b2cc..dba6b1cbf5 100644 --- a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAgentAdapter.java +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAgentAdapter.java @@ -17,6 +17,7 @@ import io.agentscope.core.agent.Agent; import io.agentscope.core.agent.Event; +import io.agentscope.core.agent.EventStreamingAgent; import io.agentscope.core.agent.EventType; import io.agentscope.core.agent.RuntimeContext; import io.agentscope.core.agent.StreamOptions; @@ -25,8 +26,24 @@ import io.agentscope.core.agui.event.AguiEvent; import io.agentscope.core.agui.model.RunAgentInput; import io.agentscope.core.agui.model.ToolMergeMode; +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.event.ConfirmResult; +import io.agentscope.core.event.RequireUserConfirmEvent; +import io.agentscope.core.event.TextBlockDeltaEvent; +import io.agentscope.core.event.TextBlockEndEvent; +import io.agentscope.core.event.TextBlockStartEvent; +import io.agentscope.core.event.ThinkingBlockDeltaEvent; +import io.agentscope.core.event.ThinkingBlockEndEvent; +import io.agentscope.core.event.ThinkingBlockStartEvent; +import io.agentscope.core.event.ToolCallDeltaEvent; +import io.agentscope.core.event.ToolCallEndEvent; +import io.agentscope.core.event.ToolCallStartEvent; +import io.agentscope.core.event.ToolResultEndEvent; +import io.agentscope.core.event.ToolResultStartEvent; +import io.agentscope.core.event.ToolResultTextDeltaEvent; import io.agentscope.core.message.ContentBlock; import io.agentscope.core.message.Msg; +import io.agentscope.core.message.MsgRole; import io.agentscope.core.message.TextBlock; import io.agentscope.core.message.ThinkingBlock; import io.agentscope.core.message.ToolResultBlock; @@ -39,6 +56,7 @@ import io.agentscope.core.util.JsonUtils; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -80,6 +98,17 @@ public class AguiAgentAdapter { public static final String RUNTIME_CONTEXT_STATE_KEY = "agui.state"; public static final String RUNTIME_CONTEXT_FORWARDED_PROPS_KEY = "agui.forwardedProps"; + /** + * Key under {@link RunAgentInput#getForwardedProps()} where the client sends back + * human-in-the-loop confirmation results to resume a paused run. The value is a list of maps, + * each shaped like {@code {"toolCallId": "...", "confirmed": true, "toolName": "...", + * "input": {...}}}. + */ + public static final String FORWARDED_PROPS_CONFIRM_RESULTS_KEY = "agentscope_confirm_results"; + + /** Reason string used on {@link AguiEvent.Interrupt}s emitted for HITL tool confirmation. */ + static final String CONFIRM_INTERRUPT_REASON = "tool_confirmation"; + private final Agent agent; private final AguiAdapterConfig config; private final AguiMessageConverter messageConverter; @@ -116,25 +145,18 @@ public Flux run(RunAgentInput input) { // Convert AG-UI messages to AgentScope messages List msgs = messageConverter.toMsgList(input.getMessages()); - // Create stream options - use incremental mode for true streaming - StreamOptions options = - StreamOptions.builder() - .eventTypes(EventType.ALL) - .incremental(true) - .build(); + // HITL resume: if the client sent back confirmation results via forwardedProps, + // attach them to the latest message so ReActAgent can apply them and continue. + msgs = attachConfirmResults(msgs, input); // Track state for event conversion EventConversionState state = new EventConversionState(threadId, runId); RuntimeContext runtimeContext = buildRuntimeContext(input); ToolInjection toolInjection = ToolInjection.empty(); - Flux agentEvents; + Flux convertedStream; try { toolInjection = injectFrontendTools(input); - agentEvents = agent.stream(msgs, options, runtimeContext); - if (agentEvents == null) { - agentEvents = agent.stream(msgs, options); - } - agentEvents = Objects.requireNonNull(agentEvents, "agent stream is null"); + convertedStream = buildConvertedStream(msgs, runtimeContext, state); } catch (Throwable error) { toolInjection.close(); return Flux.concat( @@ -148,10 +170,8 @@ public Flux run(RunAgentInput input) { // Emit RUN_STARTED Flux.just( new AguiEvent.RunStarted(threadId, runId, null, input)), - // Stream agent events and convert to AG-UI events - // Use concatMapIterable to preserve strict event ordering - agentEvents.concatMapIterable( - event -> convertEvent(event, state)), + // Stream converted AG-UI events + convertedStream, // Emit any pending end events and RUN_FINISHED Flux.defer(() -> finishRun(state))) .doFinally(signalType -> activeToolInjection.close()) @@ -159,6 +179,167 @@ public Flux run(RunAgentInput input) { }); } + /** + * Build the stream of AG-UI events converted from the agent's event stream. + * + *

When the underlying agent is an {@link EventStreamingAgent} (such as {@code ReActAgent} or + * {@code HarnessAgent}), this consumes the fine-grained v2 {@link AgentEvent} stream via {@link + * EventStreamingAgent#streamEvents(List, RuntimeContext)}. For any other {@link Agent} + * implementation it falls back to the deprecated v1 {@link Event} stream so that custom agents + * (and existing integrations) keep working unchanged. + * + * @param msgs the converted input messages + * @param runtimeContext the per-run runtime context + * @param state the conversion state tracker + * @return a Flux of AG-UI events (without RUN_STARTED / RUN_FINISHED bookends) + */ + private Flux buildConvertedStream( + List msgs, RuntimeContext runtimeContext, EventConversionState state) { + if (agent instanceof EventStreamingAgent streamingAgent) { + Flux agentEvents = streamingAgent.streamEvents(msgs, runtimeContext); + agentEvents = Objects.requireNonNull(agentEvents, "agent stream is null"); + return agentEvents.concatMapIterable(event -> convertAgentEvent(event, state)); + } + + // Fallback: deprecated v1 Event stream for agents that do not support event streaming. + StreamOptions options = + StreamOptions.builder().eventTypes(EventType.ALL).incremental(true).build(); + Flux agentEvents = agent.stream(msgs, options, runtimeContext); + if (agentEvents == null) { + agentEvents = agent.stream(msgs, options); + } + agentEvents = Objects.requireNonNull(agentEvents, "agent stream is null"); + return agentEvents.concatMapIterable(event -> convertEvent(event, state)); + } + + /** + * Translate human-in-the-loop confirmation results carried in {@link + * RunAgentInput#getForwardedProps()} into a {@code List} attached to the last + * message under {@link Msg#METADATA_CONFIRM_RESULTS}, so a resumed {@link ReActAgent} can apply + * them to its ASKING tool calls and continue. + * + *

The expected {@code forwardedProps["agentscope_confirm_results"]} value is a {@code List} + * of maps, each shaped like {@code {"toolCallId": "...", "confirmed": true, "toolName": "...", + * "input": {...}}}. Entries missing a {@code toolCallId} are ignored. When no confirmation + * results are present the input messages are returned unchanged. + * + * @param msgs the converted input messages + * @param input the AG-UI run input + * @return the (possibly modified) message list to feed the agent + */ + private List attachConfirmResults(List msgs, RunAgentInput input) { + List confirmResults = parseConfirmResults(input.getForwardedProps()); + if (confirmResults.isEmpty()) { + return msgs; + } + + List result = new ArrayList<>(msgs); + Map metadata = new HashMap<>(); + metadata.put(Msg.METADATA_CONFIRM_RESULTS, confirmResults); + + if (result.isEmpty()) { + // No carrier message from the client; synthesise a minimal user message. + result.add( + Msg.builder() + .name("user") + .role(MsgRole.USER) + .textContent("[confirm]") + .metadata(metadata) + .build()); + return result; + } + + // Merge the confirm-result metadata onto the last message, preserving its existing fields. + int lastIdx = result.size() - 1; + Msg last = result.get(lastIdx); + Map merged = new HashMap<>(); + if (last.getMetadata() != null) { + merged.putAll(last.getMetadata()); + } + merged.put(Msg.METADATA_CONFIRM_RESULTS, confirmResults); + result.set( + lastIdx, + Msg.builder() + .id(last.getId()) + .name(last.getName()) + .role(last.getRole()) + .content(last.getContent()) + .metadata(merged) + .build()); + return result; + } + + /** + * Parse the raw {@code forwardedProps} confirmation payload into {@link ConfirmResult}s. + * + * @param forwardedProps the AG-UI forwardedProps map (may be null/empty) + * @return the parsed confirmation results, never null + */ + @SuppressWarnings("unchecked") + private List parseConfirmResults(Map forwardedProps) { + if (forwardedProps == null || forwardedProps.isEmpty()) { + return Collections.emptyList(); + } + Object raw = forwardedProps.get(FORWARDED_PROPS_CONFIRM_RESULTS_KEY); + if (!(raw instanceof List rawList) || rawList.isEmpty()) { + return Collections.emptyList(); + } + + List results = new ArrayList<>(); + for (Object element : rawList) { + if (!(element instanceof Map entry)) { + continue; + } + Object toolCallId = entry.get("toolCallId"); + if (toolCallId == null) { + toolCallId = entry.get("toolUseId"); + } + if (toolCallId == null) { + continue; + } + boolean confirmed = toBoolean(entry.get("confirmed"), true); + Object toolName = entry.get("toolName"); + Object inputObj = entry.get("input"); + Map toolInput = + inputObj instanceof Map m ? (Map) m : Map.of(); + + // The tool executor validates a tool call against its raw-args JSON string + // (ToolUseBlock.getContent()), not its parsed input map, and applying a + // ConfirmResult fully replaces the stored ToolUseBlock. So we must also carry the + // args as a JSON string, otherwise the resumed tool call fails schema validation + // with a null "content". Prefer an explicit client-provided string, else serialize + // the input map. + Object rawContent = entry.get("content"); + if (rawContent == null) { + rawContent = entry.get("argsJson"); + } + String toolContent = + rawContent instanceof String s && !s.isBlank() + ? s + : serializeToolArgs(toolInput); + + ToolUseBlock toolCall = + ToolUseBlock.builder() + .id(String.valueOf(toolCallId)) + .name(toolName != null ? String.valueOf(toolName) : "") + .input(toolInput) + .content(toolContent) + .build(); + results.add(new ConfirmResult(confirmed, toolCall, null)); + } + return results; + } + + private static boolean toBoolean(Object value, boolean defaultValue) { + if (value instanceof Boolean b) { + return b; + } + if (value instanceof String s) { + return Boolean.parseBoolean(s); + } + return defaultValue; + } + private RuntimeContext buildRuntimeContext(RunAgentInput input) { return RuntimeContext.builder() .sessionId(input.getThreadId()) @@ -391,9 +572,213 @@ private List convertEvent(Event event, EventConversionState state) { return events; } + /** + * Convert a fine-grained v2 {@link AgentEvent} to AG-UI events. + * + *

Maps the granular streaming events emitted by {@link ReActAgent#streamEvents} onto the + * AG-UI protocol: + *

    + *
  • {@link TextBlockStartEvent}/{@link TextBlockDeltaEvent}/{@link TextBlockEndEvent} + * → TEXT_MESSAGE_START / TEXT_MESSAGE_CONTENT / TEXT_MESSAGE_END
    • + *
  • {@link ThinkingBlockStartEvent}/{@link ThinkingBlockDeltaEvent}/{@link + * ThinkingBlockEndEvent} → REASONING_MESSAGE_* (only when reasoning is enabled)
    • + *
  • {@link ToolCallStartEvent}/{@link ToolCallDeltaEvent}/{@link ToolCallEndEvent} + * → TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END
    • + *
  • {@link ToolResultTextDeltaEvent} accumulated and flushed at {@link ToolResultEndEvent} + * → TOOL_CALL_RESULT
    • + *
+ * + *

Agent lifecycle events (AgentStart/Result/End, ModelCall*) are intentionally ignored here: + * the adapter emits RUN_STARTED / RUN_FINISHED around this stream itself. + * + * @param event the v2 agent event + * @param state the conversion state + * @return list of AG-UI events + */ + private List convertAgentEvent(AgentEvent event, EventConversionState state) { + List events = new ArrayList<>(); + + if (event instanceof TextBlockStartEvent textStart) { + String messageId = textStart.getBlockId(); + if (!state.hasStartedMessage(messageId)) { + events.add( + new AguiEvent.TextMessageStart( + state.threadId, state.runId, messageId, "assistant")); + state.startMessage(messageId); + } + } else if (event instanceof TextBlockDeltaEvent textDelta) { + String messageId = textDelta.getBlockId(); + String delta = textDelta.getDelta(); + if (delta != null && !delta.isEmpty()) { + if (!state.hasStartedMessage(messageId)) { + events.add( + new AguiEvent.TextMessageStart( + state.threadId, state.runId, messageId, "assistant")); + state.startMessage(messageId); + } + events.add( + new AguiEvent.TextMessageContent( + state.threadId, state.runId, messageId, delta)); + } + } else if (event instanceof TextBlockEndEvent textEnd) { + String messageId = textEnd.getBlockId(); + if (state.hasStartedMessage(messageId) && !state.hasEndedMessage(messageId)) { + events.add(new AguiEvent.TextMessageEnd(state.threadId, state.runId, messageId)); + state.endMessage(messageId); + } + } else if (event instanceof ThinkingBlockStartEvent thinkingStart) { + if (config.isEnableReasoning()) { + String messageId = thinkingStart.getBlockId(); + if (!state.hasStartedReasoningMessage(messageId)) { + events.add( + new AguiEvent.ReasoningMessageStart( + state.threadId, state.runId, messageId, "reasoning")); + state.startReasoningMessage(messageId); + } + } + } else if (event instanceof ThinkingBlockDeltaEvent thinkingDelta) { + if (config.isEnableReasoning()) { + String messageId = thinkingDelta.getBlockId(); + String delta = thinkingDelta.getDelta(); + if (delta != null && !delta.isEmpty()) { + if (!state.hasStartedReasoningMessage(messageId)) { + events.add( + new AguiEvent.ReasoningMessageStart( + state.threadId, state.runId, messageId, "reasoning")); + state.startReasoningMessage(messageId); + } + events.add( + new AguiEvent.ReasoningMessageContent( + state.threadId, state.runId, messageId, delta)); + } + } + } else if (event instanceof ThinkingBlockEndEvent thinkingEnd) { + if (config.isEnableReasoning()) { + String messageId = thinkingEnd.getBlockId(); + if (state.hasStartedReasoningMessage(messageId) + && !state.hasEndedReasoningMessage(messageId)) { + events.add( + new AguiEvent.ReasoningMessageEnd( + state.threadId, state.runId, messageId)); + state.endReasoningMessage(messageId); + } + } + } else if (event instanceof ToolCallStartEvent toolStart) { + // Close any active text / reasoning message before starting a tool call. + if (state.hasActiveTextMessage()) { + String activeMessageId = state.getCurrentTextMessageId(); + events.add( + new AguiEvent.TextMessageEnd(state.threadId, state.runId, activeMessageId)); + state.endMessage(activeMessageId); + } + if (state.hasActiveReasoningMessage()) { + String activeReasoningMessageId = state.getCurrentReasoningMessageId(); + events.add( + new AguiEvent.ReasoningMessageEnd( + state.threadId, state.runId, activeReasoningMessageId)); + state.endReasoningMessage(activeReasoningMessageId); + } + + String toolCallId = toolStart.getToolCallId(); + if (toolCallId == null) { + toolCallId = UUID.randomUUID().toString(); + } + if (!state.hasStartedToolCall(toolCallId)) { + events.add( + new AguiEvent.ToolCallStart( + state.threadId, + state.runId, + toolCallId, + toolStart.getToolCallName())); + state.startToolCall(toolCallId); + } + } else if (event instanceof ToolCallDeltaEvent toolDelta) { + if (config.isEmitToolCallArgs()) { + String toolCallId = toolDelta.getToolCallId(); + String delta = toolDelta.getDelta(); + if (toolCallId != null && delta != null && !delta.isEmpty()) { + events.add( + new AguiEvent.ToolCallArgs( + state.threadId, state.runId, toolCallId, delta)); + } + } + } else if (event instanceof ToolCallEndEvent toolEnd) { + String toolCallId = toolEnd.getToolCallId(); + if (toolCallId != null && !state.hasEndedToolCall(toolCallId)) { + if (!state.hasStartedToolCall(toolCallId)) { + events.add( + new AguiEvent.ToolCallStart( + state.threadId, + state.runId, + toolCallId, + toolEnd.getToolCallName())); + state.startToolCall(toolCallId); + } + events.add(new AguiEvent.ToolCallEnd(state.threadId, state.runId, toolCallId)); + state.endToolCall(toolCallId); + } + } else if (event instanceof ToolResultStartEvent toolResultStart) { + state.beginToolResult(toolResultStart.getToolCallId()); + } else if (event instanceof ToolResultTextDeltaEvent toolResultDelta) { + state.appendToolResultText(toolResultDelta.getToolCallId(), toolResultDelta.getDelta()); + } else if (event instanceof ToolResultEndEvent toolResultEnd) { + String toolCallId = toolResultEnd.getToolCallId(); + if (toolCallId != null) { + if (!state.hasStartedToolCall(toolCallId)) { + String toolName = toolResultEnd.getToolCallName(); + if (toolName == null || toolName.isBlank()) { + toolName = "unknown"; + } + events.add( + new AguiEvent.ToolCallStart( + state.threadId, state.runId, toolCallId, toolName)); + state.startToolCall(toolCallId); + } + if (!state.hasEndedToolCall(toolCallId)) { + events.add(new AguiEvent.ToolCallEnd(state.threadId, state.runId, toolCallId)); + state.endToolCall(toolCallId); + } + String result = state.takeToolResultText(toolCallId); + events.add( + new AguiEvent.ToolCallResult( + state.threadId, + state.runId, + toolCallId, + result, + "tool", + UUID.randomUUID().toString())); + } + } else if (event instanceof RequireUserConfirmEvent confirm) { + // HITL: the agent paused and is asking the user to confirm these tool calls. Close any + // dangling text/reasoning message, then record the pending tool calls so finishRun() + // surfaces them as a RUN_FINISHED interrupt outcome. + if (state.hasActiveTextMessage()) { + String activeMessageId = state.getCurrentTextMessageId(); + events.add( + new AguiEvent.TextMessageEnd(state.threadId, state.runId, activeMessageId)); + state.endMessage(activeMessageId); + } + if (state.hasActiveReasoningMessage()) { + String activeReasoningMessageId = state.getCurrentReasoningMessageId(); + events.add( + new AguiEvent.ReasoningMessageEnd( + state.threadId, state.runId, activeReasoningMessageId)); + state.endReasoningMessage(activeReasoningMessageId); + } + state.markPausedForConfirmation(confirm.getToolCalls()); + } + + return events; + } + /** * Finish the run by emitting any pending end events and RUN_FINISHED. * + *

When the run paused for human-in-the-loop confirmation (a {@link + * io.agentscope.core.event.RequireUserConfirmEvent} was observed), the RUN_FINISHED event + * carries a {@link AguiEvent.RunFinishedInterruptOutcome} describing the pending tool calls the + * client must confirm, instead of a bare completion. + * * @param state The conversion state * @return Flux of final events */ @@ -422,8 +807,33 @@ private Flux finishRun(EventConversionState state) { } } - // Emit RUN_FINISHED - events.add(new AguiEvent.RunFinished(state.threadId, state.runId)); + // Emit RUN_FINISHED - with an interrupt outcome if the run paused for HITL confirmation. + if (state.isPausedForConfirmation()) { + List interrupts = new ArrayList<>(); + for (ToolUseBlock pending : state.getPendingConfirmations()) { + String toolCallId = + pending.getId() != null ? pending.getId() : UUID.randomUUID().toString(); + interrupts.add( + new AguiEvent.Interrupt( + toolCallId, + CONFIRM_INTERRUPT_REASON, + "Tool call '" + + pending.getName() + + "' requires confirmation before it can run.", + toolCallId, + null, + null, + null)); + } + events.add( + new AguiEvent.RunFinished( + state.threadId, + state.runId, + null, + new AguiEvent.RunFinishedInterruptOutcome(interrupts))); + } else { + events.add(new AguiEvent.RunFinished(state.threadId, state.runId)); + } return Flux.fromIterable(events); } @@ -536,6 +946,12 @@ private static class EventConversionState { private final Set endedReasoningMessages = new LinkedHashSet<>(); private String currentTextMessageId = null; private String currentReasoningMessageId = null; + // Accumulates streamed tool-result text (v2 ToolResultTextDeltaEvent) keyed by toolCallId, + // flushed into a single TOOL_CALL_RESULT at ToolResultEndEvent. + private final Map toolResultBuffers = new LinkedHashMap<>(); + // Pending HITL tool calls captured from RequireUserConfirmEvent; when non-null the run + // finishes with a RunFinishedInterruptOutcome instead of a bare RUN_FINISHED. + private List pendingConfirmations = null; EventConversionState(String threadId, String runId) { this.threadId = threadId; @@ -626,5 +1042,48 @@ boolean hasActiveReasoningMessage() { Set getStartedReasoningMessages() { return startedReasoningMessages; } + + // ===== Tool-result text buffering (v2) ===== + + void beginToolResult(String toolCallId) { + if (toolCallId != null) { + toolResultBuffers.computeIfAbsent(toolCallId, k -> new StringBuilder()); + } + } + + void appendToolResultText(String toolCallId, String delta) { + if (toolCallId == null || delta == null || delta.isEmpty()) { + return; + } + toolResultBuffers.computeIfAbsent(toolCallId, k -> new StringBuilder()).append(delta); + } + + /** + * Return the accumulated tool-result text for the given tool call and clear the buffer. + * + * @return the buffered text, or {@code null} if nothing was accumulated + */ + String takeToolResultText(String toolCallId) { + StringBuilder sb = toolResultBuffers.remove(toolCallId); + if (sb == null || sb.length() == 0) { + return null; + } + return sb.toString(); + } + + // ===== HITL confirmation tracking ===== + + void markPausedForConfirmation(List toolCalls) { + this.pendingConfirmations = + toolCalls != null ? List.copyOf(toolCalls) : Collections.emptyList(); + } + + boolean isPausedForConfirmation() { + return pendingConfirmations != null; + } + + List getPendingConfirmations() { + return pendingConfirmations != null ? pendingConfirmations : Collections.emptyList(); + } } } diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAgentAdapterV2Test.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAgentAdapterV2Test.java new file mode 100644 index 0000000000..a398d37edf --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAgentAdapterV2Test.java @@ -0,0 +1,400 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.agui.adapter; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.agentscope.core.ReActAgent; +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.model.AguiMessage; +import io.agentscope.core.agui.model.RunAgentInput; +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.event.ConfirmResult; +import io.agentscope.core.event.RequireUserConfirmEvent; +import io.agentscope.core.event.TextBlockDeltaEvent; +import io.agentscope.core.event.TextBlockEndEvent; +import io.agentscope.core.event.TextBlockStartEvent; +import io.agentscope.core.event.ThinkingBlockDeltaEvent; +import io.agentscope.core.event.ThinkingBlockEndEvent; +import io.agentscope.core.event.ThinkingBlockStartEvent; +import io.agentscope.core.event.ToolCallDeltaEvent; +import io.agentscope.core.event.ToolCallEndEvent; +import io.agentscope.core.event.ToolCallStartEvent; +import io.agentscope.core.event.ToolResultEndEvent; +import io.agentscope.core.event.ToolResultStartEvent; +import io.agentscope.core.event.ToolResultTextDeltaEvent; +import io.agentscope.core.message.Msg; +import io.agentscope.core.message.ToolResultState; +import io.agentscope.core.message.ToolUseBlock; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import reactor.core.publisher.Flux; + +/** + * Unit tests for the {@link AguiAgentAdapter} v2 event-stream migration and the human-in-the-loop + * (HITL) bridging added to close agentscope-java issue #2437. + * + *

These tests exercise the {@code agent instanceof ReActAgent} branch, which consumes the + * fine-grained {@link AgentEvent} stream from {@link ReActAgent#streamEvents(List, RuntimeContext)} + * and translates it into AG-UI protocol events. The existing {@code AguiAgentAdapterTest} continues + * to cover the deprecated v1 {@code stream(...)} fallback path. + */ +class AguiAgentAdapterV2Test { + + private static final String REPLY = "reply-1"; + + private ReActAgent mockReActAgent() { + return mock(ReActAgent.class); + } + + private AguiAgentAdapter adapterFor(ReActAgent agent, AguiAdapterConfig config) { + return new AguiAgentAdapter(agent, config); + } + + private void stubStream(ReActAgent agent, AgentEvent... events) { + when(agent.streamEvents(anyList(), any(RuntimeContext.class))) + .thenReturn(Flux.fromArray(events)); + } + + private RunAgentInput input() { + return RunAgentInput.builder() + .threadId("thread-1") + .runId("run-1") + .messages(List.of(AguiMessage.userMessage("m-1", "Hello"))) + .build(); + } + + private RunAgentInput inputWithForwardedProps(Map props) { + return RunAgentInput.builder() + .threadId("thread-1") + .runId("run-1") + .messages(List.of(AguiMessage.userMessage("m-1", "Hello"))) + .forwardedProps(props) + .build(); + } + + // ------------------------------------------------------------------ + // v2 migration: text streaming + // ------------------------------------------------------------------ + + @Test + void textBlockEventsMapToTextMessageEvents() { + ReActAgent agent = mockReActAgent(); + stubStream( + agent, + new TextBlockStartEvent(REPLY, "blk-1"), + new TextBlockDeltaEvent(REPLY, "blk-1", "Hello "), + new TextBlockDeltaEvent(REPLY, "blk-1", "world"), + new TextBlockEndEvent(REPLY, "blk-1")); + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + assertInstanceOf(AguiEvent.RunStarted.class, events.get(0)); + + AguiEvent.TextMessageStart start = findFirst(events, AguiEvent.TextMessageStart.class); + assertEquals("blk-1", start.messageId()); + + List contents = + findAll(events, AguiEvent.TextMessageContent.class); + assertEquals(2, contents.size()); + assertEquals("Hello ", contents.get(0).delta()); + assertEquals("world", contents.get(1).delta()); + + assertEquals(1, findAll(events, AguiEvent.TextMessageEnd.class).size()); + assertInstanceOf(AguiEvent.RunFinished.class, events.get(events.size() - 1)); + } + + @Test + void textDeltaWithoutStartStillOpensMessage() { + ReActAgent agent = mockReActAgent(); + stubStream(agent, new TextBlockDeltaEvent(REPLY, "blk-1", "orphan")); + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + // A synthetic start must precede the content and an end must be emitted at finish. + assertEquals(1, findAll(events, AguiEvent.TextMessageStart.class).size()); + assertEquals(1, findAll(events, AguiEvent.TextMessageContent.class).size()); + assertEquals(1, findAll(events, AguiEvent.TextMessageEnd.class).size()); + } + + // ------------------------------------------------------------------ + // v2 migration: reasoning gated by config + // ------------------------------------------------------------------ + + @Test + void thinkingEventsAreSuppressedWhenReasoningDisabled() { + ReActAgent agent = mockReActAgent(); + stubStream( + agent, + new ThinkingBlockStartEvent(REPLY, "think-1"), + new ThinkingBlockDeltaEvent(REPLY, "think-1", "pondering"), + new ThinkingBlockEndEvent(REPLY, "think-1")); + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + assertTrue(findAll(events, AguiEvent.ReasoningMessageStart.class).isEmpty()); + assertTrue(findAll(events, AguiEvent.ReasoningMessageContent.class).isEmpty()); + } + + @Test + void thinkingEventsMapToReasoningWhenReasoningEnabled() { + ReActAgent agent = mockReActAgent(); + stubStream( + agent, + new ThinkingBlockStartEvent(REPLY, "think-1"), + new ThinkingBlockDeltaEvent(REPLY, "think-1", "pondering"), + new ThinkingBlockEndEvent(REPLY, "think-1")); + AguiAgentAdapter adapter = + adapterFor(agent, AguiAdapterConfig.builder().enableReasoning(true).build()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + assertEquals(1, findAll(events, AguiEvent.ReasoningMessageStart.class).size()); + assertEquals(1, findAll(events, AguiEvent.ReasoningMessageContent.class).size()); + assertEquals( + "pondering", findFirst(events, AguiEvent.ReasoningMessageContent.class).delta()); + assertEquals(1, findAll(events, AguiEvent.ReasoningMessageEnd.class).size()); + } + + // ------------------------------------------------------------------ + // v2 migration: tool calls and tool results + // ------------------------------------------------------------------ + + @Test + void toolCallAndResultEventsMapInOrder() { + ReActAgent agent = mockReActAgent(); + stubStream( + agent, + new ToolCallStartEvent(REPLY, "tc-1", "search"), + new ToolCallDeltaEvent(REPLY, "tc-1", "search", "{\"q\":"), + new ToolCallDeltaEvent(REPLY, "tc-1", "search", "\"cats\"}"), + new ToolCallEndEvent(REPLY, "tc-1", "search"), + new ToolResultStartEvent(REPLY, "tc-1", "search"), + new ToolResultTextDeltaEvent(REPLY, "tc-1", "search", "found "), + new ToolResultTextDeltaEvent(REPLY, "tc-1", "search", "5 cats"), + new ToolResultEndEvent(REPLY, "tc-1", "search", ToolResultState.SUCCESS)); + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + + AguiEvent.ToolCallStart start = findFirst(events, AguiEvent.ToolCallStart.class); + assertEquals("tc-1", start.toolCallId()); + assertEquals("search", start.toolCallName()); + + List args = findAll(events, AguiEvent.ToolCallArgs.class); + assertEquals(2, args.size()); + assertEquals("{\"q\":", args.get(0).delta()); + assertEquals("\"cats\"}", args.get(1).delta()); + + // The tool call must be closed exactly once. + assertEquals(1, findAll(events, AguiEvent.ToolCallEnd.class).size()); + + AguiEvent.ToolCallResult result = findFirst(events, AguiEvent.ToolCallResult.class); + assertEquals("tc-1", result.toolCallId()); + assertEquals("found 5 cats", result.content()); + + // Ordering: ToolCallEnd must precede ToolCallResult. + assertTrue( + indexOf(events, AguiEvent.ToolCallEnd.class) + < indexOf(events, AguiEvent.ToolCallResult.class)); + } + + @Test + void toolCallArgsSuppressedWhenDisabled() { + ReActAgent agent = mockReActAgent(); + stubStream( + agent, + new ToolCallStartEvent(REPLY, "tc-1", "search"), + new ToolCallDeltaEvent(REPLY, "tc-1", "search", "{\"q\":1}"), + new ToolCallEndEvent(REPLY, "tc-1", "search")); + AguiAgentAdapter adapter = + adapterFor(agent, AguiAdapterConfig.builder().emitToolCallArgs(false).build()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + assertTrue(findAll(events, AguiEvent.ToolCallArgs.class).isEmpty()); + assertEquals(1, findAll(events, AguiEvent.ToolCallStart.class).size()); + assertEquals(1, findAll(events, AguiEvent.ToolCallEnd.class).size()); + } + + // ------------------------------------------------------------------ + // HITL outbound: RequireUserConfirmEvent -> RunFinished interrupt outcome + // ------------------------------------------------------------------ + + @Test + void requireUserConfirmProducesInterruptOutcomeOnRunFinished() { + ReActAgent agent = mockReActAgent(); + ToolUseBlock pending = + ToolUseBlock.builder() + .id("tc-danger") + .name("delete_file") + .input(Map.of("path", "/etc/hosts")) + .build(); + stubStream( + agent, + new TextBlockStartEvent(REPLY, "blk-1"), + new TextBlockDeltaEvent(REPLY, "blk-1", "I need permission"), + new RequireUserConfirmEvent(REPLY, List.of(pending))); + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + + // Any dangling text message must be closed before pausing. + assertEquals(1, findAll(events, AguiEvent.TextMessageEnd.class).size()); + + AguiEvent last = events.get(events.size() - 1); + AguiEvent.RunFinished finished = assertInstanceOf(AguiEvent.RunFinished.class, last); + assertNotNull(finished.outcome()); + AguiEvent.RunFinishedInterruptOutcome outcome = + assertInstanceOf(AguiEvent.RunFinishedInterruptOutcome.class, finished.outcome()); + assertEquals(1, outcome.interrupts().size()); + AguiEvent.Interrupt interrupt = outcome.interrupts().get(0); + assertEquals("tc-danger", interrupt.toolCallId()); + assertEquals(AguiAgentAdapter.CONFIRM_INTERRUPT_REASON, interrupt.reason()); + assertNotNull(interrupt.id()); + } + + @Test + void normalCompletionProducesRunFinishedWithoutInterrupt() { + ReActAgent agent = mockReActAgent(); + stubStream(agent, new TextBlockDeltaEvent(REPLY, "blk-1", "done")); + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + + List events = adapter.run(input()).collectList().block(); + + assertNotNull(events); + AguiEvent.RunFinished finished = + assertInstanceOf(AguiEvent.RunFinished.class, events.get(events.size() - 1)); + assertFalse(finished.outcome() instanceof AguiEvent.RunFinishedInterruptOutcome); + } + + // ------------------------------------------------------------------ + // HITL inbound: forwardedProps confirm results -> ConfirmResult metadata + // ------------------------------------------------------------------ + + @Test + void forwardedConfirmResultsAreAttachedToResumedMessage() { + ReActAgent agent = mockReActAgent(); + ArgumentCaptor> msgsCaptor = ArgumentCaptor.forClass(List.class); + when(agent.streamEvents(msgsCaptor.capture(), any(RuntimeContext.class))) + .thenReturn(Flux.empty()); + + Map confirmEntry = new java.util.HashMap<>(); + confirmEntry.put("toolCallId", "tc-danger"); + confirmEntry.put("confirmed", true); + confirmEntry.put("toolName", "delete_file"); + confirmEntry.put("input", Map.of("path", "/tmp/x")); + Map props = + Map.of(AguiAgentAdapter.FORWARDED_PROPS_CONFIRM_RESULTS_KEY, List.of(confirmEntry)); + + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + adapter.run(inputWithForwardedProps(props)).collectList().block(); + + List forwarded = msgsCaptor.getValue(); + assertNotNull(forwarded); + assertFalse(forwarded.isEmpty()); + Msg last = forwarded.get(forwarded.size() - 1); + assertNotNull(last.getMetadata()); + Object rawResults = last.getMetadata().get(Msg.METADATA_CONFIRM_RESULTS); + assertInstanceOf(List.class, rawResults); + List confirmResults = (List) rawResults; + assertEquals(1, confirmResults.size()); + ConfirmResult cr = assertInstanceOf(ConfirmResult.class, confirmResults.get(0)); + assertTrue(cr.isConfirmed()); + assertEquals("tc-danger", cr.getToolCall().getId()); + assertEquals("delete_file", cr.getToolCall().getName()); + // The resumed tool call must carry a raw-args JSON string in content: the tool executor + // validates getContent() (not the parsed input map), and applying the ConfirmResult fully + // replaces the stored ToolUseBlock. A null content would fail schema validation. + assertNotNull(cr.getToolCall().getContent()); + assertTrue(cr.getToolCall().getContent().contains("/tmp/x")); + } + + @Test + void noForwardedConfirmResultsLeavesMessagesUnchanged() { + ReActAgent agent = mockReActAgent(); + ArgumentCaptor> msgsCaptor = ArgumentCaptor.forClass(List.class); + when(agent.streamEvents(msgsCaptor.capture(), any(RuntimeContext.class))) + .thenReturn(Flux.empty()); + + AguiAgentAdapter adapter = adapterFor(agent, AguiAdapterConfig.defaultConfig()); + adapter.run(input()).collectList().block(); + + List forwarded = msgsCaptor.getValue(); + assertNotNull(forwarded); + Msg last = forwarded.get(forwarded.size() - 1); + boolean hasConfirm = + last.getMetadata() != null + && last.getMetadata().containsKey(Msg.METADATA_CONFIRM_RESULTS); + assertFalse(hasConfirm); + } + + // ------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------ + + private static T findFirst(List events, Class type) { + for (AguiEvent e : events) { + if (type.isInstance(e)) { + return type.cast(e); + } + } + throw new AssertionError("No event of type " + type.getSimpleName()); + } + + private static List findAll(List events, Class type) { + List matches = new ArrayList<>(); + for (AguiEvent e : events) { + if (type.isInstance(e)) { + matches.add(type.cast(e)); + } + } + return matches; + } + + private static int indexOf(List events, Class type) { + for (int i = 0; i < events.size(); i++) { + if (type.isInstance(events.get(i))) { + return i; + } + } + return -1; + } +} diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java index 78ffa6943c..b522d73dbb 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java @@ -19,6 +19,7 @@ import io.agentscope.core.ReActAgent; import io.agentscope.core.agent.Agent; import io.agentscope.core.agent.Event; +import io.agentscope.core.agent.EventStreamingAgent; import io.agentscope.core.agent.RuntimeContext; import io.agentscope.core.agent.StreamOptions; import io.agentscope.core.agent.config.ModelConfig; @@ -149,7 +150,7 @@ * {@link io.agentscope.core.agent.RuntimeContext}'s {@code (userId, sessionId)} to isolate state. * Calls targeting the same session are serialized automatically; different sessions run in parallel. */ -public class HarnessAgent implements Agent, AutoCloseable { +public class HarnessAgent implements Agent, EventStreamingAgent, AutoCloseable { private static final Logger log = LoggerFactory.getLogger(HarnessAgent.class);