feat(core): add AgentEvent system + Middleware framework + event stream refactoring by flystar32 · Pull Request #1449 · agentscope-ai/agentscope-java

flystar32 · 2026-05-20T11:20:19Z

Summary

Aligns AgentScope Java with Python 2.0 by adding a fine-grained event system and middleware framework, while refactoring Agent internals to emit Flux<AgentEvent> event streams.

Phase 1 — Event system

25 AgentEvent types in io.agentscope.core.event package: ReplyStart/End, ModelCallStart/End, TextBlock/ThinkingBlock/ToolCall/ToolResult Start/Delta/End, ExceedMaxIters, RequireUserConfirm, RequireExternalExecution, etc.
New message types: ToolCallState, ToolResultState enums, HintBlock content block
ContentBlock / ToolResultBlock / ToolUseBlock: added state fields for event tracking

Phase 2 — Middleware framework

Middleware interface with 5 interception points:
- Onion pattern (4): onReply, onReasoning, onActing, onModelCall
- Pipeline pattern (1): onSystemPrompt
MiddlewareChain utility: builds back-to-front onion chains via MiddlewareChain.build()
Input records: ReplyInput, ReasoningInput, ActingInput, ModelCallInput
Builder support: ReActAgent.builder().middleware(mw).build()

Phase 3 — Agent event stream refactoring

ReActAgent internals refactored: reasoning() / acting() / summarizing() now produce Flux<AgentEvent> via reasoningStream() / actingStream() / summaryStream()
New replyImpl() orchestrates the full ReAct loop as an event stream (ReplyStart → reasoning/acting events → ReplyEnd)
Middleware wired at all 5 interception points using MiddlewareChain.build()
New public API: ReActAgent.streamEvents(msgs) → Flux<AgentEvent>

Backward compatibility

call() → Mono<Msg> signature unchanged
stream() → Flux<Event> signature unchanged
Hook system fully preserved and still called alongside new event stream
All 3807 existing tests pass with 0 failures

Test plan

mvn compile -pl agentscope-core — compiles clean
mvn test -pl agentscope-core — 3807 tests pass, 0 failures, 0 errors
Full project compile (all modules except pre-existing graceful-shutdown issue) — passes
Manual verification with middleware example (logging middleware wrapping reasoning)

… + event stream refactoring Aligns AgentScope Java with Python 2.0 by introducing: Phase 1 — Event system: - 25 fine-grained AgentEvent types (io.agentscope.core.event package) - ToolCallState, ToolResultState enums and HintBlock content type Phase 2 — Middleware + event stream: - Middleware interface with 5 interception points (onReply, onReasoning, onActing, onModelCall — onion pattern; onSystemPrompt — pipeline pattern) - MiddlewareChain builder for composing middleware chains - ReActAgent internals refactored to Flux<AgentEvent> event streams (reasoningStream, actingStream, summaryStream, replyImpl) - New public streamEvents() API on ReActAgent - Builder.middleware() / middlewares() for registering middlewares Backward compatible: call() Mono<Msg> and stream() Flux<Event> signatures unchanged; Hook system preserved; all 3807 existing tests pass. Change-Id: I9a85bf3735e29a9cd722d39dce97b1cf36b02a52 Co-developed-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

flystar32 · 2026-05-20T11:37:20Z

CI failure analysis

The 2 failing tests in agentscope-harness are pre-existing failures on v2_dev, unrelated to this PR:

HarnessAgentSubagentStreamTest.stream_localSubagentEventsForwardedToParentFlux
HarnessAgentSubagentStreamTest.stream_eventOrdering_parentChildParent

Verified by running the same tests on the base commit (0f6a1db4) without any of this PR's changes — they fail identically. HarnessAgent implements Agent directly and does not extend ReActAgent, so this PR's changes to ReActAgent cannot affect these tests.

The graceful-shutdown example compile error (missing log variable) is also a pre-existing issue on v2_dev.

chickenlj

LGTM.

Copilot

Pull request overview

This PR introduces a fine-grained AgentEvent model plus a new middleware framework, and refactors ReActAgent internals to emit/react to Flux<AgentEvent>-style streams in order to align AgentScope Java with the Python 2.0 event/middleware architecture.

