applyTimeoutAndRetry(
return responseFlux;
}
+ /**
+ * Ensures the completion contains at least one content-bearing response.
+ *
+ * Chunks that carry no content block are withheld until the first content-bearing chunk
+ * arrives (then released in order); a stream that completes without one — zero chunks, or
+ * only chunks whose {@code getContent()} is null/empty — is converted to a {@code
+ * ModelException}. Withholding is what keeps fallback effective: switchOnFirst-based
+ * fallback commits on the first downstream signal, so an empty completion must reach it as
+ * an error, not preceded by contentless chunks that would already commit the primary model.
+ * Any content block counts as content-bearing (multimodal blocks included). This detection
+ * sits after timeout (which would have already fired) and before retryWhen, so an empty
+ * completion triggers retry just like any other model error.
+ *
+ * @param responseFlux the response stream to guard
+ * @param modelName the model name for error messages
+ * @param provider the provider name for error messages
+ * @return guarded flux that errors on empty completions
+ */
+ private static Flux ensureNonEmptyCompletion(
+ Flux responseFlux, String modelName, String provider) {
+ // Per-subscription state: retryWhen resubscribes this chain on each attempt, so the
+ // withheld-chunk buffer and the released flag must be recreated per subscription —
+ // chunks withheld by an attempt that failed mid-stream must not leak into a later,
+ // empty attempt. Signals are delivered serially per subscriber, so unsynchronized state
+ // is safe.
+ return Flux.defer(
+ () -> {
+ List withheld = new ArrayList<>();
+ AtomicBoolean released = new AtomicBoolean(false);
+ return responseFlux
+ .concatMap(
+ chunk -> {
+ if (released.get()) {
+ return Flux.just(chunk);
+ }
+ withheld.add(chunk);
+ if (!isContentBearing(chunk)) {
+ return Flux.empty();
+ }
+ released.set(true);
+ return Flux.fromIterable(withheld);
+ })
+ .concatWith(
+ Mono.defer(
+ () -> {
+ if (released.get()) {
+ return Mono.empty();
+ }
+ LOG.warn(
+ "Model {} ({}) returned an empty completion"
+ + " (zero content-bearing chunks)",
+ modelName,
+ provider);
+ return Mono.error(
+ new ModelException(
+ "Model returned empty completion"
+ + " (zero content-bearing"
+ + " chunks)",
+ modelName,
+ provider));
+ }));
+ });
+ }
+
+ /**
+ * Returns whether the chunk carries any content block.
+ *
+ * Type-agnostic by design: multimodal completions may carry only image/video/audio
+ * blocks, which are valid completions, not empty ones.
+ *
+ * @param chunk the response chunk to inspect
+ * @return true if the chunk carries at least one content block
+ */
+ private static boolean isContentBearing(ChatResponse chunk) {
+ return chunk.getContent() != null && !chunk.getContent().isEmpty();
+ }
+
/**
* Ensures GenerateOptions has MODEL_DEFAULTS for executionConfig applied.
*
diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentTest.java
index d71b6e6f46..06c2a8b0e4 100644
--- a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentTest.java
+++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentTest.java
@@ -40,6 +40,10 @@
import io.agentscope.core.message.ToolUseBlock;
import io.agentscope.core.model.ChatResponse;
import io.agentscope.core.model.ChatUsage;
+import io.agentscope.core.model.GenerateOptions;
+import io.agentscope.core.model.Model;
+import io.agentscope.core.model.ModelUtils;
+import io.agentscope.core.model.ToolSchema;
import io.agentscope.core.tool.Toolkit;
import io.agentscope.core.util.JsonUtils;
import java.time.Duration;
@@ -52,6 +56,7 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
@@ -578,6 +583,50 @@ void testFallbackModel() {
assertEquals(1, fallbackModel.getCallCount(), "Fallback model should be called once");
}
+ @Test
+ @DisplayName("Should switch to fallback model when the primary returns an empty completion")
+ void testFallbackModelOnEmptyCompletion() {
+ // Primary emits only contentless chunks (role-only gateway response) and completes —
+ // the empty-completion guard must surface the error as the first signal so the
+ // switchOnFirst fallback engages instead of the run ending as silent success
+ Model primaryModel =
+ new Model() {
+ @Override
+ public Flux stream(
+ List messages, List tools, GenerateOptions options) {
+ Flux responseFlux =
+ Flux.just(
+ new ChatResponse("chunk-1", List.of(), null, null, null),
+ new ChatResponse("chunk-2", null, null, null, null));
+ return ModelUtils.applyTimeoutAndRetry(
+ responseFlux, options, null, "empty-primary", "test");
+ }
+
+ @Override
+ public String getModelName() {
+ return "empty-primary";
+ }
+ };
+ MockModel fallbackModel = new MockModel("Fallback response");
+
+ agent =
+ ReActAgent.builder()
+ .name(TestConstants.TEST_REACT_AGENT_NAME)
+ .sysPrompt(TestConstants.DEFAULT_SYS_PROMPT)
+ .model(primaryModel)
+ .fallbackModel(fallbackModel)
+ .toolkit(mockToolkit)
+ .build();
+
+ Msg userMsg = TestUtils.createUserMessage("User", TestConstants.TEST_USER_INPUT);
+ Msg response =
+ agent.call(userMsg).block(Duration.ofMillis(TestConstants.DEFAULT_TEST_TIMEOUT_MS));
+
+ assertNotNull(response, "Response should not be null");
+ assertEquals("Fallback response", TestUtils.extractTextContent(response));
+ assertEquals(1, fallbackModel.getCallCount(), "Fallback model should be called once");
+ }
+
@Test
@DisplayName("Should support streaming responses")
void testStreaming() {
diff --git a/agentscope-core/src/test/java/io/agentscope/core/model/ModelTimeoutRetryTest.java b/agentscope-core/src/test/java/io/agentscope/core/model/ModelTimeoutRetryTest.java
index 69edc80107..cf589678c5 100644
--- a/agentscope-core/src/test/java/io/agentscope/core/model/ModelTimeoutRetryTest.java
+++ b/agentscope-core/src/test/java/io/agentscope/core/model/ModelTimeoutRetryTest.java
@@ -17,13 +17,16 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
+import io.agentscope.core.message.ImageBlock;
import io.agentscope.core.message.Msg;
import io.agentscope.core.message.MsgRole;
import io.agentscope.core.message.TextBlock;
+import io.agentscope.core.message.URLSource;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
@@ -386,4 +389,238 @@ private ChatResponse createMockResponse() {
null,
null);
}
+
+ /**
+ * Creates a mock model backed by the supplied flux supplier, wrapped with the shared
+ * timeout/retry/empty-completion handling. The supplier is re-invoked on every subscription,
+ * so retry attempts can vary their behavior.
+ *
+ * @param responseFlux supplies the response flux for each subscription
+ * @param modelName the model name for error messages
+ * @return a Model instance backed by the supplier
+ */
+ private Model createModelReturning(
+ Supplier> responseFlux, String modelName) {
+ return new Model() {
+ @Override
+ public Flux stream(
+ List messages, List tools, GenerateOptions options) {
+ return ModelUtils.applyTimeoutAndRetry(
+ Flux.defer(responseFlux), options, null, modelName, "test");
+ }
+
+ @Override
+ public String getModelName() {
+ return modelName;
+ }
+ };
+ }
+
+ // ==================== Empty-completion detection (issue #2962) ====================
+
+ @Test
+ @DisplayName("Should fail when the model returns zero chunks")
+ void testEmptyCompletionZeroChunks() {
+ // Zero chunks — the [DONE]-only stream case
+ Model emptyModel = createModelReturning(() -> Flux.empty(), "empty-model");
+
+ Msg testMsg =
+ Msg.builder()
+ .name("user")
+ .role(MsgRole.USER)
+ .content(TextBlock.builder().text("test").build())
+ .build();
+
+ // Empty-completion detection is unconditional (no ExecutionConfig needed)
+ StepVerifier.create(emptyModel.stream(List.of(testMsg), null, null))
+ .expectErrorMatches(
+ error ->
+ error instanceof ModelException
+ && error.getMessage().contains("empty completion"))
+ .verify();
+ }
+
+ @Test
+ @DisplayName("Should fail when all chunks have null or empty content")
+ void testEmptyCompletionAllChunksEmpty() {
+ // Two chunks, both contentless — the role-only/usage-only chunks of a well-formed but
+ // empty gateway response
+ Model emptyContentModel =
+ createModelReturning(
+ () ->
+ Flux.just(
+ new ChatResponse("chunk-1", List.of(), null, null, null),
+ new ChatResponse("chunk-2", null, null, null, null)),
+ "empty-content-model");
+
+ Msg testMsg =
+ Msg.builder()
+ .name("user")
+ .role(MsgRole.USER)
+ .content(TextBlock.builder().text("test").build())
+ .build();
+
+ // Contentless chunks are withheld, so the error must be the first downstream signal:
+ // any emitted chunk would commit a switchOnFirst-based fallback to the primary model
+ StepVerifier.create(emptyContentModel.stream(List.of(testMsg), null, null))
+ .expectErrorMatches(
+ error ->
+ error instanceof ModelException
+ && error.getMessage().contains("empty completion"))
+ .verify();
+ }
+
+ @Test
+ @DisplayName("Empty completion should trigger retry when maxAttempts > 1")
+ void testEmptyCompletionRetriesAndSucceeds() {
+ AtomicInteger attemptCount = new AtomicInteger(0);
+
+ Model retryingEmptyModel =
+ createModelReturning(
+ () ->
+ attemptCount.incrementAndGet() < 2
+ ? Flux.empty() // First attempt: empty
+ : Flux.just(createMockResponse()), // Then: success
+ "retrying-empty-model");
+
+ ExecutionConfig executionConfig =
+ ExecutionConfig.builder()
+ .maxAttempts(2)
+ .initialBackoff(Duration.ofMillis(10))
+ .build();
+
+ GenerateOptions options =
+ GenerateOptions.builder().executionConfig(executionConfig).build();
+
+ Msg testMsg =
+ Msg.builder()
+ .name("user")
+ .role(MsgRole.USER)
+ .content(TextBlock.builder().text("test").build())
+ .build();
+
+ // Should retry the empty completion and succeed on second attempt
+ StepVerifier.create(retryingEmptyModel.stream(List.of(testMsg), null, options))
+ .expectNextCount(1)
+ .verifyComplete();
+
+ assertEquals(2, attemptCount.get(), "Should have made exactly 2 attempts");
+ }
+
+ @Test
+ @DisplayName(
+ "Empty completion on retry must fail even if a content-bearing attempt errored first")
+ void testEmptyCompletionOnRetryAfterContentBearingFailure() {
+ AtomicInteger attemptCount = new AtomicInteger(0);
+
+ Model model =
+ createModelReturning(
+ () -> {
+ if (attemptCount.incrementAndGet() == 1) {
+ // First attempt: content arrives, then the stream fails
+ // mid-flight
+ return Flux.concat(
+ Flux.just(createMockResponse()),
+ Flux.error(new RuntimeException("boom")));
+ }
+ // Retry: empty completion must still fail
+ return Flux.empty();
+ },
+ "retry-after-content-model");
+
+ ExecutionConfig executionConfig =
+ ExecutionConfig.builder()
+ .maxAttempts(2)
+ .initialBackoff(Duration.ofMillis(10))
+ .build();
+
+ GenerateOptions options =
+ GenerateOptions.builder().executionConfig(executionConfig).build();
+
+ Msg testMsg =
+ Msg.builder()
+ .name("user")
+ .role(MsgRole.USER)
+ .content(TextBlock.builder().text("test").build())
+ .build();
+
+ // The empty completion on the second attempt must surface as ModelException; the chunk
+ // seen on the first attempt must not satisfy the check for the retried attempt.
+ // Retry.backoff wraps the last failure in RetryExhaustedException when retries run out
+ StepVerifier.create(model.stream(List.of(testMsg), null, options))
+ .expectNextCount(1)
+ .expectErrorMatches(
+ error ->
+ error.getCause() instanceof ModelException
+ && error.getCause()
+ .getMessage()
+ .contains("empty completion"))
+ .verify();
+
+ assertEquals(2, attemptCount.get(), "Should have made exactly 2 attempts");
+ }
+
+ @Test
+ @DisplayName("Multimodal-only chunks (image blocks) are not empty completions")
+ void testMultimodalOnlyCompletionIsNotEmpty() {
+ ChatResponse imageChunk =
+ new ChatResponse(
+ "chunk-1",
+ List.of(
+ ImageBlock.builder()
+ .source(
+ URLSource.builder()
+ .url("https://example.com/image.jpg")
+ .build())
+ .build()),
+ null,
+ null,
+ null);
+ Model multimodalModel =
+ createModelReturning(() -> Flux.just(imageChunk), "multimodal-model");
+
+ Msg testMsg =
+ Msg.builder()
+ .name("user")
+ .role(MsgRole.USER)
+ .content(TextBlock.builder().text("test").build())
+ .build();
+
+ // Image-only chunks carry content; the stream must complete normally
+ StepVerifier.create(multimodalModel.stream(List.of(testMsg), null, null))
+ .expectNextCount(1)
+ .verifyComplete();
+ }
+
+ @Test
+ @DisplayName("Should release withheld contentless chunks in order once content arrives")
+ void testLeadingContentlessChunksReleasedWithFirstContentChunk() {
+ Model model =
+ createModelReturning(
+ () ->
+ Flux.just(
+ new ChatResponse("role-chunk", List.of(), null, null, null),
+ createMockResponse(),
+ new ChatResponse("usage-chunk", null, null, null, null)),
+ "leading-contentless-model");
+
+ Msg testMsg =
+ Msg.builder()
+ .name("user")
+ .role(MsgRole.USER)
+ .content(TextBlock.builder().text("test").build())
+ .build();
+
+ // The withheld chunks stream out together with the first content-bearing chunk, and
+ // trailing contentless chunks pass through untouched
+ StepVerifier.create(model.stream(List.of(testMsg), null, null))
+ .expectNextMatches(chunk -> "role-chunk".equals(chunk.getId()))
+ .expectNextMatches(
+ chunk ->
+ "test-id".equals(chunk.getId())
+ && chunk.getContent() != null
+ && !chunk.getContent().isEmpty())
+ .expectNextMatches(chunk -> "usage-chunk".equals(chunk.getId()))
+ .verifyComplete();
+ }
}
diff --git a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeChatModelTest.java b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeChatModelTest.java
index 1dbac5c29b..3ebaeda90e 100644
--- a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeChatModelTest.java
+++ b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeChatModelTest.java
@@ -556,19 +556,16 @@ void testDoStreamWithAdditionHeadersAndParams() throws Exception {
MockWebServer mockServer = new MockWebServer();
mockServer.start();
+ // The response must carry content: contentless completions are rejected as upstream
+ // anomalies (issue #2962)
mockServer.enqueue(
new MockResponse()
.setResponseCode(200)
.setBody(
- """
- {
- "request_id": "test",
- "output": {
- "choices": []
- }
- }
- """)
- .setHeader("Content-Type", "application/json"));
+ "data:"
+ + " {\"request_id\":\"test\",\"output\":{\"choices\":[{\"message\":{\"role\":\"assistant\",\"content\":\"ok\"}}]}}\n\n"
+ + "data: [DONE]\n\n")
+ .setHeader("Content-Type", "text/event-stream"));
DashScopeChatModel chatModel =
DashScopeChatModel.builder().apiKey(mockApiKey).modelName("qwen-plus").stream(true)
@@ -609,17 +606,25 @@ void testDoNonStreamWithAdditionHeadersAndParams() throws Exception {
MockWebServer mockServer = new MockWebServer();
mockServer.start();
+ // The response must carry content: contentless completions are rejected as upstream
+ // anomalies (issue #2962)
mockServer.enqueue(
new MockResponse()
.setResponseCode(200)
.setBody(
"""
- {
- "request_id": "test",
- "output": {
- "choices": []
+ {
+ "request_id": "test",
+ "output": {
+ "choices": [{
+ "finish_reason": "stop",
+ "message": {
+ "role": "assistant",
+ "content": "ok"
}
- }
+ }]
+ }
+ }
""")
.setHeader("Content-Type", "application/json"));
@@ -940,12 +945,18 @@ void testCacheControlApplied() throws Exception {
.setResponseCode(200)
.setBody(
"""
- {
- "request_id": "test",
- "output": {
- "choices": []
+ {
+ "request_id": "test",
+ "output": {
+ "choices": [{
+ "finish_reason": "stop",
+ "message": {
+ "role": "assistant",
+ "content": "ok"
}
- }
+ }]
+ }
+ }
""")
.setHeader("Content-Type", "application/json"));
@@ -996,12 +1007,18 @@ void testCacheControlNotAppliedWhenDisabled() throws Exception {
.setResponseCode(200)
.setBody(
"""
- {
- "request_id": "test",
- "output": {
- "choices": []
+ {
+ "request_id": "test",
+ "output": {
+ "choices": [{
+ "finish_reason": "stop",
+ "message": {
+ "role": "assistant",
+ "content": "ok"
}
- }
+ }]
+ }
+ }
""")
.setHeader("Content-Type", "application/json"));
@@ -1099,19 +1116,17 @@ void testEnableWebExtractorTool() throws Exception {
DashScopeParameters parameters =
DashScopeParameters.builder().searchOptions(searchOptions).build();
+ // enableThinking forces streaming regardless of stream(false), so the mock must be an
+ // SSE body carrying content — contentless completions are rejected as upstream
+ // anomalies (issue #2962)
mockServer.enqueue(
new MockResponse()
.setResponseCode(200)
.setBody(
- """
- {
- "request_id": "test",
- "output": {
- "choices": []
- }
- }
- """)
- .setHeader("Content-Type", "application/json"));
+ "data:"
+ + " {\"request_id\":\"test\",\"output\":{\"choices\":[{\"message\":{\"role\":\"assistant\",\"content\":\"ok\"}}]}}\n\n"
+ + "data: [DONE]\n\n")
+ .setHeader("Content-Type", "text/event-stream"));
DashScopeChatModel chatModel =
DashScopeChatModel.builder().apiKey(mockApiKey).modelName("qwen-plus").stream(false)
@@ -1170,12 +1185,18 @@ void testEnableCodeInterpreterTool() throws Exception {
.setResponseCode(200)
.setBody(
"""
- {
- "request_id": "test",
- "output": {
- "choices": []
+ {
+ "request_id": "test",
+ "output": {
+ "choices": [{
+ "finish_reason": "stop",
+ "message": {
+ "role": "assistant",
+ "content": "ok"
}
- }
+ }]
+ }
+ }
""")
.setHeader("Content-Type", "application/json"));
@@ -1226,12 +1247,18 @@ void testEnableParallelToolCalls() throws Exception {
.setResponseCode(200)
.setBody(
"""
- {
- "request_id": "test",
- "output": {
- "choices": []
+ {
+ "request_id": "test",
+ "output": {
+ "choices": [{
+ "finish_reason": "stop",
+ "message": {
+ "role": "assistant",
+ "content": "ok"
}
- }
+ }]
+ }
+ }
""")
.setHeader("Content-Type", "application/json"));
diff --git a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeNonStreamingBlockingBehaviorTest.java b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeNonStreamingBlockingBehaviorTest.java
index 5630749225..e0a84a28d8 100644
--- a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeNonStreamingBlockingBehaviorTest.java
+++ b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeNonStreamingBlockingBehaviorTest.java
@@ -54,10 +54,15 @@ void tearDown() throws IOException {
@Test
@DisplayName("DashScopeChatModel - Should be NON-BLOCKING in non-streaming mode")
void testDashScopeChatModelNonBlocking() throws Exception {
+ // The response must carry content: contentless completions are rejected as upstream
+ // anomalies (issue #2962), and this test needs an onNext to observe the delivering
+ // thread
mockServer.enqueue(
new MockResponse()
.setResponseCode(200)
- .setBody("{\"request_id\":\"test\",\"output\":{\"choices\":[]}}")
+ .setBody(
+ "{\"request_id\":\"test\",\"output\":{\"choices\":[{\"finish_reason\":"
+ + "\"stop\",\"message\":{\"role\":\"assistant\",\"content\":\"ok\"}}]}}")
.setHeader("Content-Type", "application/json"));
DashScopeChatModel model =
diff --git a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/OpenAIChatModelTest.java b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/OpenAIChatModelTest.java
index 9b3cbf5b44..e4bfcdd714 100644
--- a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/OpenAIChatModelTest.java
+++ b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/OpenAIChatModelTest.java
@@ -600,6 +600,9 @@ void testProxyInHttpTransportConfig() {
@Test
@DisplayName("Should enable parallel tool calls when set parallel_tool_calls to true")
void testEnableParallelToolCalls() throws Exception {
+ // The response must carry content: a completion with no content-bearing chunk is
+ // rejected as an upstream anomaly (issue #2962), and this test asserts on the request,
+ // not the response
String responseJson =
"""
{
@@ -607,7 +610,14 @@ void testEnableParallelToolCalls() throws Exception {
"object": "chat.completion",
"created": 1677652280,
"model": "gpt-4",
- "choices": []
+ "choices": [{
+ "index": 0,
+ "message": {
+ "role": "assistant",
+ "content": "ok"
+ },
+ "finish_reason": "stop"
+ }]
}
""";