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 085e1fedf fix(ai): stream the degraded notice only for agent-engine
turns (#5818)
085e1fedf is described below
commit 085e1fedfcf60ccbf285bf12e7143c307d0b228c
Author: 0 <[email protected]>
AuthorDate: Mon Oct 12 16:14:50 2026 +0800
fix(ai): stream the degraded notice only for agent-engine turns (#5818)
Two fixes to `AiRunService.sendMessage`, which decides both what a degraded
conversation tells the person watching it and whether an agent workspace is
prepared at all; folded, the two conditions collapse into one `degraded =
(preparation == null && !httpEngine) ? degradedNotice(conversation) : null`, so
an `http` turn neither persists nor streams a notice it cannot act on.
- #5818 `AiRunService.java:236-238` pushed the degraded/unbound notice
through `sink.writeTimeline(...)` only, while live frames reach an attached
observer exclusively through `registry.publish(...)`, so a client watching the
run over SSE saw nothing until it reloaded the conversation → the notice is
held in a local `TimelineEvent.Notice`, persisted exactly as before, and
re-published after the `RunStarted` frame at `:243` as
`registry.publish(run.getId(), 0L, new LiveEvent.Notice(deg [...]
- #5819 `AiRunService.java:209` prepared an agent workspace for every
engine, so a pure `http` completion turn — the engine
`LlmConfigService.java:346-352` also falls back to for any unknown value — was
failed with a 500 by a `RuntimeException` out of `prepare()` (`:210-220`) or
collected a "RocketMQ tools unavailable" notice on every turn →
`workspace.prepare(...)` is skipped when `AiRunExecutor.isHttpEngine(engine)`
(a package-private static at `AiRunExecutor.java:630-633`, reachabl [...]
Tests: `AiRunServiceTest` gains two cases — one counts `"type":"notice"`
frames on a live emitter and pins the count at exactly 1, so removing the
`registry.publish` call or duplicating it both fail, and one drives an `http`
turn whose `workspace.prepare` would throw, asserting `verify(workspace,
never()).prepare(...)`, a timeline with no notice row, and a run that does not
end FAILED.
Folded: #5819 was closed as folded into this commit (same author).
---
.../studio/ops/ai/conversation/AiRunService.java | 24 ++++++++++---
.../ops/ai/conversation/AiRunServiceTest.java | 39 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 4 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
index 1aec07b2e..63f346c59 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
@@ -203,10 +203,14 @@ public class AiRunService {
// Prepared at admission, not on the worker: an unusable workspace
configuration is an operator
// error the caller should hear about now, and a degraded conversation
must say so in its first
- // persisted event rather than in a log line nobody reads.
+ // persisted event rather than in a log line nobody reads. The http
engine is the exception: it
+ // has no agent CLI and never reads the workspace, so preparing one
would let a broken tools
+ // configuration fail - and a missing one warn on every turn of - a
plain completion.
+ boolean httpEngine = AiRunExecutor.isHttpEngine(engine);
RmqctlWorkspace.Preparation preparation;
try {
- preparation = workspace.prepare(conversation.getId(),
conversation.getInstanceId()).orElse(null);
+ preparation = httpEngine ? null
+ : workspace.prepare(conversation.getId(),
conversation.getInstanceId()).orElse(null);
} catch (RuntimeException exception) {
// The row is already inserted and would otherwise stay QUEUED
with no owner: every later
// message would be refused 409 until the orphan sweep reaps it.
Finalize it here — through
@@ -232,9 +236,13 @@ public class AiRunService {
// is persisted, and it reaches the timeline as a thinking block
with source=enhance — which
// is where it belongs, since merging it into the model's
reasoning is the bug this design
// exists to prevent.
+ // One verdict, used by both the persisted row and the live frame
below: an http-engine
+ // turn has no workspace to be degraded about, so it gets neither.
+ TimelineEvent.Notice degraded =
+ preparation == null && !httpEngine ?
degradedNotice(conversation) : null;
int admissionSeq = sink.writeUser(message, null);
- if (preparation == null) {
- admissionSeq =
sink.writeTimeline(degradedNotice(conversation));
+ if (degraded != null) {
+ admissionSeq = sink.writeTimeline(degraded);
}
applyTurnToConversation(conversation, message, request.mode());
session.noteWatermark(admissionSeq);
@@ -242,6 +250,14 @@ public class AiRunService {
session.finishReplay();
registry.publish(run.getId(), 0L, new
LiveEvent.RunStarted(run.getId(), conversation.getId(),
conversation.getTitle(), turn));
+ if (degraded != null) {
+ // The persisted notice has to reach the wire too: a client
that watches the turn
+ // live must see the same warning a reload shows, which is the
live/replay
+ // equivalence the projector promises. seq 0, because the row
already carried the
+ // notice and the observer noted that watermark above.
+ registry.publish(run.getId(), 0L,
+ new LiveEvent.Notice(degraded.level(),
degraded.message()));
+ }
runExecutor.submit(context);
} catch (RejectedExecutionException exception) {
log.warn("agent run {} was rejected: the run executor is
saturated", run.getId());
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 984273bab..7a73a69ce 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
@@ -49,6 +49,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -56,6 +57,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -215,6 +217,20 @@ class AiRunServiceTest {
assertThat(inserted.get(1).getSeq()).isEqualTo(2);
}
+ @Test
+ void aDegradedConversationShouldTellTheLiveStreamTooTest() {
+ when(workspace.prepare(anyLong(), any())).thenReturn(Optional.empty());
+
+ service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of("hello"));
+
+ // Live and replayed must render alike: the notice is persisted as the
run's first event, so
+ // the observer that is still attached has to hear it as well - not
only whoever reloads
+ // after the fact.
+ assertThat(AiRunTestSupport.typesOf(inserted)).startsWith("user",
"notice");
+
assertThat(emitters.get(0).eventCount("\"type\":\"notice\"")).isEqualTo(1);
+
assertThat(emitters.get(0).eventText()).contains(RmqctlWorkspace.DEGRADED_NOTICE_MESSAGE);
+ }
+
@Test
void
aConversationWithNoInstanceShouldSayItIsUnboundRatherThanDegradedTest() {
conversation.setInstanceId(null);
@@ -225,6 +241,29 @@ class AiRunServiceTest {
assertThat(inserted.get(1).getPayload()).contains(RmqctlWorkspace.UNBOUND_NOTICE_MESSAGE);
}
+ @Test
+ void anHttpTurnShouldNotPrepareAnAgentWorkspaceItCannotUseTest() {
+ when(llmClient.supports(any())).thenReturn(true);
+ doAnswer(invocation -> {
+ Consumer<String> tokens = invocation.getArgument(3);
+ tokens.accept("pong");
+ return null;
+ }).when(llmClient).stream(any(), any(), any(), any());
+ when(workspace.prepare(anyLong(), any()))
+ .thenThrow(new BusinessException(500, "the workspace directory
is unusable"));
+
+ service.sendMessage(CONVERSATION_ID, new AiRunService.RunRequest(
+ "hello", null, LlmConfigVO.ENGINE_HTTP, null, false, null));
+
+ // A completion turn has no agent CLI, so it never reads the
workspace: a broken tools
+ // configuration must not fail it, and an unbound conversation must
not be warned about
+ // tools this engine never had.
+ verify(workspace, never()).prepare(anyLong(), any());
+ assertThat(AiRunTestSupport.typesOf(inserted)).containsExactly("user",
"text", "run_status");
+ assertThat(runUpdates.get(runUpdates.size() - 1).getStatus())
+ .isEqualTo(RunStatus.COMPLETED.name());
+ }
+
@Test
void
theFirstTurnShouldNameTheConversationAndALaterOneShouldNotRenameItTest() {
service.sendMessage(CONVERSATION_ID, AiRunService.RunRequest.of("why
is my consumer lagging"));
