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 a10faa9a [ISSUE #1517] Enforce timeout in Claude CLI streaming (#1518)
a10faa9a is described below
commit a10faa9a7b5bcbff57e56fe2b6507f5aa671765b
Author: youngkermit8-coder <[email protected]>
AuthorDate: Tue Aug 11 20:20:51 2026 +0800
[ISSUE #1517] Enforce timeout in Claude CLI streaming (#1518)
Signed-off-by: youngkermit8-coder <[email protected]>
---
.../studio/ops/ai/ClaudeCodeAgentProvider.java | 67 +++++++++++++---
.../studio/ops/ai/ClaudeCodeAgentProviderTest.java | 90 ++++++++++++++++++++++
2 files changed, 147 insertions(+), 10 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
index ea7e8a52..8e83df37 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
@@ -24,13 +24,17 @@ import org.springframework.util.StringUtils;
import java.io.BufferedReader;
import java.io.IOException;
+import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
@@ -47,6 +51,7 @@ public class ClaudeCodeAgentProvider extends CliAgentProvider
{
private static final String COMPATIBLE_MODE_SUFFIX = "/compatible-mode/v1";
private static final String ANTHROPIC_APP_SUFFIX = "/apps/anthropic";
private static final long STREAM_TIMEOUT_SECONDS = 300;
+ private static final long OUTPUT_DRAIN_TIMEOUT_SECONDS = 10;
private final LlmProperties llmProperties;
private final ObjectMapper objectMapper = new ObjectMapper();
@@ -96,22 +101,20 @@ public class ClaudeCodeAgentProvider extends
CliAgentProvider {
Process process = builder.start();
AtomicBoolean emitted = new AtomicBoolean(false);
StringBuilder resultText = new StringBuilder();
- try (BufferedReader reader = new BufferedReader(
- new InputStreamReader(process.getInputStream(),
StandardCharsets.UTF_8))) {
- String line;
- while ((line = reader.readLine()) != null) {
- parseStreamLine(line, tokenConsumer, emitted, resultText);
- }
- }
- boolean finished = process.waitFor(STREAM_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ CompletableFuture<Void> stdoutFuture = drainStdout(
+ process.getInputStream(), tokenConsumer, emitted,
resultText);
+ CompletableFuture<String> stderrFuture =
readAsync(process.getErrorStream());
+ long timeoutSeconds = streamTimeoutSeconds();
+ boolean finished = process.waitFor(timeoutSeconds,
TimeUnit.SECONDS);
if (!finished) {
process.destroyForcibly();
throw new LlmGatewayException(504, "llm.provider.timeout",
- binaryName() + " CLI stream timed out after " +
STREAM_TIMEOUT_SECONDS + "s",
+ binaryName() + " CLI stream timed out after " +
timeoutSeconds + "s",
"Retry with a shorter prompt or check the gateway
latency.");
}
+ await(stdoutFuture);
+ String stderr = await(stderrFuture);
if (process.exitValue() != 0 && !emitted.get()) {
- String stderr = new
String(process.getErrorStream().readAllBytes(), StandardCharsets.UTF_8);
throw new LlmGatewayException(502, "llm.provider.cli_error",
binaryName() + " CLI failed: " +
(StringUtils.hasText(stderr) ? stderr.trim() : "unknown error"),
"Check the provider credentials, base URL and model
name.");
@@ -130,6 +133,50 @@ public class ClaudeCodeAgentProvider extends
CliAgentProvider {
}
}
+ protected long streamTimeoutSeconds() {
+ return STREAM_TIMEOUT_SECONDS;
+ }
+
+ private CompletableFuture<Void> drainStdout(InputStream stdout,
Consumer<String> tokenConsumer,
+ AtomicBoolean emitted,
StringBuilder resultText) {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ Thread.ofVirtual().start(() -> {
+ try (BufferedReader reader = new BufferedReader(
+ new InputStreamReader(stdout, StandardCharsets.UTF_8))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ parseStreamLine(line, tokenConsumer, emitted, resultText);
+ }
+ result.complete(null);
+ } catch (Exception exception) {
+ result.completeExceptionally(exception);
+ }
+ });
+ return result;
+ }
+
+ private CompletableFuture<String> readAsync(InputStream stream) {
+ CompletableFuture<String> result = new CompletableFuture<>();
+ Thread.ofVirtual().start(() -> {
+ try (stream) {
+ result.complete(new String(stream.readAllBytes(),
StandardCharsets.UTF_8));
+ } catch (Exception exception) {
+ result.completeExceptionally(exception);
+ }
+ });
+ return result;
+ }
+
+ private <T> T await(CompletableFuture<T> future) throws IOException,
InterruptedException {
+ try {
+ return future.get(OUTPUT_DRAIN_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (ExecutionException exception) {
+ throw new IOException("Failed to drain Claude CLI output",
exception.getCause());
+ } catch (TimeoutException exception) {
+ throw new IOException("Timed out while draining Claude CLI
output", exception);
+ }
+ }
+
/** Parses one stream-json line: emits text deltas, records the final
result. */
private void parseStreamLine(String line, Consumer<String> tokenConsumer,
AtomicBoolean emitted, StringBuilder
resultText) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java
new file mode 100644
index 00000000..8cedc2dd
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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 org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ClaudeCodeAgentProviderTest {
+
+ @Test
+ void streamShouldDrainLargeStderrOutputTest() {
+ TestClaudeCodeAgentProvider provider = new
TestClaudeCodeAgentProvider(List.of(
+ "sh", "-c", "yes error | head -c 131072 >&2; "
+ + "printf eyJ0eXBlIjoicmVzdWx0IiwicmVzdWx0IjoiZG9uZSJ9
| base64 -d"), 5);
+ List<String> tokens = new ArrayList<>();
+
+ provider.stream(LlmConfigVO.builder().build(), "prompt", null,
tokens::add);
+
+ assertThat(tokens).containsExactly("done");
+ }
+
+ @Test
+ void streamShouldEnforceTimeoutBeforeWaitingForStdoutTest() {
+ TestClaudeCodeAgentProvider provider = new TestClaudeCodeAgentProvider(
+ List.of("sh", "-c", "sleep 2"), 1);
+
+ assertThatThrownBy(() -> provider.stream(
+ LlmConfigVO.builder().build(), "prompt", null, ignored -> { }))
+ .isInstanceOf(LlmGatewayException.class)
+ .satisfies(exception -> assertThat(((LlmGatewayException)
exception).getStatusCode())
+ .isEqualTo(504));
+ }
+
+ private static class TestClaudeCodeAgentProvider extends
ClaudeCodeAgentProvider {
+
+ private final List<String> command;
+ private final long timeoutSeconds;
+
+ TestClaudeCodeAgentProvider(List<String> command, long timeoutSeconds)
{
+ super(null);
+ this.command = command;
+ this.timeoutSeconds = timeoutSeconds;
+ }
+
+ @Override
+ public boolean available() {
+ return true;
+ }
+
+ @Override
+ protected List<String> buildCommand(LlmConfigVO config, String prompt,
String modelOverride) {
+ return new ArrayList<>(command);
+ }
+
+ @Override
+ protected Map<String, String> childEnv(LlmConfigVO config) {
+ return Map.of();
+ }
+
+ @Override
+ protected String binaryName() {
+ return "sh";
+ }
+
+ @Override
+ protected long streamTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+ }
+}
