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 4c546ad6c fix(ai): fail a run whose agent stream ends without a result
frame (#6391)
4c546ad6c is described below
commit 4c546ad6cf7bba8c13c8d0de594bd463dddd6917
Author: 0 <[email protected]>
AuthorDate: Thu Oct 15 20:20:15 2026 +0800
fix(ai): fail a run whose agent stream ends without a result frame (#6391)
* fix(ai): fail a run whose agent stream ends without a result frame
* test(ai): close the flush-failure turn with a success frame
aDatabaseFailureInsideAFlushShouldNotKillTheRunTest emitted only a
TextDelta and then asserted COMPLETED. Under the new no-result branch in
decide() that assertion flips to FAILED, so the case stopped proving what
it was written for: a database failure inside a flush must not kill the
run. Emit a ResultMeta success frame, the same way the recovered-run case
in AiRunServiceTest now does, and keep the COMPLETED assertion. The run
under test is a healthy turn whose event writes fail, not a stream that
ended with no result frame.
---------
Co-authored-by: lizhimins <[email protected]>
---
.../studio/ops/ai/conversation/AiRunExecutor.java | 10 ++++++++
.../ops/ai/conversation/AiRunExecutorTest.java | 5 ++++
.../ops/ai/conversation/AiRunServiceTest.java | 28 ++++++++++++++++++++--
3 files changed, 41 insertions(+), 2 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 db71840c7..400c703b3 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
@@ -126,6 +126,8 @@ public class AiRunExecutor {
static final String ERROR_CODE_INTERNAL = "ai.run.internal_error";
private static final String PROVIDER_TIMEOUT_CODE = "llm.provider.timeout";
+ /** The agent stream closed without ever delivering a result frame. */
+ static final String ERROR_CODE_NO_RESULT = "llm.provider.no_result";
private static final Set<String> OUTPUT_LIMIT_CODES =
Set.of("llm.provider.output_too_large",
"llm.provider.response_too_large");
private static final String ERROR_MAX_TURNS_SUBTYPE = "error_max_turns";
@@ -503,6 +505,14 @@ public class AiRunExecutor {
"llm.provider." + outcome.subtype,
"agent run ended with subtype: " + outcome.subtype,
outcome, false);
}
+ if (!isHttpEngine(context.getEngine()) &&
!StringUtils.hasText(outcome.subtype)) {
+ // The agent stream closed without ever delivering a result frame:
nothing was projected,
+ // nothing failed with a frame of its own — the provider process
simply went away mid-turn.
+ // A turn with no answer is not a successful one. HTTP completion
turns never carry a result
+ // frame at all, so their fall through to COMPLETED stays correct
— that is the exclusion here.
+ return new Terminal(RunStatus.FAILED, StopReason.PROVIDER_ERROR,
ERROR_CODE_NO_RESULT,
+ "agent stream ended without a result frame", outcome,
false);
+ }
return new Terminal(RunStatus.COMPLETED, null, null, null, outcome,
false);
}
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 ecda29e5d..c423a7d51 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
@@ -454,6 +454,11 @@ class AiRunExecutorTest {
doThrow(new RuntimeException("deadlock found when trying to get lock"))
.when(eventRepository).insert(any(RmqAiEvent.class));
provider.emit(new AgentEvent.TextDelta("answer"));
+ // Close the turn with a success frame. This case is about a failing
flush, not about a stream
+ // that ends with no result frame at all — decide() now fails the
latter on purpose, so without
+ // this frame the COMPLETED assertion below would be testing the wrong
branch.
+ provider.emit(new AgentEvent.ResultMeta("session-9", 1200L, 30, 90,
+ AgentEventProjector.SUCCESS_SUBTYPE));
startAndRun();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
index e2a84ce3e..33aa3cd79 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
@@ -25,6 +25,8 @@ import org.apache.rocketmq.studio.ops.ai.LlmConfigVO;
import org.apache.rocketmq.studio.ops.ai.OpenAiCompatibleLlmClient;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.PromptEnhancer;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.RmqctlWorkspace;
+import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEvent;
+import
org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEventProjector;
import org.apache.rocketmq.studio.ops.ai.conversation.event.LiveEvent;
import org.apache.rocketmq.studio.ops.ai.conversation.event.RunStatus;
import org.apache.rocketmq.studio.ops.ai.conversation.event.StopReason;
@@ -170,8 +172,11 @@ class AiRunServiceTest {
// The worker picked the run up: QUEUED is a real state, not a label
the row keeps.
assertThat(runUpdates.get(0).getStatus()).isEqualTo(RunStatus.RUNNING.name());
assertThat(runUpdates.get(0).getStartedAt()).isNotNull();
- // ...and the run really did go on to execute and finish on the queued
engine.
-
assertThat(lastRun().getStatus()).isEqualTo(RunStatus.COMPLETED.name());
+ // ...and the run really did go on to execute and reach a terminal on
the queued engine —
+ // a terminal, not specifically a success: this stub streams no
frames, so the turn ends
+ // with the no-result failure instead of a success nothing projected.
+ assertThat(lastRun().getStatus()).isEqualTo(RunStatus.FAILED.name());
+
assertThat(lastRun().getStopReason()).isEqualTo(StopReason.PROVIDER_ERROR.name());
assertThat(emitters.get(0).timeoutMillis()).isEqualTo(AiRunExecutor.CLI_STREAM_TIMEOUT_MILLIS);
}
@@ -326,11 +331,30 @@ class AiRunServiceTest {
// 409 the stranded-QUEUED row produced for up to a day.
org.mockito.Mockito.reset(workspace);
when(workspace.prepare(anyLong(),
any())).thenReturn(Optional.of(preparation()));
+ // Script a success frame so the recovered run completes: an
unscripted stub stream now
+ // ends with the no-result failure, and this turn is meant to be a
healthy one.
+ provider.emit(new AgentEvent.ResultMeta("session-2", 10L, 3, 4,
+ AgentEventProjector.SUCCESS_SUBTYPE));
service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of("second try"));
assertThat(runInserts).hasSize(2);
assertThat(lastRun().getStatus()).isEqualTo(RunStatus.COMPLETED.name());
}
+ @Test
+ void aStreamThatEndsWithoutAResultFrameMustNotCompleteTheRunTest() {
+ // The stub streams no frames at all: the agent stream closes without
a result frame, so
+ // nothing was projected and nothing errored with a frame of its own —
the provider process
+ // simply went away mid-turn. Reporting COMPLETED would present a turn
with no answer as a
+ // successful one. HTTP completion turns never carry a result frame
and are deliberately
+ // excluded from this in decide().
+ service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of("anybody home"));
+
+ assertThat(lastRun().getStatus()).isEqualTo(RunStatus.FAILED.name());
+
assertThat(lastRun().getStopReason()).isEqualTo(StopReason.PROVIDER_ERROR.name());
+
assertThat(lastRun().getErrorCode()).isEqualTo(AiRunExecutor.ERROR_CODE_NO_RESULT);
+ assertThat(registry.isLive(RUN_ID)).isFalse();
+ }
+
@Test
void aBlankOrOversizedMessageShouldBeRefusedBeforeARowIsWrittenTest() {
assertThatThrownBy(() -> service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of(" ")))
