feat(streaming): add semantic text output disposition events

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

@@ -2744,7 +2744,8 @@ private Mono<Msg> acting(int iter) {
                                                         MiddlewareBase::onActing,
                                                         actingCore)
                                                 .apply(new ActingInput(toolCalls));
-                                return stream.doOnNext(
+                                return stream.doOnNext(this::publishEvent)
+                                        .doOnNext(
                                                 ev -> {
                                                     if (ev instanceof RequestStopEvent rs) {
                                                         actingStopRequested.compareAndSet(null, rs);
@@ -2884,8 +2885,7 @@ Flux<AgentEvent> actingStream(
                                         new RequestStopEvent(
                                                 "permission asking",
                                                 GenerateReason.PERMISSION_ASKING));
-                            })
-                    .doOnNext(this::publishEvent);
+                            });
         }
 
         /**
@@ -3841,7 +3841,8 @@ private Mono<Msg> emitAllToolsDeniedThroughMiddleware(
                                     core)
                             .apply(new ActingInput(deniedToolCalls));
 
-            return stream.doOnNext(
+            return stream.doOnNext(this::publishEvent)
+                    .doOnNext(
                             ev -> {
                                 if (ev instanceof RequestStopEvent rs) {
                                     stopRef.compareAndSet(null, rs);

agentscope-core/src/main/java/io/agentscope/core/event/AgentEndEvent.java

@@ -23,6 +23,13 @@
  */
 public class AgentEndEvent extends AgentEvent {
 
+    /** Metadata key describing whether a synthesized invocation end succeeded, failed, or cancelled. */
+    public static final String METADATA_INVOCATION_OUTCOME = "invocationOutcome";
+
+    public static final String OUTCOME_SUCCESS = "success";
+    public static final String OUTCOME_ERROR = "error";
+    public static final String OUTCOME_CANCELLED = "cancelled";
+
     private final String replyId;
 
     @JsonCreator

agentscope-core/src/main/java/io/agentscope/core/event/AgentEvent.java

@@ -37,6 +37,7 @@
     @JsonSubTypes.Type(value = AgentStartEvent.class, name = "AGENT_START"),
     @JsonSubTypes.Type(value = AgentEndEvent.class, name = "AGENT_END"),
     @JsonSubTypes.Type(value = AgentResultEvent.class, name = "AGENT_RESULT"),
+    @JsonSubTypes.Type(value = TextOutputDispositionEvent.class, name = "TEXT_OUTPUT_DISPOSITION"),
     @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"),

agentscope-core/src/main/java/io/agentscope/core/event/AgentEventStreams.java

@@ -0,0 +1,196 @@
+/*
+ * 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.event;
+
+import io.agentscope.core.internal.stream.ReplyLifecycleTracker;
+import io.agentscope.core.internal.stream.ReplyLifecycleTracker.Observation;
+import io.agentscope.core.internal.stream.ReplyLifecycleTracker.ReplySnapshot;
+import io.agentscope.core.internal.stream.ReplyLifecycleTracker.SourceKey;
+import io.agentscope.core.message.GenerateReason;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import reactor.core.publisher.Flux;
+
+/** Utilities for deriving optional lifecycle signals from an {@link AgentEvent} stream. */
+public final class AgentEventStreams {
+
+    private AgentEventStreams() {}
+
+    /**
+     * Adds text output disposition events without changing the source stream itself.
+     *
+     * <p>State is isolated per subscription and per {@code source + taskId}. The authoritative
+     * invocation result remains {@link AgentResultEvent}; a terminal disposition only closes the
+     * last visible reply before a normally completed {@link AgentEndEvent}.
+     *
+     * @param source source event stream
+     * @return a deferred stream containing the original events and derived disposition events
+     */
+    public static Flux<AgentEvent> withTextOutputDisposition(Flux<AgentEvent> source) {
+        Objects.requireNonNull(source, "source");
+        return Flux.defer(() -> new DispositionAnnotator().apply(source));
+    }
+
+    private static final class DispositionAnnotator {
+
+        private final ReplyLifecycleTracker tracker = new ReplyLifecycleTracker();
+        private final Map<SourceKey, AgentEndEvent> pendingTopLevelEnds = new LinkedHashMap<>();
+        private final Set<SourceKey> endedSources = new HashSet<>();
+
+        private Flux<AgentEvent> apply(Flux<AgentEvent> source) {
+            Flux<AgentEvent> processed =
+                    source.concatMap(event -> Flux.fromIterable(process(event)), 1);
+            return processed.concatWith(Flux.defer(() -> Flux.fromIterable(complete())));
+        }
+
+        private List<AgentEvent> process(AgentEvent event) {
+            SourceKey sourceKey = tracker.sourceKey(event);
+            if (endedSources.contains(sourceKey)) {
+                throw new IllegalStateException(
+                        "Received event after AgentEndEvent for source " + sourceKey);
+            }
+
+            Observation observation = tracker.observe(event);
+            return switch (observation.kind()) {
+                case MODEL_CALL_START -> onModelCallStart(event, observation);
+                case TOOL_CALL_START -> onToolCallStart(event, observation);
+                case TEXT_BLOCK_END -> onTextBlockEnd(event, observation);
+                case AGENT_END -> onAgentEnd((AgentEndEvent) event, observation);
+                default -> List.of(event);
+            };
+        }
+
+        private List<AgentEvent> onModelCallStart(AgentEvent event, Observation observation) {
+            ReplySnapshot previous = observation.before();
+            if (hasUnclassifiedText(previous)) {
+                return List.of(
+                        disposition(
+                                previous.replyId(),
+                                TextOutputDisposition.INTERMEDIATE,
+                                null,
+                                event),
+                        event);
+            }
+            return List.of(event);
+        }
+
+        private List<AgentEvent> onToolCallStart(AgentEvent event, Observation observation) {
+            ReplySnapshot current = observation.after();
+            if (observation.currentReplyEvent() && hasUnclassifiedText(current)) {
+                tracker.markDispositionEmitted(observation.sourceKey());
+                return List.of(
+                        disposition(
+                                current.replyId(), TextOutputDisposition.INTERMEDIATE, null, event),
+                        event);
+            }
+            return List.of(event);
+        }
+
+        private List<AgentEvent> onTextBlockEnd(AgentEvent event, Observation observation) {
+            ReplySnapshot current = observation.after();
+            if (observation.currentReplyEvent()
+                    && current.toolCallSeen()
+                    && hasUnclassifiedText(current)) {
+                tracker.markDispositionEmitted(observation.sourceKey());
+                return List.of(
+                        event,
+                        disposition(
+                                current.replyId(),
+                                TextOutputDisposition.INTERMEDIATE,
+                                null,
+                                event));
+            }
+            return List.of(event);
+        }
+
+        private List<AgentEvent> onAgentEnd(AgentEndEvent event, Observation observation) {
+            SourceKey sourceKey = observation.sourceKey();
+            endedSources.add(sourceKey);
+            if (sourceKey.isTopLevel()) {
+                pendingTopLevelEnds.put(sourceKey, event);
+                return List.of();
+            }
+
+            ReplySnapshot current = observation.after();
+            List<AgentEvent> output = new ArrayList<>(2);
+            if (isNormallyCompleted(event) && hasUnclassifiedText(current)) {
+                output.add(
+                        disposition(
+                                current.replyId(), TextOutputDisposition.TERMINAL, null, event));
+                tracker.markDispositionEmitted(sourceKey);
+            }
+            output.add(event);
+            return output;
+        }
+
+        private List<AgentEvent> complete() {
+            List<AgentEvent> output = new ArrayList<>(pendingTopLevelEnds.size() * 2);
+            for (Map.Entry<SourceKey, AgentEndEvent> entry : pendingTopLevelEnds.entrySet()) {
+                SourceKey sourceKey = entry.getKey();
+                AgentEndEvent end = entry.getValue();
+                ReplySnapshot current = tracker.snapshot(sourceKey);
+                AgentResultEvent result = current.lastResult();
+                if (isNormallyCompleted(end)
+                        && hasUnclassifiedText(current)
+                        && result != null
+                        && result.getResult() != null) {
+                    GenerateReason reason = result.getResult().getGenerateReason();
+                    output.add(
+                            disposition(
+                                    current.replyId(),
+                                    TextOutputDisposition.TERMINAL,
+                                    reason,
+                                    end));
+                    tracker.markDispositionEmitted(sourceKey);
+                }
+                output.add(end);
+                tracker.clearSource(sourceKey);
+            }
+            pendingTopLevelEnds.clear();
+            return output;
+        }
+
+        private static boolean isNormallyCompleted(AgentEndEvent end) {
+            Object outcome =
+                    end.getMetadata() == null
+                            ? null
+                            : end.getMetadata().get(AgentEndEvent.METADATA_INVOCATION_OUTCOME);
+            return outcome == null || AgentEndEvent.OUTCOME_SUCCESS.equals(outcome.toString());
+        }
+
+        private static boolean hasUnclassifiedText(ReplySnapshot snapshot) {
+            return snapshot.replyId() != null
+                    && snapshot.textSeen()
+                    && !snapshot.dispositionEmitted();
+        }
+
+        private static TextOutputDispositionEvent disposition(
+                String replyId,
+                TextOutputDisposition disposition,
+                GenerateReason generateReason,
+                AgentEvent trigger) {
+            TextOutputDispositionEvent event =
+                    new TextOutputDispositionEvent(replyId, disposition, generateReason);
+            event.withSource(trigger.getSource()).withMetadata(trigger.getMetadata());
+            return event;
+        }
+    }
+}

agentscope-core/src/main/java/io/agentscope/core/event/AgentEventType.java

@@ -43,6 +43,7 @@ public enum AgentEventType {
     @JsonAlias({"RUN_FINISHED", "REPLY_END"})
     AGENT_END("AGENT_END"),
     AGENT_RESULT("AGENT_RESULT"),
+    TEXT_OUTPUT_DISPOSITION("TEXT_OUTPUT_DISPOSITION"),
 
     @JsonAlias({"MODEL_CALL_STARTED"})
     MODEL_CALL_START("MODEL_CALL_START"),

agentscope-core/src/main/java/io/agentscope/core/event/TextOutputDisposition.java

@@ -0,0 +1,22 @@
+/*
+ * 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.event;
+
+/** Describes whether a streamed model reply is intermediate or terminal for an invocation. */
+public enum TextOutputDisposition {
+    INTERMEDIATE,
+    TERMINAL
+}

agentscope-core/src/main/java/io/agentscope/core/event/TextOutputDispositionEvent.java

@@ -0,0 +1,71 @@
+/*
+ * 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.event;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.agentscope.core.message.GenerateReason;
+import java.util.Objects;
+
+/**
+ * Classifies an already-streamed model reply as intermediate or terminal for its agent invocation.
+ *
+ * <p>A terminal disposition is a lifecycle signal, not an authoritative final answer. Consumers
+ * must use {@link AgentResultEvent} for the invocation result.
+ */
+public final class TextOutputDispositionEvent extends AgentEvent {
+
+    private final String replyId;
+    private final TextOutputDisposition disposition;
+    private final GenerateReason generateReason;
+
+    public TextOutputDispositionEvent(
+            String replyId, TextOutputDisposition disposition, GenerateReason generateReason) {
+        this.replyId = replyId;
+        this.disposition = Objects.requireNonNull(disposition, "disposition");
+        this.generateReason = generateReason;
+    }
+
+    @JsonCreator
+    public TextOutputDispositionEvent(
+            @JsonProperty("id") String id,
+            @JsonProperty("createdAt") String createdAt,
+            @JsonProperty("replyId") String replyId,
+            @JsonProperty("disposition") TextOutputDisposition disposition,
+            @JsonProperty("generateReason") GenerateReason generateReason) {
+        super(id, createdAt);
+        this.replyId = replyId;
+        this.disposition = Objects.requireNonNull(disposition, "disposition");
+        this.generateReason = generateReason;
+    }
+
+    @Override
+    public AgentEventType getType() {
+        return AgentEventType.TEXT_OUTPUT_DISPOSITION;
+    }
+
+    public String getReplyId() {
+        return replyId;
+    }
+
+    public TextOutputDisposition getDisposition() {
+        return disposition;
+    }
+
+    public GenerateReason getGenerateReason() {
+        return generateReason;
+    }
+}