Changes:

Added io.agentscope.core.event event taxonomy (base AgentEvent, AgentEventType, and many concrete lifecycle/block/tool events).
Added middleware SPI (Middleware, MiddlewareChain) and input records (ReplyInput, ReasoningInput, ActingInput, ModelCallInput), plus ReActAgent.Builder support.
Extended message/content model with HintBlock and tool call/result state enums/fields for event tracking.

Reviewed changes

Copilot reviewed 41 out of 41 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java	Wires middleware hooks into reasoning/acting/model-call flows and adds `streamEvents`; refactors reasoning/acting/summarizing to use event streams.
agentscope-core/src/main/java/io/agentscope/core/middleware/Middleware.java	Defines middleware interception points (onion hooks + system prompt pipeline).
agentscope-core/src/main/java/io/agentscope/core/middleware/MiddlewareChain.java	Implements back-to-front onion chain construction for middleware hooks.
agentscope-core/src/main/java/io/agentscope/core/middleware/ReplyInput.java	Reply-phase middleware input record.
agentscope-core/src/main/java/io/agentscope/core/middleware/ReasoningInput.java	Reasoning-phase middleware input record.
agentscope-core/src/main/java/io/agentscope/core/middleware/ActingInput.java	Acting-phase middleware input record.
agentscope-core/src/main/java/io/agentscope/core/middleware/ModelCallInput.java	Model-call middleware input record.
agentscope-core/src/main/java/io/agentscope/core/message/ContentBlock.java	Registers `HintBlock` in polymorphic content block hierarchy.
agentscope-core/src/main/java/io/agentscope/core/message/HintBlock.java	Adds a new content block type for injecting hints into LLM reasoning.
agentscope-core/src/main/java/io/agentscope/core/message/ToolCallState.java	Introduces tool-call lifecycle state enum.
agentscope-core/src/main/java/io/agentscope/core/message/ToolResultState.java	Introduces tool-result lifecycle state enum.
agentscope-core/src/main/java/io/agentscope/core/message/ToolUseBlock.java	Adds `state` to tool-call blocks (constructor/builder/withState).
agentscope-core/src/main/java/io/agentscope/core/message/ToolResultBlock.java	Adds `state` to tool-result blocks (constructor/builder/withState).
agentscope-core/src/main/java/io/agentscope/core/event/AgentEvent.java	Adds the polymorphic base class for all fine-grained agent events.
agentscope-core/src/main/java/io/agentscope/core/event/AgentEventType.java	Defines the enum of event types emitted during execution.
agentscope-core/src/main/java/io/agentscope/core/event/ReplyStartEvent.java	Reply lifecycle start event.
agentscope-core/src/main/java/io/agentscope/core/event/ReplyEndEvent.java	Reply lifecycle end event.
agentscope-core/src/main/java/io/agentscope/core/event/ModelCallStartEvent.java	Model-call start event.
agentscope-core/src/main/java/io/agentscope/core/event/ModelCallEndEvent.java	Model-call end event (includes usage).
agentscope-core/src/main/java/io/agentscope/core/event/TextBlockStartEvent.java	Text block start event.
agentscope-core/src/main/java/io/agentscope/core/event/TextBlockDeltaEvent.java	Text block delta event.
agentscope-core/src/main/java/io/agentscope/core/event/TextBlockEndEvent.java	Text block end event.
agentscope-core/src/main/java/io/agentscope/core/event/ThinkingBlockStartEvent.java	Thinking block start event.
agentscope-core/src/main/java/io/agentscope/core/event/ThinkingBlockDeltaEvent.java	Thinking block delta event.
agentscope-core/src/main/java/io/agentscope/core/event/ThinkingBlockEndEvent.java	Thinking block end event.
agentscope-core/src/main/java/io/agentscope/core/event/DataBlockStartEvent.java	Data block start event.
agentscope-core/src/main/java/io/agentscope/core/event/DataBlockDeltaEvent.java	Data block delta event.
agentscope-core/src/main/java/io/agentscope/core/event/DataBlockEndEvent.java	Data block end event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolCallStartEvent.java	Tool call start event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolCallDeltaEvent.java	Tool call delta event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolCallEndEvent.java	Tool call end event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolResultStartEvent.java	Tool result start event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolResultTextDeltaEvent.java	Tool result text delta event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolResultDataDeltaEvent.java	Tool result data delta event.
agentscope-core/src/main/java/io/agentscope/core/event/ToolResultEndEvent.java	Tool result end event (includes `ToolResultState`).
agentscope-core/src/main/java/io/agentscope/core/event/ExceedMaxItersEvent.java	Signals that the ReAct loop exceeded `maxIters`.
agentscope-core/src/main/java/io/agentscope/core/event/RequireUserConfirmEvent.java	Signals HITL confirmation is required before tool execution.
agentscope-core/src/main/java/io/agentscope/core/event/UserConfirmResultEvent.java	Captures the user’s confirmation decisions after HITL prompt.
agentscope-core/src/main/java/io/agentscope/core/event/ConfirmResult.java	DTO for a single tool-call confirmation decision.
agentscope-core/src/main/java/io/agentscope/core/event/RequireExternalExecutionEvent.java	Signals tools must be executed externally (out-of-process).
agentscope-core/src/main/java/io/agentscope/core/event/ExternalExecutionResultEvent.java	Captures externally executed tool results fed back to the agent.

