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 e2f6808cd fix(ai): report premature run-stream EOF (#5187)
e2f6808cd is described below
commit e2f6808cdedf5d5aa7f10c41c6cdadd3b0d0ab7d
Author: zmuxuny <[email protected]>
AuthorDate: Fri Oct 9 23:39:28 2026 -0700
fix(ai): report premature run-stream EOF (#5187)
The server can end a run stream through `detach()` / `completeWithError`
without sending a `done`
frame, and the frontend resolved the reader silently at EOF, so a truncated
run looked like a normal
completion. EOF now throws on the single shared read path.
`ai.test.ts` 34 tests green.
---
web/src/api/ai.test.ts | 65 ++++++++++++++++++++++++++++++++++++++++++++++++++
web/src/api/ai.ts | 7 ++++++
2 files changed, 72 insertions(+)
diff --git a/web/src/api/ai.test.ts b/web/src/api/ai.test.ts
index 6029a35d4..2de42feee 100644
--- a/web/src/api/ai.test.ts
+++ b/web/src/api/ai.test.ts
@@ -174,6 +174,71 @@ describe('AI API', () => {
expect(events[4]).toEqual({ type: 'text_delta', content: '集群当前有 2 个
broker。' });
});
+ it.each(['send', 'attach'] as const)(
+ 'rejectsPrematureEofAfterPartialAgentOutputFor%sTest',
+ async (operation) => {
+ vi.stubGlobal(
+ 'fetch',
+ vi
+ .fn()
+ .mockResolvedValue(
+ eventStreamResponse([agentFrame({ type: 'text_delta', content:
'partial' })]),
+ ),
+ );
+ const onEvent = vi.fn();
+ const stream =
+ operation === 'send'
+ ? openRunStream(7, body, { onEvent })
+ : attachRunStream(41, 0, { onEvent });
+
+ await expect(stream).rejects.toMatchObject({
+ name: 'AiStreamError',
+ code: 'llm.stream.unexpected_eof',
+ });
+ expect(onEvent).toHaveBeenCalledExactlyOnceWith({
+ type: 'text_delta',
+ content: 'partial',
+ });
+ },
+ );
+
+ it.each([
+ ['empty', ''],
+ ['heartbeat-only', ':hb\n\n'],
+ ['unterminated-agent', 'event: agent\ndata:
{"type":"text_delta","content":"partial"}'],
+ ])('rejectsPrematureEofFor%sResponseTest', async (_name, content) => {
+ vi.stubGlobal('fetch',
vi.fn().mockResolvedValue(eventStreamResponse([content])));
+
+ await expect(attachRunStream(41, 0, { onEvent: vi.fn()
})).rejects.toMatchObject({
+ name: 'AiStreamError',
+ code: 'llm.stream.unexpected_eof',
+ });
+ });
+
+ it('acceptsATerminalControlFrameWithoutATrailingBlankLineTest', async ()
=> {
+ vi.stubGlobal(
+ 'fetch',
+ vi.fn().mockResolvedValue(eventStreamResponse(['event: done\ndata:
{}'])),
+ );
+
+ await expect(attachRunStream(41, 0, { onEvent: vi.fn()
})).resolves.toBeUndefined();
+ });
+
+ it('preservesReaderAbortErrorsInsteadOfReplacingThemWithPrematureEofTest',
async () => {
+ const abort = new DOMException('The operation was aborted.',
'AbortError');
+ const response = new Response(
+ new ReadableStream<Uint8Array>({
+ start(controller) {
+ controller.error(abort);
+ },
+ }),
+ { headers: { 'content-type': 'text/event-stream' } },
+ );
+ vi.stubGlobal('fetch', vi.fn().mockResolvedValue(response));
+
+ await expect(attachRunStream(41, 0, { onEvent: vi.fn()
})).rejects.toBe(abort);
+ });
+
it('terminatesOnTheDoneEventNameAndCancelsTheReaderTest', async () => {
const onCancel = vi.fn();
vi.stubGlobal(
diff --git a/web/src/api/ai.ts b/web/src/api/ai.ts
index 1b7d2aade..dd8df34f4 100644
--- a/web/src/api/ai.ts
+++ b/web/src/api/ai.ts
@@ -285,6 +285,13 @@ async function consumeEventStream(
// A server that closes without a trailing blank line still delivered its
last frame.
buffer += decoder.decode();
if (buffer && dispatch(buffer)) return;
+ // A run outlives its HTTP observer. Only the explicit terminal control
frame
+ // establishes normal stream completion; EOF can leave the server run
active.
+ throw new AiStreamError(
+ 'AI stream ended before the done event',
+ 'llm.stream.unexpected_eof',
+ 'Reload the conversation to recover the persisted run and reconnect if
it is still active.',
+ );
} finally {
await reader.cancel().catch(() => undefined);
}
