Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wahllllll wants to merge 3 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
521109d
fix(core): retry streaming transport errors wrapped by ModelHttpExcep…
wahllllll Sep 10, 2026
9d6b8b5
fix(core): restrict automatic model retries to before the first strea…
wahllllll Sep 10, 2026
826495a
fix(core): gate model retries on visible content, not any emitted chunk
wahllllll Sep 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

@oss-maintainer oss-maintainer Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good fix — the status-code branch no longer short-circuits the cause-chain checks when getStatusCode() is null, and shouldNotRetryModelHttpExceptionWithoutStatusCode keeps the old contract. Since this changes retry behaviour for every provider implementing ModelHttpException, please confirm the non-streaming 4xx paths still land in isRetryableHttpStatus() in the extension-module tests (openai/anthropic/dashscope), not just the openai one.

@wahllllll wahllllll Sep 12, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Checked the other model extensions: only the openai module has a ModelHttpException implementation (OpenAIException). The anthropic and dashscope models surface transport failures as plain HttpTransportException (no status code for connection errors, 4xx/5xx with status codes), so their classification goes through the HttpTransportException branch of isRetryableError and is unchanged by this fix. To pin that down, 826495a adds AnthropicRetryClassificationTest and DashScopeRetryClassificationTest asserting: connection error without status → retryable, 429/5xx → retryable, 400/401 → not retryable.

// 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) {

@dailingtao dailingtao Sep 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Please avoid retrying after a streaming response has already emitted any chunks. With this change, a null-status ModelHttpException can fall through to a retryable transport/IO cause regardless of stream progress. ModelUtils applies retryWhen to the entire response Flux, so if the first attempt emits partial text or tool-call deltas and then the connection resets, resubscription reissues the request and downstream observes the first partial output followed by the second response, which can duplicate text or tool calls. AgentScope Python and the OpenAI Python SDK limit automatic retries to establishing the request/stream rather than consuming an already-started stream. Could we guard retries so they are allowed only before the first ChatResponse is emitted, and add tests covering both a reset before the first chunk (retries) and a reset after one chunk (does not retry)?

@wahllllll wahllllll Sep 10, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the careful review — you're right, resubscribing after partial output would duplicate already-delivered chunks downstream.

I've pushed a guard in ModelUtils.applyTimeoutAndRetry: retries are now allowed only while the current subscription has not emitted any ChatResponse yet. The emission flag is created inside Flux.defer so each model call gets a fresh flag, while it persists across the retry attempts of the same call (once any attempt has emitted, further retries are rejected regardless of the error classification). All streaming models routed through applyTimeoutAndRetry benefit.

Added the two tests you asked for in ModelTimeoutRetryTest, both using the exact exception chain from the issue (HttpTransportException with null status wrapping SocketException):

  • reset before the first chunk → retried, second attempt succeeds
  • reset after one emitted chunk → the chunk is delivered, the error propagates, no retry

Commit history was rewritten to keep the two changes as separate commits.

return mhe.isRetryableHttpStatus();
}

Expand Down
114 changes: 97 additions & 17 deletions agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -110,25 +113,88 @@ 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 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<ChatResponse> 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(

@oss-maintainer oss-maintainer Sep 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The emission guard is the right call — retrying after partial output would duplicate content. But emittedAnyResponse is set by any ChatResponse, including providers that emit an initial empty/role-only chunk before real deltas. For those streams the guard disables retries for the whole call, which is exactly the case this PR fixes. Consider gating on "emitted user-visible content" (non-empty content blocks) instead of any response, or at least logging once when a retry is suppressed by the guard so this is diagnosable in production.

@wahllllll wahllllll Sep 12, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch — role-only/usage-only chunks do map to ChatResponses with an empty content list (verified in OpenAIResponseParser.parseChunkResponse: text/thinking/tool-call deltas all land in content blocks, a role-only delta produces none). Pushed in 826495a:

  • the guard flag is now set only by chunks that carry content blocks (user-visible text/thinking/tool-call deltas), so an early reset after a role-only chunk is still retried;
  • when the guard suppresses an otherwise-retryable error, a single WARN is logged with the model name and error so it is diagnosable in production;
  • added shouldRetryAfterEmptyContentChunks covering exactly the shape you described: role-only chunk → connection reset → retried, second attempt succeeds.

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);
Expand All @@ -138,6 +204,20 @@ public static Flux<ChatResponse> applyTimeoutAndRetry(
return responseFlux;
}

/**
* Whether the response carries user-visible content (text, thinking or tool-call deltas).
*
* <p>Role-only or usage-only chunks carry no content blocks — nothing user-visible has
* been delivered, so retrying past them cannot duplicate content.</p>
*
* @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<ContentBlock> content = response.getContent();
return content != null && !content.isEmpty();
}

/**
* Ensures GenerateOptions has MODEL_DEFAULTS for executionConfig applied.
*
Expand Down
31 changes: 31 additions & 0 deletions agentscope-core/src/test/java/io/agentscope/core/model/ExecutionConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 {

Expand All @@ -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;
Expand Down
124 changes: 124 additions & 0 deletions agentscope-core/src/test/java/io/agentscope/core/model/ModelTimeoutRetryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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());
}

@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<ChatResponse> 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

/**
Expand Down
Loading
Loading