Comments suppressed due to low confidence (2)

agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java:823

modelCallStream() generates a new UUID into a variable named replyId and uses it for ModelCallStart/End and block/tool events. This clashes with the replyId emitted by ReplyStartEvent and breaks correlation across the overall reply. Either pass the outer replyId into modelCallStream(), or rename this to modelCallId and add a dedicated field in events for the model call identifier.

    private Flux<AgentEvent> modelCallStream(
            ReasoningContext context, ModelCallInput mci, boolean withToolEvents) {

        String replyId = UUID.randomUUID().toString().replace("-", "");
        AtomicBoolean textStarted = new AtomicBoolean(false);
        AtomicBoolean thinkingStarted = new AtomicBoolean(false);
        Set<String> startedToolCalls = ConcurrentHashMap.newKeySet();

agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java:1191

summarizing() publishes ExceedMaxItersEvent with an empty replyId (""), which prevents consumers from correlating the exceed-max-iters signal to the active reply. This should use the same replyId emitted by ReplyStartEvent for the current execution.

        List<Msg> messageList = prepareSummaryMessages();
        GenerateOptions generateOptions = buildGenerateOptions();
        ReasoningContext context = new ReasoningContext(getName());
        publishEvent(new ExceedMaxItersEvent("", maxIters, maxIters));

+    private String applySystemPromptMiddlewares(String prompt) {
+        if (middlewares.isEmpty()) {
+            return prompt;
+        }
+        Mono<String> result = Mono.just(prompt);
+        for (Middleware mw : middlewares) {
+            result = result.flatMap(p -> mw.onSystemPrompt(this, p));
+        }
+        return result.block();


+    public Flux<AgentEvent> streamEvents(List<Msg> msgs) {
+        String replyId = UUID.randomUUID().toString().replace("-", "");
+        return Flux.<AgentEvent>create(
+                        sink -> {
+                            activeEventSink.set(sink);
+                            sink.next(new ReplyStartEvent(null, replyId, getName()));
+                            call(msgs)
+                                    .doFinally(
+                                            signal -> {
+                                                sink.next(new ReplyEndEvent(replyId));
+                                                activeEventSink.set(null);
+                                                sink.complete();
+                                            })
+                                    .subscribe(finalMsg -> {}, sink::error);
+                        },
+                        FluxSink.OverflowStrategy.BUFFER)


+    public Flux<AgentEvent> streamEvents(List<Msg> msgs) {
+        String replyId = UUID.randomUUID().toString().replace("-", "");
+        return Flux.<AgentEvent>create(
+                        sink -> {
+                            activeEventSink.set(sink);
+                            sink.next(new ReplyStartEvent(null, replyId, getName()));
+                            call(msgs)
+                                    .doFinally(
+                                            signal -> {
+                                                sink.next(new ReplyEndEvent(replyId));
+                                                activeEventSink.set(null);
+                                                sink.complete();


+    /**
+     * Execute the full reply as a {@link Flux} of fine-grained {@link AgentEvent}s.
+     *
+     * <p>This method wraps the existing {@code doCall()} logic and captures all events emitted
+     * by the internal stream methods ({@code reasoningStream}, {@code actingStream},
+     * {@code summaryStream}). The stream is bookended by {@link ReplyStartEvent} and
+     * {@link ReplyEndEvent}.
+     *
+     * @param msgs the input messages
+     * @return event stream covering the full reply lifecycle
+     */
+    Flux<AgentEvent> replyImpl(List<Msg> msgs) {
+        String replyId = UUID.randomUUID().toString().replace("-", "");
+
+        Function<ReplyInput, Flux<AgentEvent>> core =
+                input ->
+                        Flux.<AgentEvent>create(
+                                        sink -> {
+                                            activeEventSink.set(sink);
+                                            sink.next(
+                                                    new ReplyStartEvent(null, replyId, getName()));
+
+                                            doCall(input.msgs())
+                                                    .doFinally(
+                                                            signal -> {
+                                                                sink.next(
+                                                                        new ReplyEndEvent(replyId));
+                                                                activeEventSink.set(null);
+                                                                sink.complete();
+                                                            })
+                                                    .subscribe(finalMsg -> {}, sink::error);
+                                        },
+                                        FluxSink.OverflowStrategy.BUFFER)
+                                .doOnError(e -> activeEventSink.set(null));
+
+        return MiddlewareChain.build(middlewares, this, Middleware::onReply, core)
+                .apply(new ReplyInput(msgs));
+    }
+


+        return Flux.<AgentEvent>create(
+                        sink -> {
+                            for (ToolUseBlock tool : toolCalls) {
+                                sink.next(
+                                        new ToolResultStartEvent(
+                                                replyId, tool.getId(), tool.getName()));
+                            }
+
+                            toolkit.setInternalChunkCallback(
+                                    (toolUse, chunk) -> {
+                                        if (chunk.getOutput() != null) {
+                                            for (ContentBlock block : chunk.getOutput()) {
+                                                if (block instanceof TextBlock tb) {
+                                                    sink.next(
+                                                            new ToolResultTextDeltaEvent(
+                                                                    replyId,
+                                                                    toolUse.getId(),
+                                                                    tb.getText()));
+                                                } else {
+                                                    sink.next(
+                                                            new ToolResultDataDeltaEvent(
+                                                                    replyId,
+                                                                    toolUse.getId(),
+                                                                    block));
+                                                }
+                                            }
+                                        }
+                                        notifyActingChunk(toolUse, chunk).subscribe();
+                                    });


+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+@JsonSubTypes({
+    @JsonSubTypes.Type(value = ReplyStartEvent.class, name = "REPLY_START"),
+    @JsonSubTypes.Type(value = ReplyEndEvent.class, name = "REPLY_END"),
+    @JsonSubTypes.Type(value = ModelCallStartEvent.class, name = "MODEL_CALL_START"),
+    @JsonSubTypes.Type(value = ModelCallEndEvent.class, name = "MODEL_CALL_END"),
+    @JsonSubTypes.Type(value = TextBlockStartEvent.class, name = "TEXT_BLOCK_START"),
+    @JsonSubTypes.Type(value = TextBlockDeltaEvent.class, name = "TEXT_BLOCK_DELTA"),
+    @JsonSubTypes.Type(value = TextBlockEndEvent.class, name = "TEXT_BLOCK_END"),
+    @JsonSubTypes.Type(value = ThinkingBlockStartEvent.class, name = "THINKING_BLOCK_START"),
+    @JsonSubTypes.Type(value = ThinkingBlockDeltaEvent.class, name = "THINKING_BLOCK_DELTA"),
+    @JsonSubTypes.Type(value = ThinkingBlockEndEvent.class, name = "THINKING_BLOCK_END"),
+    @JsonSubTypes.Type(value = DataBlockStartEvent.class, name = "DATA_BLOCK_START"),
+    @JsonSubTypes.Type(value = DataBlockDeltaEvent.class, name = "DATA_BLOCK_DELTA"),
+    @JsonSubTypes.Type(value = DataBlockEndEvent.class, name = "DATA_BLOCK_END"),
+    @JsonSubTypes.Type(value = ToolCallStartEvent.class, name = "TOOL_CALL_START"),
+    @JsonSubTypes.Type(value = ToolCallDeltaEvent.class, name = "TOOL_CALL_DELTA"),
+    @JsonSubTypes.Type(value = ToolCallEndEvent.class, name = "TOOL_CALL_END"),
+    @JsonSubTypes.Type(value = ToolResultStartEvent.class, name = "TOOL_RESULT_START"),
+    @JsonSubTypes.Type(value = ToolResultTextDeltaEvent.class, name = "TOOL_RESULT_TEXT_DELTA"),
+    @JsonSubTypes.Type(value = ToolResultDataDeltaEvent.class, name = "TOOL_RESULT_DATA_DELTA"),
+    @JsonSubTypes.Type(value = ToolResultEndEvent.class, name = "TOOL_RESULT_END"),
+    @JsonSubTypes.Type(value = ExceedMaxItersEvent.class, name = "EXCEED_MAX_ITERS"),
+    @JsonSubTypes.Type(value = RequireUserConfirmEvent.class, name = "REQUIRE_USER_CONFIRM"),
+    @JsonSubTypes.Type(
+            value = RequireExternalExecutionEvent.class,
+            name = "REQUIRE_EXTERNAL_EXECUTION"),
+    @JsonSubTypes.Type(value = UserConfirmResultEvent.class, name = "USER_CONFIRM_RESULT"),
+    @JsonSubTypes.Type(
+            value = ExternalExecutionResultEvent.class,
+            name = "EXTERNAL_EXECUTION_RESULT")
+})
+public abstract class AgentEvent {
+
+    private final String id;
+    private final String createdAt;
+
+    protected AgentEvent() {
+        this.id = UUID.randomUUID().toString().replace("-", "");
+        this.createdAt = Instant.now().toString();
+    }
+
+    protected AgentEvent(String id, String createdAt) {
+        this.id = id;
+        this.createdAt = createdAt;
+    }
+
+    public abstract AgentEventType getType();
+


    @JsonCreator
    public ToolResultBlock(
            @JsonProperty("id") String id,
            @JsonProperty("name") String name,
            @JsonProperty("output") List<ContentBlock> output,
-            @JsonProperty("metadata") Map<String, Object> metadata) {
+            @JsonProperty("metadata") Map<String, Object> metadata,
+            @JsonProperty("state") ToolResultState state) {
        this.id = id;
        this.name = name;
        this.output = output != null ? List.copyOf(output) : List.of();
        this.metadata = metadata != null ? Map.copyOf(metadata) : Map.of();
+        this.state = state != null ? state : ToolResultState.RUNNING;
+    }


…am refactoring (#1449)

…am refactoring (agentscope-ai#1449)

chickenlj approved these changes May 21, 2026

View reviewed changes

chickenlj requested a review from Copilot May 21, 2026 01:25

Copilot AI reviewed May 21, 2026

View reviewed changes

chickenlj merged commit 882c88e into agentscope-ai:v2_dev May 21, 2026
6 of 8 checks passed

chickenlj pushed a commit that referenced this pull request May 26, 2026

feat(core): add AgentEvent system + Middleware framework + event stre…

fc678aa

…am refactoring (#1449)

lancefoever pushed a commit to lancefoever/agentscope-java that referenced this pull request Jun 2, 2026

feat(core): add AgentEvent system + Middleware framework + event stre…

e26e8ce

…am refactoring (agentscope-ai#1449)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(core): add AgentEvent system + Middleware framework + event stream refactoring#1449

feat(core): add AgentEvent system + Middleware framework + event stream refactoring#1449
chickenlj merged 1 commit into
agentscope-ai:v2_devfrom
flystar32:feat/middleware-event-stream

flystar32 commented May 20, 2026

Uh oh!

flystar32 commented May 20, 2026

Uh oh!

chickenlj left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

flystar32 commented May 20, 2026

Summary

Phase 1 — Event system

Phase 2 — Middleware framework

Phase 3 — Agent event stream refactoring

Backward compatibility

Test plan

Uh oh!

flystar32 commented May 20, 2026

CI failure analysis

Uh oh!

chickenlj left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants