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 1053cd4be fix(ai): harden LLM streaming, endpoint resolution and agent
runtime (#6998)
1053cd4be is described below
commit 1053cd4be782210833c35c914d136de7a29c95c6
Author: Wang1rrr <[email protected]>
AuthorDate: Tue Oct 27 12:04:27 2026 +0800
fix(ai): harden LLM streaming, endpoint resolution and agent runtime (#6998)
* fix(ai): preserve query parameters when resolving API endpoints (#6998)
Both apiBase normalizations rebuilt the endpoint by string concatenation,
so a
configured base carrying a query string or a fragment produced a malformed
URL.
LlmApiBase now resolves the base through URI components and inserts the path
ahead of the query and fragment; LlmConfigService and
OpenAiCompatibleLlmClient
share it instead of keeping two private copies.
* fix(ai): share the stream timeout budget with header reads (#6983)
The SSE reader task was granted a fresh requestTimeout even though waiting
for
the response headers had already consumed part of that same budget, so a
slow
provider could hold one call for nearly twice the configured timeout. The
reader now gets the remaining budget.
* fix(ai): preserve the first event in BOM-prefixed SSE streams (#6994)
SSE permits a single UTF-8 BOM at the start of the stream, before any field
name. Strip one leading U+FEFF from the first line only and leave interior
BOMs alone, so the first event is no longer misparsed.
* fix(ai): cancel rejected stream response bodies (#6967)
assertEventStream rethrew without releasing the response body, leaking the
connection on every rejected stream. Cancel the body first, then rethrow the
original error.
* fix(ai): bound partial SSE frames after draining complete events (#6976)
The frame size limit was applied right after the read and before complete
events were drained, so a large but well-formed chunk was reported as an
oversized remainder. Only the unfinished trailing frame is bounded now.
* fix(ai): serialize capability probe cache refreshes (#6971)
Concurrent callers each started their own agent capability probe and all of
them waited the full 15s. binaries() now refreshes under a lock and reuses
the
result of the probe already in flight.
* fix(ai): resolve workspace paths before changing child cwd (#6980)
A relative configured path was resolved against the child process working
directory, which the caller had already changed. workspaceRoot() now makes
the
path absolute and normalized before it is handed down.
* fix(ai): preserve Unicode characters in terminal run errors (#7002)
Truncating the terminal error fields by char could split a surrogate pair
and
leave the run unpersistable. Use TextBounds.truncate, which respects code
point
boundaries.
Batch integration of eight related fixes by the same contributor; #6967,
#6971, #6976, #6980, #6983, #6994 and #7002 are closed as merged into this
one.
---
.../apache/rocketmq/studio/ops/ai/LlmApiBase.java | 57 +++++++++++++
.../rocketmq/studio/ops/ai/LlmConfigService.java | 14 +---
.../studio/ops/ai/OpenAiCompatibleLlmClient.java | 37 ++++-----
.../studio/ops/ai/conversation/AiRunExecutor.java | 4 +-
.../conversation/agent/AgentCapabilityProbe.java | 4 +-
.../ops/ai/conversation/agent/RmqctlWorkspace.java | 3 +-
.../studio/ops/ai/LlmConfigServiceTest.java | 20 +++++
.../ops/ai/OpenAiCompatibleLlmClientTest.java | 77 +++++++++++++++++
.../ops/ai/conversation/AiRunExecutorTest.java | 31 +++++++
.../agent/AgentCapabilityProbeConcurrencyTest.java | 96 ++++++++++++++++++++++
.../ai/conversation/agent/RmqctlWorkspaceTest.java | 20 +++++
web/src/api/ai.test.ts | 76 +++++++++++++++++
web/src/api/ai.ts | 14 +++-
13 files changed, 412 insertions(+), 41 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmApiBase.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmApiBase.java
new file mode 100644
index 000000000..12e5568e1
--- /dev/null
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmApiBase.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.ops.ai;
+
+import java.net.URI;
+
+/** Normalizes endpoint paths without changing raw query parameters or
fragments. */
+final class LlmApiBase {
+ private static final String CHAT_COMPLETIONS_PATH = "/chat/completions";
+
+ private LlmApiBase() {
+ }
+
+ static String normalize(String apiBase) {
+ String trimmed = apiBase.trim();
+ int suffixStart = suffixStart(trimmed);
+ String base = trimTrailingSlashes(trimmed.substring(0, suffixStart));
+ if (base.endsWith(CHAT_COMPLETIONS_PATH)) {
+ base = trimTrailingSlashes(base.substring(0, base.length() -
CHAT_COMPLETIONS_PATH.length()));
+ }
+ return base + trimmed.substring(suffixStart);
+ }
+
+ static URI endpoint(String apiBase, String path) {
+ String normalized = normalize(apiBase);
+ int suffixStart = suffixStart(normalized);
+ return URI.create(normalized.substring(0, suffixStart) + path +
normalized.substring(suffixStart));
+ }
+
+ private static int suffixStart(String value) {
+ int query = value.indexOf('?');
+ int fragment = value.indexOf('#');
+ return Math.min(query < 0 ? value.length() : query, fragment < 0 ?
value.length() : fragment);
+ }
+
+ private static String trimTrailingSlashes(String value) {
+ int end = value.length();
+ while (end > 0 && value.charAt(end - 1) == '/') {
+ end--;
+ }
+ return value.substring(0, end);
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
index feabb0dcd..7e1ecd4cd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
@@ -359,17 +359,7 @@ public class LlmConfigService {
}
private String normalizeApiBase(String apiBase) {
- String normalized = apiBase.trim();
- while (normalized.endsWith("/")) {
- normalized = normalized.substring(0, normalized.length() - 1);
- }
- if (normalized.endsWith(CHAT_COMPLETIONS_PATH)) {
- normalized = normalized.substring(0, normalized.length() -
CHAT_COMPLETIONS_PATH.length());
- while (normalized.endsWith("/")) {
- normalized = normalized.substring(0, normalized.length() - 1);
- }
- }
- return normalized;
+ return LlmApiBase.normalize(apiBase);
}
private boolean isValidApiBase(String apiBase) {
@@ -377,7 +367,7 @@ public class LlmConfigService {
URI uri = new URI(apiBase);
String scheme = uri.getScheme() == null ? "" :
uri.getScheme().toLowerCase(Locale.ROOT);
if (!("http".equals(scheme) || "https".equals(scheme)) ||
!StringUtils.hasText(uri.getHost())
- || apiBase.endsWith(CHAT_COMPLETIONS_PATH)) {
+ || uri.getRawPath().endsWith(CHAT_COMPLETIONS_PATH)) {
return false;
}
// SSRF guard on both the save and test paths (validate() runs for
each). Loopback is
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
index 42173aa6a..762a33fe6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
@@ -180,12 +180,13 @@ public class OpenAiCompatibleLlmClient {
messages.add(Map.of("role", "user", "content",
StringUtils.hasText(prompt) ? prompt.trim() : ""));
requestBody.put("messages", messages);
HttpRequest request = request(config, "text/event-stream",
requestBody);
+ long startedAtNanos = System.nanoTime();
try {
HttpResponse<StreamBody> response = httpClient.send(request,
streamBodyHandler());
if (response.statusCode() >= 400) {
throw upstreamException(response.statusCode(),
checkedBody(response.body().errorBody()));
}
- parseStreamWithTimeout(response.body().stream(), tokenConsumer);
+ parseStreamWithTimeout(response.body().stream(), tokenConsumer,
startedAtNanos);
} catch (ResponseLimitException exception) {
throw new LlmGatewayException(502,
"llm.provider.response_too_large",
"LLM provider stream exceeded the maximum of " +
exception.limitBytes() + " bytes",
@@ -205,7 +206,7 @@ public class OpenAiCompatibleLlmClient {
}
}
- private void parseStreamWithTimeout(InputStream input, Consumer<String>
tokenConsumer)
+ private void parseStreamWithTimeout(InputStream input, Consumer<String>
tokenConsumer, long startedAtNanos)
throws IOException, InterruptedException {
FutureTask<Void> readerTask = new FutureTask<>(() -> {
parseStream(input, tokenConsumer);
@@ -213,7 +214,10 @@ public class OpenAiCompatibleLlmClient {
});
Thread.ofVirtual().name("openai-sse-reader").start(readerTask);
try {
- readerTask.get(requestTimeout.toNanos(), TimeUnit.NANOSECONDS);
+ // The input-stream body handler returns after headers; their wait
has already
+ // consumed part of the same request budget.
+ long remainingNanos = requestTimeout.toNanos() -
(System.nanoTime() - startedAtNanos);
+ readerTask.get(remainingNanos, TimeUnit.NANOSECONDS);
} catch (TimeoutException exception) {
throw new HttpTimeoutException("LLM provider stream timed out");
} catch (ExecutionException exception) {
@@ -249,8 +253,16 @@ public class OpenAiCompatibleLlmClient {
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(input, StandardCharsets.UTF_8))) {
List<String> dataLines = new ArrayList<>();
+ boolean firstLine = true;
String line;
while ((line = reader.readLine()) != null) {
+ if (firstLine) {
+ // SSE permits one UTF-8 BOM at the start of the stream,
before any field name.
+ if (line.startsWith("\uFEFF")) {
+ line = line.substring(1);
+ }
+ firstLine = false;
+ }
if (line.isEmpty()) {
if (emitStreamEvent(dataLines, tokenConsumer)) {
return;
@@ -363,22 +375,7 @@ public class OpenAiCompatibleLlmClient {
}
private URI providerUri(LlmConfigVO config, String path) {
- String baseUrl = normalizeApiBase(config.getApiBase());
- return URI.create(baseUrl + path);
- }
-
- private String normalizeApiBase(String apiBase) {
- String baseUrl = apiBase.trim();
- while (baseUrl.endsWith("/")) {
- baseUrl = baseUrl.substring(0, baseUrl.length() - 1);
- }
- if (baseUrl.endsWith(CHAT_COMPLETIONS_PATH)) {
- baseUrl = baseUrl.substring(0, baseUrl.length() -
CHAT_COMPLETIONS_PATH.length());
- while (baseUrl.endsWith("/")) {
- baseUrl = baseUrl.substring(0, baseUrl.length() - 1);
- }
- }
- return baseUrl;
+ return LlmApiBase.endpoint(config.getApiBase(), path);
}
private void validate(LlmConfigVO config) {
@@ -415,7 +412,7 @@ public class OpenAiCompatibleLlmClient {
private boolean isValidApiBase(String apiBase) {
try {
- URI uri = new URI(normalizeApiBase(apiBase));
+ URI uri = new URI(LlmApiBase.normalize(apiBase));
String scheme = uri.getScheme() == null ? "" :
uri.getScheme().toLowerCase();
return ("http".equals(scheme) || "https".equals(scheme)) &&
StringUtils.hasText(uri.getHost());
} catch (IllegalArgumentException | URISyntaxException exception) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
index c99549af6..369cb1e47 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
@@ -21,6 +21,7 @@ import jakarta.annotation.PreDestroy;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.studio.common.util.TextBounds;
import org.apache.rocketmq.studio.ops.ai.AgentProviderRegistry;
import org.apache.rocketmq.studio.ops.ai.LlmConfigVO;
import org.apache.rocketmq.studio.ops.ai.LlmGatewayException;
@@ -656,8 +657,7 @@ public class AiRunExecutor {
if (!StringUtils.hasText(value)) {
return null;
}
- String trimmed = value.trim();
- return trimmed.length() <= maxChars ? trimmed : trimmed.substring(0,
maxChars);
+ return TextBounds.truncate(value.trim(), maxChars);
}
static boolean isHttpEngine(String engine) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbe.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbe.java
index b473fe232..4f43e7946 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbe.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbe.java
@@ -91,7 +91,7 @@ public class AgentCapabilityProbe {
}
/** The three CLI probes, cached together so one request costs at most one
round of forks. */
- public Binaries binaries() {
+ public synchronized Binaries binaries() {
Instant now = clock.instant();
Binaries current = cached.get();
if (current != null && Duration.between(current.probedAt(),
now).compareTo(CACHE_TTL) < 0) {
@@ -102,7 +102,7 @@ public class AgentCapabilityProbe {
binaryProbe.isAvailable(CLAUDE_BINARY),
binaryProbe.isAvailable(QODER_BINARY),
now);
- // Last writer wins and every writer wrote a fresh probe, so a racing
refresh is harmless.
+ // Concurrent requests reuse this round instead of launching duplicate
subprocesses.
cached.set(probed);
log.debug("probed agent binaries: rmqctl={} claude={} qoder={} (cached
for {}s)",
probed.rmqctl(), probed.claude(), probed.qoder(),
CACHE_TTL_SECONDS);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspace.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspace.java
index 26174b080..d9d5cf70e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspace.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspace.java
@@ -305,7 +305,8 @@ public class RmqctlWorkspace {
throw new BusinessException(500,
"studio.ai.conversation.workspace-dir must not be blank");
}
- return Path.of(configured.trim());
+ // The agent changes cwd to its conversation directory before reading
these paths.
+ return Path.of(configured.trim()).toAbsolutePath().normalize();
}
/** The directory one conversation runs in, also its child process working
directory. */
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
index 3b5eb3a9c..17f1b8d78 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
@@ -21,6 +21,8 @@ import org.apache.rocketmq.studio.settings.GeneralSettingsVO;
import org.apache.rocketmq.studio.settings.SettingsService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import java.util.List;
@@ -360,6 +362,24 @@ class LlmConfigServiceTest {
assertThat(captor.getValue().getBaseUrl()).isEqualTo("https://api.openai.com/v1";);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"?route=/", "?route=/chat/completions",
"?tenant=a%2Fb+%26#gateway"})
+ void saveAndReadConfigShouldNormalizeOnlyTheUrlPathTest(String suffix) {
+ llmConfigService.saveConfig(LlmConfigVO.builder()
+ .provider("ollama")
+ .apiBase(" http://localhost:11434/v1/chat/completions/"; +
suffix + " ")
+ .model("llama3")
+ .enabled(true)
+ .build());
+
+ ArgumentCaptor<GeneralSettingsVO> captor =
ArgumentCaptor.forClass(GeneralSettingsVO.class);
+ verify(settingsService).saveGeneralSettings(captor.capture());
+ String expected = "http://localhost:11434/v1"; + suffix;
+ assertThat(captor.getValue().getBaseUrl()).isEqualTo(expected);
+
when(settingsService.getGeneralSettings()).thenReturn(captor.getValue());
+
assertThat(llmConfigService.getConfig().getApiBase()).isEqualTo(expected);
+ }
+
@Test
void saveConfigShouldPreserveNotificationChannelFields() {
when(settingsService.getGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
index 1348f480a..cd2e2401e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
@@ -24,6 +24,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
@@ -118,6 +119,36 @@ class OpenAiCompatibleLlmClientTest {
assertThat(requestPath.get()).isEqualTo("/v1/chat/completions");
}
+ @ParameterizedTest
+ @CsvSource({"'',''", "'/chat/completions/',''", "'','#gateway'",
"'/chat/completions/','#gateway'"})
+ void providerRequestsShouldKeepQueryParametersAfterEndpointPathTest(String
endpoint, String fragment) {
+ List<String> requestUris = new ArrayList<>();
+ server.createContext("/", exchange -> {
+ requestUris.add(exchange.getRequestURI().toString());
+ if ("GET".equals(exchange.getRequestMethod())) {
+ respond(exchange, 200, "{\"data\":[{\"id\":\"gpt-test\"}]}",
"application/json");
+ } else if
(objectMapper.readTree(exchange.getRequestBody()).path("stream").asBoolean()) {
+ respond(exchange, 200,
+ "data:
{\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\ndata: [DONE]\n\n",
+ "text/event-stream");
+ } else {
+ respond(exchange, 200,
"{\"choices\":[{\"message\":{\"content\":\"hello\"}}]}", "application/json");
+ }
+ });
+ String query = "?tenant=a%2Fb+%26&route=/chat/completions/";
+ LlmConfigVO config = config("openai", "sk-test");
+ config.setApiBase(baseUrl + endpoint + query + fragment);
+
+ client.listModels(config);
+ client.complete(config, "hello", null);
+ List<String> tokens = new ArrayList<>();
+ client.stream(config, "hello", null, tokens::add);
+
+ assertThat(requestUris).containsExactly("/v1/models" + query,
+ "/v1/chat/completions" + query, "/v1/chat/completions" +
query);
+ assertThat(tokens).containsExactly("hello");
+ }
+
@Test
void completeShouldRejectInvalidApiBaseBeforeCallingUpstream() {
LlmConfigVO config = config("openai", "sk-test");
@@ -166,6 +197,19 @@ class OpenAiCompatibleLlmClientTest {
.hasMessageContaining("UTF-8 bytes");
}
+ @Test
+ void streamShouldIgnoreOnlyTheLeadingUtf8BomTest() {
+ server.createContext("/v1/chat/completions", exchange ->
respond(exchange, 200,
+ "\uFEFFdata:
{\"choices\":[{\"delta\":{\"content\":\"first\"}}]}\n\n"
+ + "data:
{\"choices\":[{\"delta\":{\"content\":\"inner\uFEFFtext\"}}]}\n\n"
+ + "data: [DONE]\n\n", "text/event-stream"));
+ List<String> tokens = new ArrayList<>();
+
+ client.stream(config("openai", "sk-test"), "hello", null, tokens::add);
+
+ assertThat(tokens).containsExactly("first", "inner\uFEFFtext");
+ }
+
@Test
void streamShouldParseOpenAiCompatibleSseDeltas() {
AtomicReference<JsonNode> requestBody = new AtomicReference<>();
@@ -296,6 +340,39 @@ class OpenAiCompatibleLlmClientTest {
assertThat(tokens).containsExactly("first");
}
+ @Test
+ void streamTimeoutIncludesTimeSpentWaitingForHeadersTest() {
+ OpenAiCompatibleLlmClient timeoutClient = new
OpenAiCompatibleLlmClient(
+ objectMapper, HttpClient.newHttpClient(),
Duration.ofSeconds(1));
+ server.createContext("/v1/chat/completions", exchange -> {
+ exchange.getRequestBody().readAllBytes();
+ try {
+ Thread.sleep(600);
+ exchange.getResponseHeaders().set("Content-Type",
"text/event-stream");
+ exchange.sendResponseHeaders(200, 0);
+ exchange.getResponseBody().flush();
+ Thread.sleep(600);
+ exchange.getResponseBody().write("""
+ data: {"choices":[{"delta":{"content":"late answer"}}]}
+
+ data: [DONE]
+
+ """.getBytes(StandardCharsets.UTF_8));
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ } finally {
+ exchange.close();
+ }
+ });
+
+ assertThatThrownBy(() -> timeoutClient.stream(
+ config("openai", "sk-test"), "hello", null, token -> { }))
+ .isInstanceOfSatisfying(LlmGatewayException.class, exception
-> {
+ assertThat(exception.getStatusCode()).isEqualTo(504);
+
assertThat(exception.getCode()).isEqualTo("llm.provider.timeout");
+ });
+ }
+
@Test
void ollamaShouldAllowMissingApiKeyAndOmitAuthorizationHeader() {
AtomicReference<String> authorization = new AtomicReference<>();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
index f3b5fa477..ced1a222f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
@@ -35,11 +35,14 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.nio.charset.StandardCharsets;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
@@ -492,6 +495,34 @@ class AiRunExecutorTest {
assertThat(emitters.get(0).completed()).isTrue();
}
+ @ParameterizedTest
+ @ValueSource(strings = {"\uD83D\uDE00", "\uD840\uDC00"})
+ void providerFailureShouldTruncateAtCodePointBoundariesTest(String
supplementaryCharacter) {
+ String codePrefix = "c".repeat(63) + supplementaryCharacter;
+ String messagePrefix = "m".repeat(999) + supplementaryCharacter;
+ provider.failure = new LlmGatewayException(502, codePrefix +
"overflow",
+ messagePrefix + "overflow", "Retry.");
+
+ startAndRun();
+
+ assertThat(runRow().getStatus()).isEqualTo(RunStatus.FAILED.name());
+
assertThat(StandardCharsets.UTF_8.newEncoder().canEncode(runRow().getErrorMessage())).isTrue();
+
assertThat(StandardCharsets.UTF_8.newEncoder().canEncode(runRow().getErrorCode())).isTrue();
+ assertThat(runRow().getErrorCode()).isEqualTo(codePrefix);
+ assertThat(runRow().getErrorMessage()).isEqualTo(messagePrefix);
+ assertThat(emitters.get(0).completed()).isTrue();
+ }
+
+ @Test
+ void
providerFailureShouldKeepSupplementaryCharactersWithinTheColumnBudgetTest() {
+ String message = "\uD83D\uDE00".repeat(750);
+ provider.failure = new LlmGatewayException(502, "llm.provider.error",
" " + message + " ", "Retry.");
+
+ startAndRun();
+
+ assertThat(runRow().getErrorMessage()).isEqualTo(message);
+ }
+
@Test
void anErrorFromTheWorkerShouldNotBeRecordedAsACompletedRunTest() {
provider.emit(new AgentEvent.TextDelta("half an answer"));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbeConcurrencyTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbeConcurrencyTest.java
new file mode 100644
index 000000000..61c2b6c9e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/AgentCapabilityProbeConcurrencyTest.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.ops.ai.conversation.agent;
+
+import org.apache.rocketmq.studio.ops.ai.conversation.AiConversationProperties;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class AgentCapabilityProbeConcurrencyTest {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void concurrentRequestsShareColdOrExpiredProbeRoundTest(boolean expired)
throws Exception {
+ CliBinaryProbe binaryProbe = mock(CliBinaryProbe.class);
+ Clock clock = mock(Clock.class);
+ Instant start = Instant.parse("2026-10-01T00:00:00Z");
+ when(clock.instant()).thenReturn(start);
+ when(binaryProbe.isAvailable(anyString())).thenReturn(true);
+ AiConversationProperties properties = new AiConversationProperties();
+ properties.setRmqctlEnabled(true);
+ AgentCapabilityProbe probe = new AgentCapabilityProbe(binaryProbe,
properties, true, false, clock);
+ if (expired) {
+ probe.binaries();
+ when(clock.instant()).thenReturn(start.plusSeconds(60));
+ }
+
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ when(binaryProbe.isAvailable(anyString())).thenAnswer(invocation -> {
+ calls.incrementAndGet();
+ entered.countDown();
+ assertThat(release.await(5, TimeUnit.SECONDS)).isTrue();
+ return true;
+ });
+ FutureTask<AgentCapabilityProbe.Binaries> first = new
FutureTask<>(probe::binaries);
+ FutureTask<AgentCapabilityProbe.Binaries> second = new
FutureTask<>(probe::binaries);
+ Thread firstThread = new Thread(first, "first-capability-request");
+ Thread secondThread = new Thread(second, "second-capability-request");
+ firstThread.start();
+ try {
+ assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue();
+ secondThread.start();
+ awaitProbeOrMonitor(secondThread);
+ } finally {
+ release.countDown();
+ firstThread.join(TimeUnit.SECONDS.toMillis(5));
+ secondThread.join(TimeUnit.SECONDS.toMillis(5));
+ }
+
+ assertThat(second.get(5, TimeUnit.SECONDS)).isSameAs(first.get(5,
TimeUnit.SECONDS));
+ assertThat(calls).hasValue(3);
+ assertThat(probe.binaries()).isSameAs(first.get());
+ assertThat(calls).hasValue(3);
+ }
+
+ private static void awaitProbeOrMonitor(Thread thread) {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (System.nanoTime() < deadline) {
+ Thread.State state = thread.getState();
+ StackTraceElement[] stack = thread.getStackTrace();
+ if (state == Thread.State.TIMED_WAITING || state ==
Thread.State.BLOCKED
+ && stack.length > 0 &&
stack[0].getClassName().equals(AgentCapabilityProbe.class.getName())) {
+ return;
+ }
+ Thread.yield();
+ }
+ throw new AssertionError("second request did not reach the probe or
refresh monitor");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspaceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspaceTest.java
index 24f4f2475..a59cb7b2c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspaceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/RmqctlWorkspaceTest.java
@@ -118,6 +118,26 @@ class RmqctlWorkspaceTest {
() -> new AssertionError("the workspace should have been
prepared"));
}
+ @Test
+ void relativeWorkspacePathsRemainUsableFromChildWorkingDirectoryTest()
throws Exception {
+ Path configuredRoot = root.resolve("agent workspaces");
+
properties.setWorkspaceDir(Path.of("").toAbsolutePath().relativize(configuredRoot).toString());
+
+ RmqctlWorkspace.Preparation preparation = prepared();
+ Path childDirectory =
Path.of(preparation.workspaceDir()).toAbsolutePath();
+
+
assertThat(childDirectory.resolve(preparation.mcpConfigPath())).isRegularFile();
+
assertThat(childDirectory.resolve(preparation.systemPromptPath())).isRegularFile();
+
assertThat(Path.of(preparation.childEnv().get(RmqctlWorkspace.HOME_ENV)))
+ .isAbsolute().isDirectory();
+ JsonNode args =
MAPPER.readTree(Path.of(preparation.mcpConfigPath()).toFile())
+
.path("mcpServers").path(AgentEventProjector.STUDIO_MCP_SERVER).path("args");
+
assertThat(childDirectory.resolve(args.get(3).asText())).isRegularFile();
+
assertThat(availableWorkspace().workspaceRoot()).isEqualTo(configuredRoot);
+ assertThat(availableWorkspace().delete(CONVERSATION_ID)).isTrue();
+ assertThat(childDirectory).doesNotExist();
+ }
+
@Test
void writesRmqctlConfigWithEnvironmentReferencesOnlyTest() throws
Exception {
RmqctlWorkspace.Preparation preparation = prepared();
diff --git a/web/src/api/ai.test.ts b/web/src/api/ai.test.ts
index 2de42feee..b94daa7e9 100644
--- a/web/src/api/ai.test.ts
+++ b/web/src/api/ai.test.ts
@@ -397,6 +397,48 @@ describe('AI API', () => {
expect(onCancel).toHaveBeenCalledTimes(1);
});
+ it.each([
+ ['heartbeat', ':hb\n\n'],
+ ['agent', agentFrame({ type: 'text_delta', content: 'hi' })],
+ ])('rejectsOversizedRemainderAfter%sBeforeReadingAgainTest', async (_name,
prefix) => {
+ vi.useFakeTimers();
+ const onCancel = vi.fn();
+ const response = eventStreamResponse(
+ [`${prefix}event: agent\ndata: ${'x'.repeat(1024 * 1024)}`],
+ onCancel,
+ );
+ vi.stubGlobal('fetch', vi.fn().mockResolvedValue(response));
+ const failure = vi.fn();
+ const stream = openRunStream(7, body, { onEvent: vi.fn()
}).catch(failure);
+
+ await vi.advanceTimersByTimeAsync(0);
+ const failedImmediately = failure.mock.calls.length === 1;
+ // Also settle the unfixed reader so a failed assertion cannot leave a
pending stream.
+ await vi.advanceTimersByTimeAsync(31_000);
+ await stream;
+
+ expect(failedImmediately).toBe(true);
+ expect(failure).toHaveBeenCalledWith(
+ expect.objectContaining({ code: 'llm.stream.event_too_large' }),
+ );
+ expect(onCancel).toHaveBeenCalledOnce();
+ });
+
+ it('acceptsLargeChunksContainingIndividuallyBoundedFramesTest', async ()
=> {
+ const content = 'x'.repeat(600 * 1024);
+ const frame = agentFrame({ type: 'text_delta', content });
+ vi.stubGlobal(
+ 'fetch',
+ vi.fn().mockResolvedValue(eventStreamResponse([frame + frame +
doneFrame()])),
+ );
+ const onEvent = vi.fn();
+
+ await openRunStream(7, body, { onEvent });
+
+ expect(onEvent).toHaveBeenCalledTimes(2);
+ expect(onEvent).toHaveBeenLastCalledWith({ type: 'text_delta', content
});
+ });
+
it('throwsStructuredErrorsFromErrorFramesTest', async () => {
vi.stubGlobal(
'fetch',
@@ -472,6 +514,40 @@ describe('AI API', () => {
);
});
+ it.each(['open', 'attach'] as const)(
+ 'cancelsRejectedContentTypeResponseFor%sTest',
+ async (entryPoint) => {
+ const onCancel = vi.fn();
+ const response = new Response(new ReadableStream({ cancel: onCancel
}), {
+ headers: { 'content-type': 'text/html' },
+ });
+ vi.stubGlobal('fetch', vi.fn().mockResolvedValue(response));
+
+ const stream =
+ entryPoint === 'open'
+ ? openRunStream(7, body, { onEvent: vi.fn() })
+ : attachRunStream(41, 0, { onEvent: vi.fn() });
+
+ await expect(stream).rejects.toMatchObject({
+ code: 'llm.stream.unexpected_content_type',
+ });
+ expect(onCancel).toHaveBeenCalledOnce();
+ },
+ );
+
+ it('preservesContentTypeErrorWhenCancellingTheBodyFailsTest', async () => {
+ const onCancel = vi.fn().mockRejectedValue(new Error('cancel failed'));
+ const response = new Response(new ReadableStream({ cancel: onCancel }), {
+ headers: { 'content-type': 'text/html' },
+ });
+ vi.stubGlobal('fetch', vi.fn().mockResolvedValue(response));
+
+ await expect(attachRunStream(41, 0, { onEvent: vi.fn()
})).rejects.toMatchObject({
+ code: 'llm.stream.unexpected_content_type',
+ });
+ expect(onCancel).toHaveBeenCalledOnce();
+ });
+
it('rejectsAResponseWithoutAnyContentTypeTest', async () => {
// A null body is the only way to get a Response with no content-type at
all: a string body
// makes the Fetch API default the header to text/plain.
diff --git a/web/src/api/ai.ts b/web/src/api/ai.ts
index dd8df34f4..15cd472be 100644
--- a/web/src/api/ai.ts
+++ b/web/src/api/ai.ts
@@ -270,9 +270,6 @@ async function consumeEventStream(
if (done) break;
buffer += decoder.decode(value, { stream: true });
- if (buffer.length > MAX_SSE_EVENT_CHARS && !getEventBoundary(buffer)) {
- throw eventTooLargeError();
- }
let boundary = getEventBoundary(buffer);
while (boundary) {
const frame = buffer.slice(0, boundary.index);
@@ -280,6 +277,9 @@ async function consumeEventStream(
if (dispatch(frame)) return;
boundary = getEventBoundary(buffer);
}
+ if (buffer.length > MAX_SSE_EVENT_CHARS) {
+ throw eventTooLargeError();
+ }
}
// A server that closes without a trailing blank line still delivered its
last frame.
@@ -358,7 +358,13 @@ async function consumeRunStream(
if (!response.ok) {
throw await parseHttpError(response, label);
}
- assertEventStream(response);
+ try {
+ assertEventStream(response);
+ } catch (error) {
+ // No reader owns the body yet when the content-type check rejects it.
+ await response.body?.cancel().catch(() => undefined);
+ throw error;
+ }
await consumeEventStream(
response,
(frame) => emitRunFrame(frame, handlers),
