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 12b63f347 fix(ai): correct three run-lifecycle defects around stopping
and attaching (#5440)
12b63f347 is described below
commit 12b63f34778007f53f52d3efdb8edb95d33818c8
Author: 烤化の初雪 <[email protected]>
AuthorDate: Fri Oct 9 16:28:21 2026 +0800
fix(ai): correct three run-lifecycle defects around stopping and attaching
(#5440)
Three defects in the AI run lifecycle, plus the test that was racing them.
- The composer lost a draft when Enter arrived while a run was stopping:
`canSend` keyed off
`!generating`, and `generating` is already false once a stop was
requested, so the message went into
a run that was ending. (#5425)
- `AiRunService.stop` read the single boolean from `registry.stop()` as
"this process does not own the
run" and then wrote a terminal state for a run that was still live.
(#5440)
- `useActiveRunAttach` keyed its anti-hot-loop ref on the run id alone, so
a run that left the
admission window and came back with the same id attached to a cleanly
finished empty stream instead
of an error state. The ref now carries id + status. (#5446)
-
`AiRunServiceTest.stopShouldAbortALiveRunAndLeaveItStoppedWithUserStopTest`
asserted without
waiting for the worker, which made it flaky - this is the failure that
reddened an unrelated PR's
`Backend Tests` run today. (#5450)
`AiRunServiceTest` 23 tests green with the dev profile, 0 checkstyle
violations;
`Composer.test.tsx` + `useActiveRunAttach.test.ts` 21 tests green; prettier
clean.
Folded in #5425, #5446 and #5450.
---
.../studio/ops/ai/conversation/AiRunService.java | 7 ++++++
.../ops/ai/conversation/AiRunServiceTest.java | 28 +++++++++++++++++++---
web/src/pages/ai/components/Composer.tsx | 10 +++++---
.../ai/components/__tests__/Composer.test.tsx | 17 +++++++++++++
web/src/pages/ai/hooks/useActiveRunAttach.test.ts | 27 +++++++++++++++++++++
web/src/pages/ai/hooks/useActiveRunAttach.ts | 25 +++++++++++++++----
6 files changed, 103 insertions(+), 11 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 3b5a83589..1aec07b2e 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
@@ -378,6 +378,13 @@ public class AiRunService {
}
if (registry.stop(run.getId(), AbortReason.USER_STOP)) {
log.info("agent run {} is being stopped at the user's request",
run.getId());
+ } else if (registry.handle(run.getId()).isPresent()) {
+ // This process owns the run and it is already aborting (a second
tab, a retried stop):
+ // registry.stop() returns false for that case as well as for an
unowned run, and its
+ // worker writes the terminal state itself. Writing one here would
append a second
+ // terminal row - a phantom empty bubble on reload - and close the
observers while the
+ // 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
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 0629713c2..984273bab 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
@@ -320,12 +320,14 @@ class AiRunServiceTest {
service.sendMessage(CONVERSATION_ID, AiRunService.RunRequest.of("a
long answer"));
assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(registry.isLive(RUN_ID)).isTrue();
+ // Capture the handle before stopping: stop() cancels the worker,
and the worker's own
+ // finalisation removes the registration, so looking it up again
afterwards races with
+ // that cleanup and is legitimately empty on a loaded machine.
+ AgentRunHandle handle = registry.handle(RUN_ID).orElseThrow();
RmqAiRun stopped = service.stop(RUN_ID);
- assertThat(registry.handle(RUN_ID))
- .get()
- .satisfies(handle ->
assertThat(handle.abortReason()).contains(AbortReason.USER_STOP));
+ assertThat(handle.abortReason()).contains(AbortReason.USER_STOP);
release.countDown();
assertThat(stopped.getId()).isEqualTo(RUN_ID);
assertThat(awaitRunStatus(RunStatus.STOPPED,
Duration.ofSeconds(5))).isTrue();
@@ -350,6 +352,26 @@ class AiRunServiceTest {
assertThat(inserted).isEmpty();
}
+ @Test
+ void secondStopOfAnAbortingRunShouldNotWriteItsOwnTerminalStateTest() {
+ RmqAiRun running = AiRunTestSupport.run(RUN_ID, CONVERSATION_ID, 1,
RunStatus.RUNNING);
+ when(runRepository.findById(RUN_ID)).thenReturn(Optional.of(running));
+ when(runRepository.findActiveByConversationId(CONVERSATION_ID))
+ .thenReturn(Optional.of(running));
+ // A worker in this process owns the run and is already aborting,
which is exactly what
+ // registry.stop() reports as false - the same value it returns for a
run nobody owns here.
+ AgentRunHandle handle = new AgentRunHandle(RUN_ID,
Duration.ofSeconds(3));
+ registry.register(RUN_ID, handle);
+ assertThat(handle.requestStop(AbortReason.USER_STOP)).isTrue();
+
+ service.stop(RUN_ID);
+
+ // The worker writes the terminal state itself; a fabricated one here
would append a second
+ // terminal row and close the observers mid-flush.
+ assertThat(runUpdates).isEmpty();
+ assertThat(inserted).isEmpty();
+ }
+
@Test
void reportSpeedShouldTouchOnlyTheSpeedColumnTest() {
// The client reports the speed as its stream closes, which also
happens mid-run when the
diff --git a/web/src/pages/ai/components/Composer.tsx
b/web/src/pages/ai/components/Composer.tsx
index 3ea68e9bf..4d3637134 100644
--- a/web/src/pages/ai/components/Composer.tsx
+++ b/web/src/pages/ai/components/Composer.tsx
@@ -59,12 +59,13 @@ import SendStopButton, { type SendStopState } from
'./SendStopButton';
* ```ts
* const generating = isStreaming && !stopRequested;
* state = stopRequested ? 'stopping' : generating ? 'stop' : 'send';
- * canSend = value.trim() && !disabled && !readOnlyRole && llmReady &&
!generating;
+ * canSend = value.trim() && !disabled && !readOnlyRole && llmReady &&
!isStreaming;
* ```
*
* `stopRequested` is what separates "the stop is on its way" from "the run is
still going": the stop
* is a server round-trip and the stream stays open to deliver the terminal
frames, so the button sits
- * in `stopping` until they arrive rather than flipping straight back to
`send`.
+ * in `stopping` until they arrive rather than flipping straight back to
`send`. `canSend` follows the
+ * whole streaming window, so the keyboard cannot submit into that same window.
*
* ─── Keyboard ──────────────────────────────────────────────────
* Enter sends, Shift+Enter inserts a newline, and Escape stops while
generating. The `isComposing`
@@ -247,7 +248,10 @@ const Composer = ({
const generating = isStreaming && !stopRequested;
const sendStopState: SendStopState = stopRequested ? 'stopping' : generating
? 'stop' : 'send';
- const canSend = !disabled && !readOnlyRole && llmReady && !generating &&
value.trim().length > 0;
+ // `isStreaming`, not `generating`: while the stop is still in flight the
button is in its
+ // disabled `stopping` state because the stream guard would refuse a new
run, and Enter must not
+ // be a hidden second way in - it used to clear the draft for a send that
silently never left.
+ const canSend = !disabled && !readOnlyRole && llmReady && !isStreaming &&
value.trim().length > 0;
const contextPercent = Math.min(100, Math.round((contextTokens /
CONTEXT_WINDOW_TOKENS) * 100));
const contextColor =
diff --git a/web/src/pages/ai/components/__tests__/Composer.test.tsx
b/web/src/pages/ai/components/__tests__/Composer.test.tsx
index cc8dce226..3c170d7b5 100644
--- a/web/src/pages/ai/components/__tests__/Composer.test.tsx
+++ b/web/src/pages/ai/components/__tests__/Composer.test.tsx
@@ -167,6 +167,23 @@ describe('Composer', () => {
expect(props.onSend).not.toHaveBeenCalled();
});
+ it('keepsTheDraftWhenEnterArrivesWhileTheStopIsStillInFlightTest', () => {
+ const { props } = renderComposer({
+ value: '检查集群状态',
+ isStreaming: true,
+ stopRequested: true,
+ });
+
+ // The run stays in flight until the terminal frames land, so a send
started now would be
+ // refused by the stream guard while the composer had already cleared the
box. Enter must not
+ // be a hidden second way to submit when the button next to it is disabled.
+ fireEvent.keyDown(textarea(), { key: 'Enter' });
+
+ expect(props.onSend).not.toHaveBeenCalled();
+ expect(props.onChange).not.toHaveBeenCalled();
+ expect(textarea()).toHaveValue('检查集群状态');
+ });
+
it('sendsOnEnterAndClearsTheDraftTest', () => {
const { props } = renderComposer({ value: '检查集群状态' });
diff --git a/web/src/pages/ai/hooks/useActiveRunAttach.test.ts
b/web/src/pages/ai/hooks/useActiveRunAttach.test.ts
index 6e4298831..458948389 100644
--- a/web/src/pages/ai/hooks/useActiveRunAttach.test.ts
+++ b/web/src/pages/ai/hooks/useActiveRunAttach.test.ts
@@ -74,4 +74,31 @@ describe('useActiveRunAttach', () => {
await waitFor(() => expect(attach).toHaveBeenCalledWith(9, 77, 0));
expect(attach).toHaveBeenCalledTimes(2);
});
+
+ it('reattachesWhenTheRunAdvancesPastTheAdmissionWindowItWasAttachedInTest',
async () => {
+ const attach = vi.fn().mockResolvedValue(undefined);
+ const { rerender } = renderHook(
+ ({ status }: { status: 'QUEUED' | 'RUNNING' }) =>
+ useActiveRunAttach(
+ 7,
+ { activeRun: { id: 42, status }, lastSeq: 0 },
+ { isStreaming: false, attach },
+ ),
+ { initialProps: { status: 'QUEUED' as 'QUEUED' | 'RUNNING' } },
+ );
+
+ // A reload during admission attaches to a QUEUED row that has no worker
yet: the server closes
+ // that stream with a bare `done` and no status, so nothing was observed.
+ await waitFor(() => expect(attach).toHaveBeenCalledWith(7, 42, 0));
+
+ // The run now has a worker. Without re-attaching, the answer streams with
nobody watching and
+ // the next send is refused with 409.
+ rerender({ status: 'RUNNING' });
+ await waitFor(() => expect(attach).toHaveBeenCalledTimes(2));
+
+ // The same state again is not new information, so the guard still holds
and nothing loops.
+ rerender({ status: 'RUNNING' });
+ await new Promise((resolve) => setTimeout(resolve, 20));
+ expect(attach).toHaveBeenCalledTimes(2);
+ });
});
diff --git a/web/src/pages/ai/hooks/useActiveRunAttach.ts
b/web/src/pages/ai/hooks/useActiveRunAttach.ts
index ce4f51764..ab46c8db6 100644
--- a/web/src/pages/ai/hooks/useActiveRunAttach.ts
+++ b/web/src/pages/ai/hooks/useActiveRunAttach.ts
@@ -27,11 +27,14 @@ import type { UseConversationTimelineResult } from
'./useConversationTimeline';
* that row into exactly one `attach(conversationId, runId, lastSeq)` per run
id per visit — `lastSeq`
* being the replay cursor that keeps the attach from rendering persisted rows
twice.
*
- * ─── One attempt per run id per visit ───────────────────────────
+ * ─── One attempt per run state per visit ────────────────────────
* `requestedRunRef` is what stands between a dead attach (idle timeout,
server restart mid-run) and
* a hot re-attach loop: the failure surfaces through the run's error state
and a reload retries.
* The ref is cleared when the conversation changes, so navigating away
mid-run and BACK re-attaches
- * to the same run id — which is the whole point of the exercise.
+ * to the same run id — which is the whole point of the exercise. It also lets
a run that has moved
+ * on from the state this visit attached to be attached once more: a reload
during admission observes
+ * QUEUED, whose stream closes immediately because no worker exists yet, and
the run only becomes
+ * observable when the timeline reports it RUNNING.
*
* The reset effect is declared before the attach effect on purpose: within
one commit effects run
* in declaration order, and arriving at a conversation must clear the
previous visit's marker
@@ -42,7 +45,12 @@ export function useActiveRunAttach(
timeline: Pick<UseConversationTimelineResult, 'activeRun' | 'lastSeq'>,
run: Pick<UseAgentRunResult, 'isStreaming' | 'attach'>,
): void {
- const requestedRunRef = useRef<number | null>(null);
+ // What the last attach observed, not just which run: a reload during
admission attaches to a
+ // QUEUED row that has no worker yet, the server closes that stream with a
bare `done`, and the run
+ // then starts generating with nobody watching. Remembering the status too
lets the next timeline
+ // fetch - which reports the same run as RUNNING - attach again, while a
repeated identical status
+ // still cannot loop.
+ const requestedRunRef = useRef<{ id: number; status: string } | null>(null);
const { activeRun, lastSeq } = timeline;
const { isStreaming, attach } = run;
@@ -53,8 +61,15 @@ export function useActiveRunAttach(
useEffect(() => {
if (conversationId === null || activeRun === null || isStreaming) return;
if (activeRun.status !== 'QUEUED' && activeRun.status !== 'RUNNING')
return;
- if (requestedRunRef.current === activeRun.id) return;
- requestedRunRef.current = activeRun.id;
+ const requested = requestedRunRef.current;
+ if (
+ requested !== null &&
+ requested.id === activeRun.id &&
+ requested.status === activeRun.status
+ ) {
+ return;
+ }
+ requestedRunRef.current = { id: activeRun.id, status: activeRun.status };
void attach(conversationId, activeRun.id, lastSeq);
}, [activeRun, attach, conversationId, isStreaming, lastSeq]);
}
