diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java index 1b269cc6ad..7fbcfc344c 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java +++ b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java @@ -103,7 +103,11 @@ private static boolean isRetryableError(Throwable error) { return hte.isRetryable(); } - if (error instanceof ModelHttpException mhe) { + // Only treat a ModelHttpException as an HTTP response error when a status code is + // present. Implementations without a status code (e.g. OpenAIException wrapping a + // streaming transport failure) must fall through to the transport/IO and cause-chain + // checks below instead of being classified as a permanent client error (issue #3057). + if (error instanceof ModelHttpException mhe && mhe.getStatusCode() != null) { return mhe.isRetryableHttpStatus(); } diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java b/agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java index c8f0cf3cd8..5549542d02 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java +++ b/agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java @@ -15,7 +15,10 @@ */ package io.agentscope.core.model; +import io.agentscope.core.message.ContentBlock; import java.time.Duration; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Predicate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -110,25 +113,88 @@ public static Flux applyTimeoutAndRetry( if (retryOn == null) { retryOn = error -> true; // retry all errors by default } + // Lambda captures below require effectively final variables + final int retryCount = maxAttempts - 1; + final Duration effectiveInitialBackoff = initialBackoff; + final Duration effectiveMaxBackoff = maxBackoff; + final Predicate effectiveRetryOn = retryOn; - Retry retrySpec = - Retry.backoff(maxAttempts - 1, initialBackoff) - .maxBackoff(maxBackoff) - .jitter(0.5) - .filter(retryOn) - .doBeforeRetry( - signal -> - LOG.warn( - "Retrying model request (attempt {}/{}) due" - + " to: {}", - signal.totalRetriesInARow() + 1, - maxAttempts - 1, - signal.failure().getMessage(), - signal.failure())); - - responseFlux = responseFlux.retryWhen(retrySpec); + // Retrying is only safe before user-visible content is emitted: resubscribing + // after partial output would reissue the request and downstream would observe + // the first partial response followed by the retried one, duplicating content. + // The visibility flag is created inside defer so every subscription (model + // call) gets a fresh flag, while it persists across retry attempts of the same + // call. Role-only or usage-only chunks carry no content blocks and do not + // disable retries — nothing user-visible has been delivered yet. + final Flux source = responseFlux; + responseFlux = + Flux.defer( + () -> { + AtomicBoolean emittedVisibleContent = new AtomicBoolean(false); + return source.doOnNext( + response -> { + if (hasVisibleContent(response)) { + emittedVisibleContent.set(true); + } + }) + .retryWhen( + Retry.backoff( + retryCount, + effectiveInitialBackoff) + .maxBackoff(effectiveMaxBackoff) + .jitter(0.5) + .filter( + error -> { + if (emittedVisibleContent + .get()) { + if (effectiveRetryOn + .test(error)) { + LOG.warn( + "Skipping" + + " retry" + + " of retryable" + + " error" + + " because" + + " visible" + + " content" + + " was already" + + " emitted" + + " (retrying" + + " would" + + " duplicate" + + " it):" + + " model={}," + + " error={}", + modelName, + error + .getMessage()); + } + return false; + } + return effectiveRetryOn + .test(error); + }) + .doBeforeRetry( + signal -> + LOG.warn( + "Retrying model" + + " request" + + " (attempt" + + " {}/{})" + + " due to:" + + " {}", + signal + .totalRetriesInARow() + + 1, + retryCount, + signal.failure() + .getMessage(), + signal + .failure()))); + }); LOG.debug( - "Applied retry config: maxAttempts={}, initialBackoff={} for model: {}", + "Applied retry config: maxAttempts={}, initialBackoff={}," + + " retryBeforeFirstResponse=true for model: {}", maxAttempts, initialBackoff, modelName); @@ -138,6 +204,20 @@ public static Flux applyTimeoutAndRetry( return responseFlux; } + /** + * Whether the response carries user-visible content (text, thinking or tool-call deltas). + * + *

Role-only or usage-only chunks carry no content blocks — nothing user-visible has + * been delivered, so retrying past them cannot duplicate content.

+ * + * @param response the response chunk to inspect + * @return true if the chunk contains at least one content block + */ + private static boolean hasVisibleContent(ChatResponse response) { + List content = response.getContent(); + return content != null && !content.isEmpty(); + } + /** * Ensures GenerateOptions has MODEL_DEFAULTS for executionConfig applied. * diff --git a/agentscope-core/src/test/java/io/agentscope/core/model/ExecutionConfigTest.java b/agentscope-core/src/test/java/io/agentscope/core/model/ExecutionConfigTest.java index 7f3d25051e..6b6229c8ab 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/model/ExecutionConfigTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/model/ExecutionConfigTest.java @@ -18,6 +18,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import io.agentscope.core.model.transport.HttpTransportException; +import java.net.SocketException; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; @@ -57,6 +59,30 @@ void shouldNotRetryModelHttpExceptionWithoutStatusCode() { assertFalse(ExecutionConfig.RETRYABLE_ERRORS.test(new TestModelHttpException(null))); } + @Test + @DisplayName("Should retry model HTTP exception without status code wrapping transport error") + void shouldRetryModelHttpExceptionWithoutStatusCodeWrappingTransportError() { + // Reproduces issue #3057: streaming transport failures are wrapped by model + // exceptions without an HTTP status code (e.g. OpenAIException), so the cause + // chain must be consulted instead of treating them as permanent client errors. + HttpTransportException transportError = + new HttpTransportException( + "SSE/NDJSON stream failed: Connection reset", + new SocketException("Connection reset")); + TestModelHttpException wrapped = new TestModelHttpException(null, transportError); + + assertTrue(ExecutionConfig.RETRYABLE_ERRORS.test(wrapped)); + } + + @Test + @DisplayName("Should retry model HTTP exception without status code wrapping IO error") + void shouldRetryModelHttpExceptionWithoutStatusCodeWrappingIoError() { + TestModelHttpException wrapped = + new TestModelHttpException(null, new SocketException("Connection reset")); + + assertTrue(ExecutionConfig.RETRYABLE_ERRORS.test(wrapped)); + } + private static final class TestModelHttpException extends RuntimeException implements ModelHttpException { @@ -67,6 +93,11 @@ private TestModelHttpException(Integer statusCode) { this.statusCode = statusCode; } + private TestModelHttpException(Integer statusCode, Throwable cause) { + super("HTTP " + statusCode, cause); + this.statusCode = statusCode; + } + @Override public Integer getStatusCode() { return statusCode; 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..beff260958 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 @@ -20,6 +20,8 @@ import io.agentscope.core.message.Msg; import io.agentscope.core.message.MsgRole; import io.agentscope.core.message.TextBlock; +import io.agentscope.core.model.transport.HttpTransportException; +import java.net.SocketException; import java.time.Duration; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; @@ -239,6 +241,128 @@ void testNoRetryWhenNull() { assertEquals(1, attemptCount.get()); } + @Test + @DisplayName("Should retry connection reset before the first response is emitted") + void shouldRetryConnectionResetBeforeFirstResponseEmitted() { + AtomicInteger attemptCount = new AtomicInteger(0); + + // First attempt resets the connection before emitting anything (issue #3057); + // the retry opens a fresh connection and succeeds. + Flux source = + Flux.defer( + () -> { + if (attemptCount.incrementAndGet() == 1) { + return Flux.error( + new HttpTransportException( + "SSE/NDJSON stream failed: Connection reset", + new SocketException("Connection reset"))); + } + return Flux.just(createMockResponse()); + }); + + ExecutionConfig executionConfig = + ExecutionConfig.builder() + .maxAttempts(3) + .initialBackoff(Duration.ofMillis(10)) + .build(); + GenerateOptions options = + GenerateOptions.builder().executionConfig(executionConfig).build(); + + StepVerifier.create( + ModelUtils.applyTimeoutAndRetry( + source, options, null, "test-model", "test")) + .expectNextCount(1) + .verifyComplete(); + + // Nothing had been emitted when the first attempt failed, so a retry was allowed + assertEquals(2, attemptCount.get()); + } + + @Test + @DisplayName("Should not retry after a response has already been emitted") + void shouldNotRetryAfterResponseEmitted() { + AtomicInteger attemptCount = new AtomicInteger(0); + + // Every attempt emits one response chunk and then the connection resets: + // retrying would reissue the request and duplicate the already-delivered chunk. + Flux source = + Flux.defer( + () -> { + attemptCount.incrementAndGet(); + return Flux.concat( + Flux.just(createMockResponse()), + Flux.error( + new HttpTransportException( + "SSE/NDJSON stream failed: Connection reset", + new SocketException("Connection reset")))); + }); + + ExecutionConfig executionConfig = + ExecutionConfig.builder() + .maxAttempts(3) + .initialBackoff(Duration.ofMillis(10)) + .build(); + GenerateOptions options = + GenerateOptions.builder().executionConfig(executionConfig).build(); + + // The already-emitted chunk is delivered, then the error propagates without retry + StepVerifier.create( + ModelUtils.applyTimeoutAndRetry( + source, options, null, "test-model", "test")) + .expectNextCount(1) + .expectError(HttpTransportException.class) + .verify(); + + assertEquals(1, attemptCount.get()); + } + + @Test + @DisplayName("Should retry after role-only chunks that carry no visible content") + void shouldRetryAfterEmptyContentChunks() { + AtomicInteger attemptCount = new AtomicInteger(0); + + // First attempt emits a role-only chunk (no content blocks) before the connection + // resets: nothing user-visible was delivered, so retrying cannot duplicate content. + Flux source = + Flux.defer( + () -> { + if (attemptCount.incrementAndGet() == 1) { + return Flux.concat( + Flux.just( + new ChatResponse( + "role-only-chunk", + List.of(), + null, + null, + null)), + Flux.error( + new HttpTransportException( + "SSE/NDJSON stream failed: Connection" + + " reset", + new SocketException("Connection reset")))); + } + return Flux.just(createMockResponse()); + }); + + ExecutionConfig executionConfig = + ExecutionConfig.builder() + .maxAttempts(3) + .initialBackoff(Duration.ofMillis(10)) + .build(); + GenerateOptions options = + GenerateOptions.builder().executionConfig(executionConfig).build(); + + StepVerifier.create( + ModelUtils.applyTimeoutAndRetry( + source, options, null, "test-model", "test")) + // role-only chunk of attempt 1 + the full response of attempt 2 + .expectNextCount(2) + .verifyComplete(); + + // The role-only chunk carries no visible content, so a retry was allowed + assertEquals(2, attemptCount.get()); + } + // Helper methods to create test models /** diff --git a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-anthropic/src/test/java/io/agentscope/extensions/model/anthropic/AnthropicRetryClassificationTest.java b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-anthropic/src/test/java/io/agentscope/extensions/model/anthropic/AnthropicRetryClassificationTest.java new file mode 100644 index 0000000000..c72450a982 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-anthropic/src/test/java/io/agentscope/extensions/model/anthropic/AnthropicRetryClassificationTest.java @@ -0,0 +1,73 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed 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 io.agentscope.extensions.model.anthropic; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.model.ExecutionConfig; +import io.agentscope.core.model.transport.HttpTransportException; +import java.net.SocketException; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +/** + * Confirms the retry classification for the exception shapes this module actually produces. + * The anthropic model reports transport failures as {@link HttpTransportException} (it has no + * {@code ModelHttpException} implementation), so its classification is driven by the + * {@code HttpTransportException} branch of {@code ExecutionConfig.isRetryableError} and is + * unaffected by the null-status {@code ModelHttpException} fix — these tests pin that down. + * + * @see io.agentscope.core.model.ExecutionConfig#RETRYABLE_ERRORS + */ +@Tag("unit") +@DisplayName("Anthropic Model Retry Classification Tests") +class AnthropicRetryClassificationTest { + + @Test + @DisplayName("Should retry connection errors without status code") + void shouldRetryConnectionErrorsWithoutStatusCode() { + HttpTransportException connectionError = + new HttpTransportException( + "SSE/NDJSON stream failed: java.net.SocketException: Connection reset", + new SocketException("Connection reset")); + + assertTrue(ExecutionConfig.RETRYABLE_ERRORS.test(connectionError)); + } + + @Test + @DisplayName("Should retry rate limiting and server errors") + void shouldRetryRateLimitingAndServerErrors() { + assertTrue( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Too Many Requests", 429, null))); + assertTrue( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Service Unavailable", 503, null))); + } + + @Test + @DisplayName("Should not retry client errors") + void shouldNotRetryClientErrors() { + assertFalse( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Bad Request", 400, null))); + assertFalse( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Unauthorized", 401, null))); + } +} diff --git a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeRetryClassificationTest.java b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeRetryClassificationTest.java new file mode 100644 index 0000000000..e3ac0d5fc0 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-dashscope/src/test/java/io/agentscope/extensions/model/dashscope/DashScopeRetryClassificationTest.java @@ -0,0 +1,73 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed 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 io.agentscope.extensions.model.dashscope; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.model.ExecutionConfig; +import io.agentscope.core.model.transport.HttpTransportException; +import java.net.SocketException; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +/** + * Confirms the retry classification for the exception shapes this module actually produces. + * The dashscope model reports transport failures as {@link HttpTransportException} (it has no + * {@code ModelHttpException} implementation), so its classification is driven by the + * {@code HttpTransportException} branch of {@code ExecutionConfig.isRetryableError} and is + * unaffected by the null-status {@code ModelHttpException} fix — these tests pin that down. + * + * @see io.agentscope.core.model.ExecutionConfig#RETRYABLE_ERRORS + */ +@Tag("unit") +@DisplayName("DashScope Model Retry Classification Tests") +class DashScopeRetryClassificationTest { + + @Test + @DisplayName("Should retry connection errors without status code") + void shouldRetryConnectionErrorsWithoutStatusCode() { + HttpTransportException connectionError = + new HttpTransportException( + "SSE/NDJSON stream failed: java.net.SocketException: Connection reset", + new SocketException("Connection reset")); + + assertTrue(ExecutionConfig.RETRYABLE_ERRORS.test(connectionError)); + } + + @Test + @DisplayName("Should retry rate limiting and server errors") + void shouldRetryRateLimitingAndServerErrors() { + assertTrue( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Too Many Requests", 429, null))); + assertTrue( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Service Unavailable", 503, null))); + } + + @Test + @DisplayName("Should not retry client errors") + void shouldNotRetryClientErrors() { + assertFalse( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Bad Request", 400, null))); + assertFalse( + ExecutionConfig.RETRYABLE_ERRORS.test( + new HttpTransportException("Unauthorized", 401, null))); + } +} diff --git a/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/exception/OpenAIExceptionRetryClassificationTest.java b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/exception/OpenAIExceptionRetryClassificationTest.java new file mode 100644 index 0000000000..fa9fc31016 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-model/agentscope-extensions-model-openai/src/test/java/io/agentscope/extensions/model/openai/exception/OpenAIExceptionRetryClassificationTest.java @@ -0,0 +1,92 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed 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 io.agentscope.extensions.model.openai.exception; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.model.ExecutionConfig; +import io.agentscope.core.model.transport.HttpTransportException; +import java.net.SocketException; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +/** + * Tests that {@link OpenAIException} participates in the {@link ExecutionConfig#RETRYABLE_ERRORS} + * classification correctly, in particular for streaming transport failures wrapped without an + * HTTP status code (issue #3057). + */ +@Tag("unit") +@DisplayName("OpenAIException Retry Classification Tests") +class OpenAIExceptionRetryClassificationTest { + + @Test + @DisplayName("Should retry streaming transport error wrapped without status code") + void shouldRetryStreamingTransportErrorWrappedWithoutStatusCode() { + // Reproduces the production exception chain from issue #3057: + // OpenAIClient.stream wraps HttpTransportException (no status code) into + // OpenAIException, which implements ModelHttpException with a null status. + HttpTransportException transportError = + new HttpTransportException( + "SSE/NDJSON stream failed: java.net.SocketException: Connection reset", + new SocketException("Connection reset")); + OpenAIException exception = + new OpenAIException( + "HTTP transport error during streaming: " + transportError.getMessage(), + transportError); + + assertTrue(ExecutionConfig.RETRYABLE_ERRORS.test(exception)); + } + + @Test + @DisplayName("Should retry wrapped IO error without status code") + void shouldRetryWrappedIoErrorWithoutStatusCode() { + OpenAIException exception = + new OpenAIException( + "HTTP transport error during streaming", + new SocketException("Connection reset")); + + assertTrue(ExecutionConfig.RETRYABLE_ERRORS.test(exception)); + } + + @Test + @DisplayName("Should keep HTTP status based classification for real responses") + void shouldKeepHttpStatusBasedClassificationForRealResponses() { + // 429/5xx remain retryable + assertTrue( + ExecutionConfig.RETRYABLE_ERRORS.test( + OpenAIException.create(429, "Rate limited", null, null))); + assertTrue( + ExecutionConfig.RETRYABLE_ERRORS.test( + OpenAIException.create(503, "Service unavailable", null, null))); + // Other 4xx remain non-retryable + assertFalse( + ExecutionConfig.RETRYABLE_ERRORS.test( + OpenAIException.create(400, "Bad request", null, null))); + assertFalse( + ExecutionConfig.RETRYABLE_ERRORS.test( + OpenAIException.create(401, "Unauthorized", null, null))); + } + + @Test + @DisplayName("Should not retry unknown error without status code or cause") + void shouldNotRetryUnknownErrorWithoutStatusCodeOrCause() { + OpenAIException exception = new OpenAIException("Something went wrong"); + + assertFalse(ExecutionConfig.RETRYABLE_ERRORS.test(exception)); + } +}