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 c9b61540d fix(ai): fail the run row when workspace preparation throws
(#4796)
c9b61540d is described below
commit c9b61540d28362245641ed8b90bcae509b07310e
Author: 烤化の初雪 <[email protected]>
AuthorDate: Thu Sep 24 18:17:02 2026 +0800
fix(ai): fail the run row when workspace preparation throws (#4796)
test(ai): pin that the conversation is reusable right after a failed
preparation
The recovery half of the contract: once the failed row is terminal, a
second send admits a new run and completes - it must not hit the 409
the stranded-QUEUED row produced for up to a day.
fix(ai): fail the run row when workspace preparation throws
sendMessage inserts the run row before preparing the agent workspace,
and prepare() deliberately throws BusinessException for configuration
problems (a bound instance whose credential is gone, an unusable
rmqctl-server-url or workspace-dir). The exception left the inserted
row QUEUED with no owner: every later message for the conversation was
refused 409 until the scheduled orphan sweep reaped it, up to a day
later. Route the failure through the exactly-once terminal path via
detachedContext - the same mechanism the stop of an owner-less run
uses - and rethrow so the caller still hears the reason.
---
.../studio/ops/ai/conversation/AiRunService.java | 16 +++++++++--
.../ops/ai/conversation/AiRunServiceTest.java | 31 ++++++++++++++++++++++
2 files changed, 45 insertions(+), 2 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 0a5166b06..7993ae50f 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
@@ -204,8 +204,20 @@ 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.
- RmqctlWorkspace.Preparation preparation =
- workspace.prepare(conversation.getId(),
conversation.getInstanceId()).orElse(null);
+ RmqctlWorkspace.Preparation preparation;
+ try {
+ preparation = 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
+ // the exactly-once terminal path, like the stop of an owner-less
run — and rethrow so the
+ // caller still hears the reason.
+ log.error("could not prepare the agent workspace for run {};
failing the run", run.getId(),
+ exception);
+ runExecutor.terminate(detachedContext(run), RunStatus.FAILED,
StopReason.PROVIDER_ERROR,
+ AiRunExecutor.ERROR_CODE_INTERNAL, exception.toString());
+ throw exception;
+ }
long timeoutMillis = runExecutor.streamTimeoutMillis(engine);
AgentRunHandle handle = runExecutor.newHandle(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 215ba468a..51a8a1a76 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
@@ -261,6 +261,37 @@ class AiRunServiceTest {
assertThat(registry.isLive(RUN_ID)).isFalse();
}
+ @Test
+ void
aFailingWorkspacePreparationShouldFinalizeTheRunInsteadOfStrandingItTest() {
+ // prepare() throws for configuration problems (a bound instance whose
credential is gone,
+ // an unusable rmqctl-server-url). The run row is already inserted by
then; if the exception
+ // simply propagates, the row stays QUEUED, every later message is
refused 409 "busy", and
+ // only the scheduled orphan sweep (up to a day) reaps it. The row
must reach a terminal
+ // state now, and the caller must still hear the reason.
+ when(workspace.prepare(anyLong(), any()))
+ .thenThrow(new BusinessException(404, "Instance not found:
localtest"));
+
+ assertThatThrownBy(() -> service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of("hello")))
+ .isInstanceOfSatisfying(BusinessException.class,
+ exception ->
assertThat(exception.getCode()).isEqualTo(404));
+
+ // The stranded-QUEUED regression: exactly one insert, updated
straight to a terminal row.
+ assertThat(runInserts).hasSize(1);
+
assertThat(runInserts.get(0).getStatus()).isEqualTo(RunStatus.QUEUED.name());
+ assertThat(lastRun().getStatus()).isEqualTo(RunStatus.FAILED.name());
+
assertThat(lastRun().getStopReason()).isEqualTo(StopReason.PROVIDER_ERROR.name());
+ assertThat(registry.isLive(RUN_ID)).isFalse();
+
+ // Recovery half of the contract: the failed row is terminal, so the
conversation is
+ // immediately usable again — a second message admits a new run
instead of hitting the
+ // 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()));
+ service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of("second try"));
+ assertThat(runInserts).hasSize(2);
+
assertThat(lastRun().getStatus()).isEqualTo(RunStatus.COMPLETED.name());
+ }
+
@Test
void aBlankOrOversizedMessageShouldBeRefusedBeforeARowIsWrittenTest() {
assertThatThrownBy(() -> service.sendMessage(CONVERSATION_ID,
AiRunService.RunRequest.of(" ")))
