fix(core): retry streaming transport errors wrapped by ModelHttpException without status code

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();
         }
 

agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java

@@ -16,6 +16,7 @@
 package io.agentscope.core.model;
 
 import java.time.Duration;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -110,25 +111,57 @@ public static Flux<ChatResponse> 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<Throwable> 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 the first response 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 emission flag is created inside defer so every subscription (model call)
+                // gets a fresh flag, while it persists across retry attempts of the same call.
+                final Flux<ChatResponse> source = responseFlux;
+                responseFlux =
+                        Flux.defer(
+                                () -> {
+                                    AtomicBoolean emittedAnyResponse = new AtomicBoolean(false);
+                                    return source.doOnNext(response -> emittedAnyResponse.set(true))
+                                            .retryWhen(
+                                                    Retry.backoff(
+                                                                    retryCount,
+                                                                    effectiveInitialBackoff)
+                                                            .maxBackoff(effectiveMaxBackoff)
+                                                            .jitter(0.5)
+                                                            .filter(
+                                                                    error ->
+                                                                            !emittedAnyResponse
+                                                                                            .get()
+                                                                                    && 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);

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;

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,81 @@ 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<ChatResponse> 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<ChatResponse> 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());
+    }
+
     // Helper methods to create test models
 
     /**

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));
+    }
+}