From 178a98209e0482f9152fe75daf80cc7a52480bed Mon Sep 17 00:00:00 2001 From: gxgeek-n <189542381+gxgeek-n@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:58:04 +0800 Subject: [PATCH 1/3] fix(model): stop retrying a streaming response once chunks were delivered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ModelUtils.applyTimeoutAndRetry attaches retryWhen directly to the streaming Flux. retryWhen re-subscribes its upstream, so when a retryable error arrives mid-stream the model regenerates the whole response and downstream consumers (middlewares, memory, UI) receive the already-delivered chunks a second time. With MODEL_DEFAULTS (maxAttempts=3) the same content can be delivered three times. It also inflates latency by a full regeneration, which for long or thinking-mode responses is tens of seconds of apparent silence. Reproduced without a live model — upstream emits two chunks then throws IOException (retryable per ExecutionConfig.RETRYABLE_ERRORS): before: upstream subscribed 3x, downstream got [Hello, " world", Hello, " world", Hello, " world"] after: upstream subscribed 1x, downstream got [Hello, " world"] Fix: track whether anything has been emitted and fold that into the retry filter, so retries stop after the first chunk reaches downstream. Failures *before* the first chunk (connection setup, HTTP 429 ahead of the first token) still retry — that is where retrying is both safe and valuable. Affects the streaming path of every provider routed through this helper: OpenAI, DashScope, Anthropic and Ollama. Adds ModelUtilsStreamingRetryTest covering four cases: mid-stream retryable error, failure before the first chunk, non-retryable mid-stream error, and a clean stream. Only the first fails without this change; the other three pass both ways, pinning that retries are not over-disabled. --- .../io/agentscope/core/model/ModelUtils.java | 14 +- .../model/ModelUtilsStreamingRetryTest.java | 183 ++++++++++++++++++ 2 files changed, 195 insertions(+), 2 deletions(-) create mode 100644 agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java 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..ec4d89a21b 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 @@ -111,11 +111,20 @@ public static Flux applyTimeoutAndRetry( retryOn = error -> true; // retry all errors by default } + // A retry re-subscribes the upstream. For a streaming response that has already + // delivered chunks downstream, re-subscribing makes the model regenerate the whole + // response, so downstream consumers (middlewares, memory, UI) observe the content + // twice. Only retry while nothing has been emitted yet — that still covers the + // valuable cases (connection setup failures, 429 before the first token). + java.util.concurrent.atomic.AtomicBoolean emitted = + new java.util.concurrent.atomic.AtomicBoolean(false); + final Predicate retryableError = retryOn; + Retry retrySpec = Retry.backoff(maxAttempts - 1, initialBackoff) .maxBackoff(maxBackoff) .jitter(0.5) - .filter(retryOn) + .filter(error -> !emitted.get() && retryableError.test(error)) .doBeforeRetry( signal -> LOG.warn( @@ -126,7 +135,8 @@ public static Flux applyTimeoutAndRetry( signal.failure().getMessage(), signal.failure())); - responseFlux = responseFlux.retryWhen(retrySpec); + responseFlux = + responseFlux.doOnNext(response -> emitted.set(true)).retryWhen(retrySpec); LOG.debug( "Applied retry config: maxAttempts={}, initialBackoff={} for model: {}", maxAttempts, diff --git a/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java b/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java new file mode 100644 index 0000000000..defa61b5b5 --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java @@ -0,0 +1,183 @@ +/* + * 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.core.model; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import io.agentscope.core.message.ContentBlock; +import io.agentscope.core.message.TextBlock; +import java.io.IOException; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; + +/** + * Retry semantics for streaming model responses in {@link ModelUtils#applyTimeoutAndRetry}. + * + *

{@code retryWhen} re-subscribes its upstream. For a streaming response that has already handed + * chunks to downstream consumers (middlewares, memory, UI), re-subscribing makes the model + * regenerate the whole response, so the same content is delivered twice — a correctness problem, not + * merely a latency one. It also inflates end-to-end latency by however long a full regeneration + * takes, which is significant for long/thinking-mode responses. + * + *

Retrying is still valuable before the first chunk (connection setup failures, HTTP 429 + * ahead of the first token), so the guard is "stop retrying once anything has been emitted" rather + * than "never retry". + */ +@DisplayName("ModelUtils streaming retry semantics") +class ModelUtilsStreamingRetryTest { + + private static final Duration BLOCK_TIMEOUT = Duration.ofSeconds(5); + + private static ChatResponse chunk(String text) { + return ChatResponse.builder() + .content(List.of(TextBlock.builder().text(text).build())) + .build(); + } + + private static String textOf(ChatResponse response) { + return response.getContent().stream() + .filter(TextBlock.class::isInstance) + .map(TextBlock.class::cast) + .map(TextBlock::getText) + .findFirst() + .orElse(""); + } + + /** Mirrors the retry shape of {@code MODEL_DEFAULTS} with a 1 ms backoff to keep tests fast. */ + private static GenerateOptions retryOptions() { + return GenerateOptions.builder() + .executionConfig( + ExecutionConfig.builder() + .maxAttempts(3) + .initialBackoff(Duration.ofMillis(1)) + .maxBackoff(Duration.ofMillis(2)) + .backoffMultiplier(2.0) + .retryOn(ExecutionConfig.RETRYABLE_ERRORS) + .build()) + .build(); + } + + private static List collect(Flux upstream, GenerateOptions options) { + List delivered = + ModelUtils.applyTimeoutAndRetry(upstream, options, options, "test-model", "test") + .map(ModelUtilsStreamingRetryTest::textOf) + .onErrorResume(error -> Flux.empty()) + .collectList() + .block(BLOCK_TIMEOUT); + assertNotNull(delivered); + return delivered; + } + + @Test + @DisplayName("mid-stream retryable error: already-delivered chunks are not replayed") + void midStreamRetryableError_doesNotDuplicateDeliveredChunks() { + AtomicInteger subscriptions = new AtomicInteger(); + + // A model streams part of its answer, then the connection drops. IOException is classified + // retryable by ExecutionConfig.RETRYABLE_ERRORS. + Flux upstream = + Flux.defer( + () -> { + subscriptions.incrementAndGet(); + return Flux.concat( + Flux.just(chunk("Hello"), chunk(" world")), + Flux.error(new IOException("connection reset by peer"))); + }); + + List delivered = collect(upstream, retryOptions()); + + assertEquals( + List.of("Hello", " world"), + delivered, + "chunks already delivered downstream must not be replayed — a retry here makes the" + + " model regenerate the whole response and the user sees duplicated" + + " content"); + assertEquals( + 1, + subscriptions.get(), + "upstream must not be re-subscribed after chunks were emitted"); + } + + @Test + @DisplayName("failure before the first chunk: retry still happens") + void failureBeforeFirstChunk_stillRetries() { + AtomicInteger subscriptions = new AtomicInteger(); + + // Fails on the first two attempts before emitting anything, then succeeds. Nothing has + // reached downstream yet, so retrying is safe and desirable. + Flux upstream = + Flux.defer( + () -> { + int attempt = subscriptions.incrementAndGet(); + if (attempt < 3) { + return Flux.error(new IOException("connect timed out")); + } + return Flux.just(chunk("recovered")); + }); + + List delivered = collect(upstream, retryOptions()); + + assertEquals( + List.of("recovered"), + delivered, + "a failure before the first chunk must still be retried"); + assertEquals(3, subscriptions.get(), "expected two retries before the successful attempt"); + } + + @Test + @DisplayName("non-retryable mid-stream error: no retry, delivered chunks kept") + void nonRetryableMidStreamError_noRetry() { + AtomicInteger subscriptions = new AtomicInteger(); + + // IllegalStateException is not in RETRYABLE_ERRORS. + Flux upstream = + Flux.defer( + () -> { + subscriptions.incrementAndGet(); + return Flux.concat( + Flux.just(chunk("partial")), + Flux.error(new IllegalStateException("bad response shape"))); + }); + + List delivered = collect(upstream, retryOptions()); + + assertEquals(List.of("partial"), delivered); + assertEquals(1, subscriptions.get(), "non-retryable errors must not trigger a retry"); + } + + @Test + @DisplayName("clean stream: retry config does not alter delivery") + void cleanStream_unaffectedByRetryConfig() { + AtomicInteger subscriptions = new AtomicInteger(); + + Flux upstream = + Flux.defer( + () -> { + subscriptions.incrementAndGet(); + return Flux.just(chunk("a"), chunk("b"), chunk("c")); + }); + + List delivered = collect(upstream, retryOptions()); + + assertEquals(List.of("a", "b", "c"), delivered); + assertEquals(1, subscriptions.get()); + } +} From 0004c04e8ba6968476d3f4c8500238c301eb6479 Mon Sep 17 00:00:00 2001 From: gxgeek-n <189542381+gxgeek-n@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:25:40 +0800 Subject: [PATCH 2/3] fix(model): make the streaming retry guard per-subscription Review feedback: the emitted flag and the Retry instance were created once per applyTimeoutAndRetry call and captured by the returned Flux. A Flux may be subscribed more than once, so a second subscriber could inherit emitted=true from an earlier subscription and skip retries it should have performed. Moved the retry wiring inside Flux.defer so both the flag and the Retry spec are allocated per subscription. Adds emittedFlagIsPerSubscription: the same wrapped Flux is subscribed twice, each subscription emitting a chunk before a retryable failure. Shared state would make the second subscription diverge from the first. ModelUtilsStreamingRetryTest: 5 tests, 0 failures. --- .../io/agentscope/core/model/ModelUtils.java | 57 +++++++++++++------ .../model/ModelUtilsStreamingRetryTest.java | 35 ++++++++++++ 2 files changed, 74 insertions(+), 18 deletions(-) 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 ec4d89a21b..ad1f4dc6d9 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 @@ -116,27 +116,48 @@ public static Flux applyTimeoutAndRetry( // response, so downstream consumers (middlewares, memory, UI) observe the content // twice. Only retry while nothing has been emitted yet — that still covers the // valuable cases (connection setup failures, 429 before the first token). - java.util.concurrent.atomic.AtomicBoolean emitted = - new java.util.concurrent.atomic.AtomicBoolean(false); + // + // The flag lives inside Flux.defer so it is per-subscription: a Flux may be + // subscribed more than once, and a flag shared across subscriptions would let a + // later subscriber inherit "already emitted" from an earlier one and skip retries + // it should have performed. final Predicate retryableError = retryOn; - - Retry retrySpec = - Retry.backoff(maxAttempts - 1, initialBackoff) - .maxBackoff(maxBackoff) - .jitter(0.5) - .filter(error -> !emitted.get() && retryableError.test(error)) - .doBeforeRetry( - signal -> - LOG.warn( - "Retrying model request (attempt {}/{}) due" - + " to: {}", - signal.totalRetriesInARow() + 1, - maxAttempts - 1, - signal.failure().getMessage(), - signal.failure())); + final Duration retryInitialBackoff = initialBackoff; + final Duration retryMaxBackoff = maxBackoff; + final int retryMaxAttempts = maxAttempts; + final Flux retrySource = responseFlux; responseFlux = - responseFlux.doOnNext(response -> emitted.set(true)).retryWhen(retrySpec); + Flux.defer( + () -> { + java.util.concurrent.atomic.AtomicBoolean emitted = + new java.util.concurrent.atomic.AtomicBoolean(false); + Retry retrySpec = + Retry.backoff(retryMaxAttempts - 1, retryInitialBackoff) + .maxBackoff(retryMaxBackoff) + .jitter(0.5) + .filter( + error -> + !emitted.get() + && retryableError.test( + error)) + .doBeforeRetry( + signal -> + LOG.warn( + "Retrying model request" + + " (attempt {}/{})" + + " due to: {}", + signal + .totalRetriesInARow() + + 1, + retryMaxAttempts - 1, + signal.failure() + .getMessage(), + signal.failure())); + return retrySource + .doOnNext(response -> emitted.set(true)) + .retryWhen(retrySpec); + }); LOG.debug( "Applied retry config: maxAttempts={}, initialBackoff={} for model: {}", maxAttempts, diff --git a/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java b/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java index defa61b5b5..b790278d05 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java @@ -180,4 +180,39 @@ void cleanStream_unaffectedByRetryConfig() { assertEquals(List.of("a", "b", "c"), delivered); assertEquals(1, subscriptions.get()); } + + @Test + @DisplayName("emitted state is per-subscription, not shared across subscribers") + void emittedFlagIsPerSubscription() { + // Every subscription emits one chunk and then fails with a retryable error. The first + // subscription therefore ends with "something was emitted". If that state were shared + // across subscriptions (allocated once outside Flux.defer), the second subscriber would + // inherit it and behave differently from the first. + Flux upstream = + Flux.concat( + Flux.just(chunk("chunk")), + Flux.error(new IOException("connection reset by peer"))); + + GenerateOptions options = retryOptions(); + Flux wrapped = + ModelUtils.applyTimeoutAndRetry(upstream, options, options, "test-model", "test"); + + List first = + wrapped.map(ModelUtilsStreamingRetryTest::textOf) + .onErrorResume(error -> Flux.empty()) + .collectList() + .block(BLOCK_TIMEOUT); + List second = + wrapped.map(ModelUtilsStreamingRetryTest::textOf) + .onErrorResume(error -> Flux.empty()) + .collectList() + .block(BLOCK_TIMEOUT); + + assertEquals(List.of("chunk"), first, "first subscription should deliver one chunk"); + assertEquals( + first, + second, + "a second subscription must behave identically — retry state leaking across" + + " subscribers would make it diverge from the first"); + } } From 668a7fa851f8aafa4213856f84c7c088f778297c Mon Sep 17 00:00:00 2001 From: gxgeek-n <189542381+gxgeek-n@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:45:44 +0800 Subject: [PATCH 3/3] test(model): cover the non-retryable-before-first-chunk branch Codecov flagged one partial branch on the retry filter `!emitted.get() && retryableError.test(error)`. The existing non-retryable case emits a chunk first, so the emitted guard short-circuits and the predicate's false path was never taken. Adds a case that fails with a non-retryable error before emitting anything, so both operands of the condition are exercised in both directions. 6 tests, 0 failures. --- .../model/ModelUtilsStreamingRetryTest.java | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java b/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java index b790278d05..1f52a82b4d 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/model/ModelUtilsStreamingRetryTest.java @@ -163,6 +163,30 @@ void nonRetryableMidStreamError_noRetry() { assertEquals(1, subscriptions.get(), "non-retryable errors must not trigger a retry"); } + @Test + @DisplayName("non-retryable error before the first chunk: no retry") + void nonRetryableErrorBeforeFirstChunk_noRetry() { + AtomicInteger subscriptions = new AtomicInteger(); + + // Nothing emitted yet, so the "already emitted" guard does not short-circuit and the + // retryOn predicate is what has to reject this error. Completes the branch matrix of + // `!emitted.get() && retryableError.test(error)`. + Flux upstream = + Flux.defer( + () -> { + subscriptions.incrementAndGet(); + return Flux.error(new IllegalStateException("bad request shape")); + }); + + List delivered = collect(upstream, retryOptions()); + + assertEquals(List.of(), delivered, "a non-retryable failure yields nothing"); + assertEquals( + 1, + subscriptions.get(), + "a non-retryable error must not be retried even when nothing was emitted yet"); + } + @Test @DisplayName("clean stream: retry config does not alter delivery") void cleanStream_unaffectedByRetryConfig() {