feat(streaming): add semantic text output disposition events #3013
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,254 @@ | ||
| /* | ||
| * 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 org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| 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}. | ||
| * | ||
| * <p>The wrapper is purely additive and never fails the source stream. An {@link | ||
| * AgentEndEvent} is forwarded as soon as it is observed, preceded by the derived terminal | ||
| * disposition when one applies, and any event that arrives after the end of a source is | ||
| * forwarded unchanged rather than rejected. Per-source bookkeeping is reclaimed when the source | ||
| * ends, so a subscription that fans out to many subagents does not retain them. | ||
| * | ||
| * <p>A top-level terminal requires an authoritative {@link AgentResultEvent} for the same | ||
| * source. A subagent terminal additionally requires a synthesized {@link AgentEndEvent} that | ||
| * explicitly reports {@link AgentEndEvent#OUTCOME_SUCCESS}: an unknown or abnormal outcome never | ||
| * closes the reply. | ||
| * | ||
| * @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 Logger log = LoggerFactory.getLogger(AgentEventStreams.class); | ||
|
|
||
| static final class DispositionAnnotator { | ||
|
|
||
| private final ReplyLifecycleTracker tracker = new ReplyLifecycleTracker(); | ||
| private final Map<SourceKey, AgentResultEvent> authoritativeResults = new LinkedHashMap<>(); | ||
| private final Set<SourceKey> endedSources = new HashSet<>(); | ||
|
|
||
| Flux<AgentEvent> apply(Flux<AgentEvent> source) { | ||
| return source.concatMap(event -> Flux.fromIterable(process(event)), 1); | ||
| } | ||
|
|
||
| /** | ||
| * Number of sources whose per-subscription bookkeeping is still retained. Package private so | ||
| * regression tests can assert that ended sources are reclaimed. | ||
| */ | ||
| int retainedSourceCount() { | ||
| return endedSources.size() + authoritativeResults.size(); | ||
| } | ||
|
|
||
| private List<AgentEvent> process(AgentEvent event) { | ||
| SourceKey sourceKey = tracker.sourceKey(event); | ||
| if (endedSources.contains(sourceKey)) { | ||
|
Collaborator
There was a problem hiding this comment. Failing the whole stream with |
||
| if (event instanceof ModelCallStartEvent) { | ||
| // A new model call on a closed source starts a fresh invocation. | ||
| endedSources.remove(sourceKey); | ||
| } else { | ||
| // A disposition is an opt-in, additive signal, so an ordering anomaly must | ||
| // never fail the wrapped stream: forward the event without classifying it. | ||
| log.warn( | ||
| "Received {} after AgentEndEvent for source {}; forwarding it without " | ||
| + "a disposition classification", | ||
| event.getType(), | ||
| sourceKey); | ||
| return List.of(event); | ||
| } | ||
| } | ||
|
|
||
| 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_RESULT -> onAgentResult((AgentResultEvent) event, observation); | ||
| case AGENT_END -> onAgentEnd((AgentEndEvent) event, observation); | ||
| default -> List.of(event); | ||
| }; | ||
| } | ||
|
|
||
| private List<AgentEvent> onAgentResult(AgentResultEvent event, Observation observation) { | ||
| // Only the top-level terminal consults the authoritative result; a subagent terminal is | ||
| // derived from its own end event. Recording child results would retain every subagent | ||
| // that never reports an end for the life of the subscription. | ||
| if (observation.sourceKey().isTopLevel()) { | ||
| authoritativeResults.put(observation.sourceKey(), event); | ||
| } | ||
| return 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(); | ||
|
|
||
| ReplySnapshot current = observation.after(); | ||
| AgentResultEvent result = authoritativeResults.remove(sourceKey); | ||
| List<AgentEvent> output = new ArrayList<>(2); | ||
| if (hasUnclassifiedText(current)) { | ||
| if (sourceKey.isTopLevel()) { | ||
| // ReActAgent emits the authoritative result immediately before the end event, | ||
| // so the top-level terminal disposition can be derived inline as well. | ||
| if (!isAbnormalEnd(event) && result != null && result.getResult() != null) { | ||
| output.add( | ||
| disposition( | ||
| current.replyId(), | ||
| TextOutputDisposition.TERMINAL, | ||
| result.getResult().getGenerateReason(), | ||
| event)); | ||
| tracker.markDispositionEmitted(sourceKey); | ||
| } | ||
| } else if (isSuccessfulChildEnd(event)) { | ||
| // A synthesized child end only closes the reply when it explicitly reports | ||
| // success, so an unknown or abnormal outcome never yields a terminal. | ||
| output.add( | ||
| disposition( | ||
| current.replyId(), | ||
| TextOutputDisposition.TERMINAL, | ||
| null, | ||
| event)); | ||
| tracker.markDispositionEmitted(sourceKey); | ||
| } | ||
| } | ||
| output.add(event); | ||
| tracker.clearSource(sourceKey); | ||
| if (sourceKey.isTopLevel()) { | ||
| // The top-level key is reused by the next user turn, so one tombstone is enough: | ||
| // the next model call reopens it and trailing events stay gated until then. | ||
| endedSources.add(sourceKey); | ||
| } else { | ||
| // Child keys are minted per invocation, so a tombstone each would retain every | ||
| // subagent this subscription has ever seen. The reply state is reclaimed above and | ||
| // a reused key simply starts a fresh source. | ||
| endedSources.remove(sourceKey); | ||
| } | ||
| return output; | ||
| } | ||
|
|
||
| private static Object invocationOutcome(AgentEndEvent end) { | ||
| return end.getMetadata() == null | ||
| ? null | ||
| : end.getMetadata().get(AgentEndEvent.METADATA_INVOCATION_OUTCOME); | ||
| } | ||
|
|
||
| private static boolean isSuccessfulChildEnd(AgentEndEvent end) { | ||
| Object outcome = invocationOutcome(end); | ||
| return outcome != null && AgentEndEvent.OUTCOME_SUCCESS.equals(outcome.toString()); | ||
| } | ||
|
|
||
| private static boolean isAbnormalEnd(AgentEndEvent end) { | ||
| Object outcome = invocationOutcome(end); | ||
| 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()); | ||
| // Carry the correlation keys only: metadata on the trigger describes the triggering | ||
| // invocation (for example its outcome), not this derived classification. | ||
| if (trigger.getMetadata() != null) { | ||
| Object taskId = trigger.getMetadata().get(AgentEvent.METADATA_TASK_ID); | ||
| if (taskId != null) { | ||
| event.withMetadataEntry(AgentEvent.METADATA_TASK_ID, taskId); | ||
| } | ||
| } | ||
| return event; | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Moving
.doOnNext(this::publishEvent)from the tail ofactingStream(...)out to the middleware-wrapped stream changes who controls publication: tool/text events are now published only if a middleware forwards the stream. A middleware that short-circuits, retries or filtersonActingused to still publish the core events and now will not (and conversely, events a middleware injects are published where they weren't before). That may well be the intent, but it touches the HITL paths this file keeps churning on (#3099/#3104), so could you confirm with a test that a stop/deny path still emits the permission-denied tool result events end-to-end?