This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 8171c2a1d fix(ai): keep a projected success terminal over a later stop
(#6385)
8171c2a1d is described below
commit 8171c2a1de5b26844225bdc7fc4c183e2a639d73
Author: 0 <[email protected]>
AuthorDate: Thu Oct 15 20:18:36 2026 +0800
fix(ai): keep a projected success terminal over a later stop (#6385)
* fix(ai): keep a projected success terminal over a later stop
decide() let an abort win unconditionally, so a stop that landed after
the success frame had already been projected - the run_status row, the
live run_finished, everything - wrote a second terminal event into the
timeline and rewrote the row to STOPPED for an answer the user watched
complete. The transcript rendered a phantom empty stopped bubble behind
the finished one.
The projected success is now checked first: once the projector has
persisted the terminal, finalisation honours it exactly-once. An abort
that landed before the success frame still wins, which is the
stopRacedSuccess case where the frame was swallowed and nothing terminal
was projected.
* test(ai): reuse StubAgentProvider via an afterEvent hook and share the
properties helper
---
.../studio/ops/ai/conversation/AiRunExecutor.java | 17 +++++---
.../ops/ai/conversation/AiRunExecutorTest.java | 49 ++++++++++++++++++++--
.../ops/ai/conversation/AiRunTestSupport.java | 13 +++++-
3 files changed, 69 insertions(+), 10 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
index 7d6b46839..db71840c7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
@@ -468,19 +468,24 @@ public class AiRunExecutor {
}
}
- /** Maps what happened onto the terminal state to write. An abort always
wins: it is the decision
- * a human or the server made about this run, and a provider frame that
arrived afterwards does not
- * unmake it. */
+ /**
+ * Maps what happened onto the terminal state to write. A success frame
that was already projected
+ * wins over everything: the projector persisted the terminal {@code
run_status} and published the
+ * live {@code run_finished} at that moment, so any abort that lands
afterwards (a user stop that
+ * raced the finalisation, say) must not write a second terminal event
over a finished run. An
+ * abort that landed <em>before</em> the success frame still wins - that
is the
+ * {@code stopRacedSuccess} case, where the frame was swallowed and
nothing terminal was projected.
+ */
private Terminal decide(RunContext context, Outcome outcome) {
+ if (outcome.successTerminalProjected) {
+ return new Terminal(RunStatus.COMPLETED, null, null, null,
outcome, true);
+ }
AbortReason abort = context.getHandle() == null
? null
: context.getHandle().abortReason().orElse(null);
if (abort != null) {
return new Terminal(abort.status(), abort.stopReason(), null,
null, outcome, false);
}
- if (outcome.successTerminalProjected) {
- return new Terminal(RunStatus.COMPLETED, null, null, null,
outcome, true);
- }
if (outcome.gatewayFailure != null) {
LlmGatewayException failure = outcome.gatewayFailure;
return new Terminal(RunStatus.FAILED,
stopReasonForCode(failure.getCode()), failure.getCode(),
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
index ff47659a8..ecda29e5d 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
@@ -149,10 +149,8 @@ class AiRunExecutorTest {
: "conversation:last_seq=" + update.getLastSeq());
return null;
}).when(conversationRepository).update(any(RmqAiConversation.class));
- AiConversationProperties properties = new AiConversationProperties();
- properties.setStopGrace(Duration.ofMillis(1));
executor = new AiRunExecutor(new
AgentProviderRegistry(List.of(provider)), llmClient, promptEnhancer,
- registry, runRepository, conversationRepository,
eventRepository, properties, objectMapper,
+ registry, runRepository, conversationRepository,
eventRepository, properties(), objectMapper,
AiRunTestSupport.directExecutor(),
AiRunTestSupport.recordingInto(emitters), scheduler,
Clock.systemUTC());
conversation = AiRunTestSupport.conversation(CONVERSATION_ID,
"tester");
@@ -218,6 +216,45 @@ class AiRunExecutorTest {
assertThat(registry.isLive(RUN_ID)).isFalse();
}
+ @Test
+ void
aStopThatLandsAfterTheSuccessFrameMustNotOverwriteTheCompletedRunTest() {
+ // The success frame is projected the moment it arrives - the
run_status row, the live
+ // run_finished, everything. A stop that lands after it (the user
pressed the button while
+ // the answer was already being finalised) can only find a finished
run: writing a second
+ // terminal would corrupt the timeline with two run_status events and
rewrite the row to
+ // STOPPED for an answer the user watched complete.
+ // The stop must land between the success frame and the projector's
terminal write — inside
+ // the stream, not before the script. The stub's afterEvent hook fires
exactly there: one
+ // scripted success frame, then requestStop runs with the frame's own
projection already done.
+ AiRunTestSupport.StubAgentProvider lateStop = new
AiRunTestSupport.StubAgentProvider();
+ lateStop.emit(new AgentEvent.ResultMeta("session-7", 10L, 3, 4,
+ AgentEventProjector.SUCCESS_SUBTYPE));
+ lateStop.afterEvent = (event, options) -> ((AgentRunHandle)
options.getProcessSink())
+ .requestStop(AbortReason.USER_STOP);
+ AiRunExecutor lateExecutor = new AiRunExecutor(new
AgentProviderRegistry(List.of(lateStop)),
+ llmClient, promptEnhancer, registry, runRepository,
conversationRepository,
+ eventRepository, properties(), objectMapper,
AiRunTestSupport.directExecutor(),
+ AiRunTestSupport.recordingInto(emitters), scheduler,
Clock.systemUTC());
+ AgentRunHandle handle = lateExecutor.newHandle(RUN_ID);
+ AiEventSink sink = lateExecutor.newSink(CONVERSATION_ID, RUN_ID, 1, 0);
+ session = lateExecutor.newSession(RUN_ID,
lateExecutor.streamTimeoutMillis(ENGINE));
+ registry.register(RUN_ID, handle);
+ sink.writeUser("hello", null);
+ registry.attach(RUN_ID, session);
+ session.finishReplay();
+ lateExecutor.submit(new AiRunExecutor.RunContext(conversation, run,
sink, handle,
+
LlmConfigVO.builder().engine(ENGINE).model("qwen3.8-max").enabled(true).build(),
+ ENGINE, "hello", null, false, Duration.ofSeconds(300), null));
+
+ assertThat(runRow().getStatus()).isEqualTo(RunStatus.COMPLETED.name());
+ assertThat(runRow().getStopReason()).isNull();
+ // Exactly one terminal event, and it is the completed one the
projector already wrote.
+ assertThat(types()).containsExactly("user", "run_status");
+ assertThat(payloads().get(1)).contains(RunStatus.COMPLETED.name());
+
assertThat(emitters.get(0).eventCount("\"type\":\"run_finished\"")).isEqualTo(1);
+ assertThat(registry.isLive(RUN_ID)).isFalse();
+ }
+
@Test
void
aStopDuringTheRunShouldWriteStoppedEvenWhenTheProviderReportsSuccessTest() {
// claude flushes its result frame on SIGTERM, so this is the normal
shape of a stop that lands
@@ -553,6 +590,12 @@ class AiRunExecutorTest {
// --- harness
---------------------------------------------------------------
+ private AiConversationProperties properties() {
+ AiConversationProperties properties = new AiConversationProperties();
+ properties.setStopGrace(Duration.ofMillis(1));
+ return properties;
+ }
+
private void startAndRun() {
startAndRun(false);
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunTestSupport.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunTestSupport.java
index 23860b958..412c33101 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunTestSupport.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunTestSupport.java
@@ -38,6 +38,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.LongFunction;
@@ -274,6 +275,13 @@ final class AiRunTestSupport {
final List<AgentEvent> scripted = new ArrayList<>();
Consumer<AgentStreamOptions> beforeStream = options -> {
};
+ /**
+ * Runs after each scripted frame reaches the sink, with the stream's
options in hand — the
+ * hook for a test whose stop must land <em>between</em> frames, which
{@code beforeStream}
+ * (it fires before the script) cannot express.
+ */
+ BiConsumer<AgentEvent, AgentStreamOptions> afterEvent = (event,
options) -> {
+ };
RuntimeException failure;
/**
* Thrown instead of any RuntimeException, so a test can reach the
paths that only an {@link Error}
@@ -313,7 +321,10 @@ final class AiRunTestSupport {
lastPrompt = options.getPrompt();
lastOptions = options;
beforeStream.accept(options);
- scripted.forEach(sink);
+ scripted.forEach(event -> {
+ sink.accept(event);
+ afterEvent.accept(event, options);
+ });
if (error != null) {
throw error;
}
