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 99ed333c4 fix(ai): re-read the run row before a stop writes the
terminal state (#6384)
99ed333c4 is described below
commit 99ed333c465cc1b2aa4edde3b4ce583d0d0a6898
Author: 0 <[email protected]>
AuthorDate: Thu Oct 15 18:09:03 2026 +0800
fix(ai): re-read the run row before a stop writes the terminal state (#6384)
* fix(ai): re-read the run row before a stop writes the terminal state
stop() read the run once at the top, and the registry lookup that came
back empty - the branch that means nobody owns the run - could not tell
an orphaned row from a worker that finalised between the read and the
lookup. Writing STOPPED on that stale read duplicated the run_status
event and flipped a run the user watched complete to STOPPED.
The row is re-read before the terminal write; an already-terminal run
makes the stop the no-op it always promised to be.
* fix(ai): let the stop path fall through to the shared terminal re-read
---
.../studio/ops/ai/conversation/AiRunService.java | 18 ++++++++++++++----
.../studio/ops/ai/conversation/AiRunServiceTest.java | 20 ++++++++++++++++++++
2 files changed, 34 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 63f346c59..5ac501dc0 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
@@ -402,11 +402,21 @@ public class AiRunService {
// worker is still flushing its buffered text.
log.info("agent run {} is already stopping", run.getId());
} else {
- // The row says active but nothing in this process owns it, so no
worker will ever write the
- // terminal state. Writing it here is what keeps the stop button
from spinning until the
- // orphan sweep gets around to it.
+ // The row said active but the worker may have finalised between
the read at the top of
+ // this method and the registry lookup, whose empty result is what
reached this branch.
+ // Re-read before writing anything: a terminal write over a
finished run would duplicate
+ // the run_status event and flip a run the user watched complete
to STOPPED.
+ RmqAiRun current = runRepository.findById(runId).orElse(run);
+ RunStatus currentStatus = statusOf(current.getStatus());
+ if (currentStatus == null || currentStatus.isTerminal()) {
+ log.debug("stop of run {} found it already terminal ({})",
run.getId(), current.getStatus());
+ return current;
+ }
+ // The row really is active with no owner in this process, so no
worker will ever write
+ // the terminal state. Writing it here is what keeps the stop
button from spinning until
+ // the orphan sweep gets around to it.
log.warn("run {} is non-terminal with no owner in this process;
stopping it directly", run.getId());
- runExecutor.terminate(detachedContext(run), RunStatus.STOPPED,
StopReason.USER_STOP, null, null);
+ runExecutor.terminate(detachedContext(current), RunStatus.STOPPED,
StopReason.USER_STOP, null, null);
}
return runRepository.findById(runId).orElse(run);
}
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 7a73a69ce..e2a84ce3e 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
@@ -458,6 +458,26 @@ class AiRunServiceTest {
assertThat(runUpdates).isEmpty();
}
+ @Test
+ void aStopThatLosesTheFinalisationRaceMustNotWriteASecondTerminalTest() {
+ // The run row is read once at the top of stop(); the worker can
finalise between that read
+ // and the registry lookup, whose empty result then looks like "nobody
owns this run".
+ // Writing the terminal state on that stale read would duplicate the
run_status event and
+ // flip a run the user watched complete to STOPPED, so the row is
re-read first.
+ RmqAiRun running = AiRunTestSupport.run(RUN_ID, CONVERSATION_ID, 1,
RunStatus.RUNNING);
+ RmqAiRun finished = AiRunTestSupport.run(RUN_ID, CONVERSATION_ID, 1,
RunStatus.COMPLETED);
+ when(runRepository.findById(RUN_ID))
+ .thenReturn(Optional.of(running))
+ .thenReturn(Optional.of(finished));
+
when(runRepository.findActiveByConversationId(CONVERSATION_ID)).thenReturn(Optional.of(running));
+
+ RmqAiRun stopped = service.stop(RUN_ID);
+
+ assertThat(stopped.getStatus()).isEqualTo(RunStatus.COMPLETED.name());
+ assertThat(runUpdates).isEmpty();
+ assertThat(inserted).isEmpty();
+ }
+
@Test
void stopOfARunNobodyOwnsAnyMoreShouldStillWriteTheTerminalStateTest() {
// The row says active, but nothing in this process owns it: the
worker is gone and will never
