From a97469c400ae39d4c88384581364a44981743ffb Mon Sep 17 00:00:00 2001 From: "jixing.xia" Date: Wed, 28 Oct 2026 14:44:51 +0800 Subject: [PATCH] fix(spring-ai): emit final response on stream completion instead of punctuation heuristic Streaming chunks were classified as partial/final by sentence-punctuation heuristics. Turns ending with CJK terminal punctuation were never marked final, so the response was silently not persisted to the session; a '.' inside a longer answer could fire a premature final mid-stream. Streamed chunks are now always partial; StreamingResponseAggregator produces the single aggregated final response on stream completion, carrying full text, accumulated tool calls, and the stream's last usage metadata (previously lost with the unpersisted partial events). An empty stream still emits one final event, so BaseLlmFlow does not re-call the model until maxLlmCalls. Rebased onto the new candidate-merging MessageConverter and rebased tests accordingly. Fixes #1608 --- .../adk/models/springai/MessageConverter.java | 17 +- .../google/adk/models/springai/SpringAI.java | 10 + .../springai/StreamingResponseAggregator.java | 129 ++++++++-- .../models/springai/MessageConverterTest.java | 15 +- .../adk/models/springai/SpringAITest.java | 225 +++++++++++++++++- 5 files changed, 353 insertions(+), 43 deletions(-) diff --git a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java index 71ef570d8..e59b7a6aa 100644 --- a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java +++ b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java @@ -45,7 +45,6 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; -import java.util.stream.Collectors; import org.jspecify.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -321,7 +320,8 @@ public LlmResponse toLlmResponse(@Nullable ChatResponse chatResponse, boolean is } Content content = Content.builder().role("model").parts(mergedParts).build(); - boolean isPartial = isStreaming && isPartialResponse(mergedParts); + // Streamed chunks are partial; SpringAI emits the aggregated final on stream completion. + boolean isPartial = isStreaming; boolean isTurnComplete = !isStreaming || isTurnCompleteFinishReason(firstFinishReason); responseBuilder.content(content).partial(isPartial).turnComplete(isTurnComplete); @@ -680,19 +680,6 @@ private int nullSafeInt(@Nullable Integer value) { return value != null ? value.intValue() : 0; } - /** - * Determines if the merged parts of a streamed candidate are partial: they carry no function call - * and their text does not end with sentence punctuation. - */ - private static boolean isPartialResponse(List parts) { - if (parts.stream().anyMatch(part -> part.functionCall().isPresent())) { - return false; - } - String text = - parts.stream().map(part -> part.text().orElse("")).collect(Collectors.joining()).trim(); - return !text.isEmpty() && !text.endsWith(".") && !text.endsWith("!") && !text.endsWith("?"); - } - /** Determines if a candidate's finish reason indicates the turn is complete. */ private boolean isTurnCompleteFinishReason(@Nullable String finishReason) { if (finishReason == null) { diff --git a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java index ece9b1e2f..5767ad510 100644 --- a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java +++ b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java @@ -204,6 +204,10 @@ private Flowable generateStreamingContent(LlmRequest llmRequest) { Prompt prompt = messageConverter.toLlmPrompt(llmRequest, resolveDefaultOptions()); observabilityHandler.logRequest(prompt.toString(), model()); + // Per-stream aggregator: accumulates chunk text / tool calls and produces the + // single final response on stream completion. + StreamingResponseAggregator aggregator = new StreamingResponseAggregator(); + Flux responseFlux = streamingChatModel.stream(prompt); responseFlux @@ -221,6 +225,7 @@ private Flowable generateStreamingContent(LlmRequest llmRequest) { // Use enhanced streaming-aware conversion LlmResponse llmResponse = messageConverter.toLlmResponse(chatResponse, true); + aggregator.processStreamingResponse(llmResponse); emitter.onNext(llmResponse); } catch (Exception e) { observabilityHandler.recordError(context, e); @@ -240,6 +245,11 @@ private Flowable generateStreamingContent(LlmRequest llmRequest) { () -> { // Record success for streaming completion observabilityHandler.recordSuccess(context, 0, 0, 0); + // The final (persisted) response is emitted exactly once, on stream + // completion. Always emit one final response, even an empty one: without + // it every event is partial, and BaseLlmFlow treats the turn as + // unfinished and re-calls the model until maxLlmCalls. + emitter.onNext(aggregator.getFinalResponse()); emitter.onComplete(); }); } catch (Exception e) { diff --git a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java index da0710493..d7a272557 100644 --- a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java +++ b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java @@ -17,6 +17,8 @@ import com.google.adk.models.LlmResponse; import com.google.genai.types.Content; +import com.google.genai.types.FinishReason; +import com.google.genai.types.GenerateContentResponseUsageMetadata; import com.google.genai.types.Part; import java.util.ArrayList; import java.util.List; @@ -36,11 +38,32 @@ */ public class StreamingResponseAggregator { - private final StringBuffer textAccumulator = new StringBuffer(); + private final List textParts = new CopyOnWriteArrayList<>(); + + /** Tracks the thought flag of the part currently being accumulated (Kotlin parity). */ + private Boolean currentThoughtFlag = null; + + /** + * Latest non-empty usage metadata seen on the stream, carried on the final response. Visibility + * is guaranteed by {@code lock}: written in {@code processStreamingResponse} and read in {@code + * getFinalResponse()}, both under the write lock. + */ + private GenerateContentResponseUsageMetadata usageMetadata; + private final List toolCallParts = new CopyOnWriteArrayList<>(); + + /** Non-text, non-function-call parts (inlineData/fileData) passed through as-is. */ + private final List otherParts = new CopyOnWriteArrayList<>(); + private final ReadWriteLock lock = new ReentrantReadWriteLock(); private volatile boolean isFirstResponse = true; + /** Last finish reason seen on the stream; carried on the final, non-STOP becomes an error. */ + private FinishReason finishReason; + + private String errorMessage; + private String modelVersion; + /** * Processes a streaming LlmResponse and returns the current aggregated state. * @@ -59,21 +82,36 @@ public LlmResponse processStreamingResponse(LlmResponse response) { lock.writeLock().lock(); try { - // Process each part in the response + response.usageMetadata().ifPresent(usage -> this.usageMetadata = usage); + response.finishReason().ifPresent(reason -> this.finishReason = reason); + response.errorMessage().ifPresent(message -> this.errorMessage = message); + response.modelVersion().ifPresent(version -> this.modelVersion = version); + + // Process each part in the response. Text is grouped into parts by its thought flag + // (Kotlin parity): reasoning and answer must not merge into one part, or reasoning is + // persisted as answer text and replayed to the model on the next turn. for (Part part : content.parts().get()) { - if (part.text().isPresent()) { - textAccumulator.append(part.text().get()); - } else if (part.functionCall().isPresent()) { + if (part.functionCall().isPresent()) { // Tool calls are typically complete in each response toolCallParts.add(part); + } else if (part.text().isPresent()) { + boolean thought = part.thought().orElse(false); + if (currentThoughtFlag != null && currentThoughtFlag == thought && !textParts.isEmpty()) { + Part last = textParts.get(textParts.size() - 1); + textParts.set( + textParts.size() - 1, withText(last, last.text().orElse("") + part.text().get())); + } else { + textParts.add(withThought(Part.fromText(part.text().get()), thought)); + currentThoughtFlag = thought; + } + } else { + // inlineData/fileData and other non-text parts pass through as-is + otherParts.add(part); } } // Create aggregated content - List aggregatedParts = new ArrayList<>(); - if (textAccumulator.length() > 0) { - aggregatedParts.add(Part.fromText(textAccumulator.toString())); - } + List aggregatedParts = new ArrayList<>(textParts); aggregatedParts.addAll(toolCallParts); Content aggregatedContent = Content.builder().role("model").parts(aggregatedParts).build(); @@ -104,20 +142,42 @@ public LlmResponse processStreamingResponse(LlmResponse response) { public LlmResponse getFinalResponse() { lock.writeLock().lock(); try { - List finalParts = new ArrayList<>(); - if (textAccumulator.length() > 0) { - finalParts.add(Part.fromText(textAccumulator.toString())); - } + List finalParts = new ArrayList<>(textParts); finalParts.addAll(toolCallParts); + finalParts.addAll(otherParts); Content finalContent = Content.builder().role("model").parts(finalParts).build(); - LlmResponse finalResponse = - LlmResponse.builder().content(finalContent).partial(false).turnComplete(true).build(); + LlmResponse.Builder finalBuilder = + LlmResponse.builder().content(finalContent).partial(false).turnComplete(true); + // Carry the last finish reason: a non-STOP stream is an error turn, not a clean one + // (Kotlin aggregator / Python LiteLLM parity). + if (finishReason != null) { + finalBuilder.finishReason(finishReason); + if (finishReason.knownEnum() != FinishReason.Known.STOP) { + finalBuilder.errorCode(finishReason); + if (errorMessage == null && finishReason.knownEnum() == FinishReason.Known.MAX_TOKENS) { + errorMessage = "The generation ran out of tokens before completing."; + } + } + } + if (errorMessage != null) { + finalBuilder.errorMessage(errorMessage); + } + if (modelVersion != null) { + finalBuilder.modelVersion(modelVersion); + } + LlmResponse finalResponse = finalBuilder.usageMetadata(usageMetadata).build(); // Reset internal state without calling reset() to avoid nested locking - textAccumulator.setLength(0); + textParts.clear(); + currentThoughtFlag = null; + usageMetadata = null; toolCallParts.clear(); + otherParts.clear(); + finishReason = null; + errorMessage = null; + modelVersion = null; isFirstResponse = true; return finalResponse; @@ -130,8 +190,14 @@ public LlmResponse getFinalResponse() { public void reset() { lock.writeLock().lock(); try { - textAccumulator.setLength(0); + textParts.clear(); + currentThoughtFlag = null; + usageMetadata = null; toolCallParts.clear(); + otherParts.clear(); + finishReason = null; + errorMessage = null; + modelVersion = null; isFirstResponse = true; } finally { lock.writeLock().unlock(); @@ -142,7 +208,7 @@ public void reset() { public boolean isEmpty() { lock.readLock().lock(); try { - return textAccumulator.length() == 0 && toolCallParts.isEmpty(); + return textParts.isEmpty() && toolCallParts.isEmpty() && otherParts.isEmpty(); } finally { lock.readLock().unlock(); } @@ -152,9 +218,34 @@ public boolean isEmpty() { public int getAccumulatedTextLength() { lock.readLock().lock(); try { - return textAccumulator.length(); + return textParts.stream().mapToInt(part -> part.text().map(String::length).orElse(0)).sum(); } finally { lock.readLock().unlock(); } } + + /** Replaces a part's text, preserving its thought flag and signature. */ + private static Part withText(Part part, String text) { + Part.Builder builder = Part.builder().text(text); + if (part.thought().isPresent()) { + builder.thought(part.thought().get()); + } + if (part.thoughtSignature().isPresent()) { + builder.thoughtSignature(part.thoughtSignature().get()); + } + return builder.build(); + } + + /** Returns the part with the thought flag set (Kotlin parity: reasoning segments). */ + private static Part withThought(Part part, boolean thought) { + if (!thought) { + return part; + } + Part.Builder builder = Part.builder().text(part.text().orElse("")); + builder.thought(true); + if (part.thoughtSignature().isPresent()) { + builder.thoughtSignature(part.thoughtSignature().get()); + } + return builder.build(); + } } diff --git a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java index 404e1bace..7312c7cec 100644 --- a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java +++ b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java @@ -532,8 +532,10 @@ void testToLlmResponseNonStreamingMode() { } @Test - void testPartialResponseDetection() { - // Test partial response (no punctuation ending) + void testToLlmResponseStreamingAlwaysMarksChunksPartial() { + // Streamed chunks are always partial regardless of punctuation: the final (persisted) + // response is derived from the stream completion signal, not mid-stream guessing. + // Punctuation-based classification dropped turns ending with CJK terminal punctuation. AssistantMessage partialMessage = new AssistantMessage("I am thinking"); Generation partialGeneration = new Generation(partialMessage); ChatResponse partialResponse = new ChatResponse(List.of(partialGeneration)); @@ -541,13 +543,12 @@ void testPartialResponseDetection() { LlmResponse partialLlmResponse = messageConverter.toLlmResponse(partialResponse, true); assertThat(partialLlmResponse.partial()).contains(true); - // Test complete response (ends with punctuation) AssistantMessage completeMessage = new AssistantMessage("I am done."); Generation completeGeneration = new Generation(completeMessage); ChatResponse completeResponse = new ChatResponse(List.of(completeGeneration)); LlmResponse completeLlmResponse = messageConverter.toLlmResponse(completeResponse, true); - assertThat(completeLlmResponse.partial()).contains(false); + assertThat(completeLlmResponse.partial()).contains(true); } @Test @@ -1068,7 +1069,8 @@ void toLlmResponse_streamingThoughtThenToolCall_isCompleteNotPartial() { LlmResponse response = messageConverter.toLlmResponse(new ChatResponse(List.of(thoughtGen, toolCallGen)), true); - assertThat(response.partial()).contains(false); + // Streamed chunks are always partial; the aggregated final is emitted on completion. + assertThat(response.partial()).contains(true); assertThat(response.turnComplete()).contains(true); assertThat(response.finishReason()).contains(new FinishReason(FinishReason.Known.STOP)); } @@ -1095,7 +1097,8 @@ void toLlmResponse_streamingGenerationWithoutOutput_doesNotThrow() { LlmResponse response = messageConverter.toLlmResponse(new ChatResponse(List.of(new Generation(null))), true); - assertThat(response.partial()).contains(false); + // Streamed chunks are always partial; the aggregated final is emitted on completion. + assertThat(response.partial()).contains(true); assertThat(response.content().orElseThrow().parts().orElseThrow()).isEmpty(); } diff --git a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java index 51fe60abb..b4674fcef 100644 --- a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java +++ b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java @@ -24,14 +24,19 @@ import com.google.adk.models.LlmRequest; import com.google.adk.models.LlmResponse; import com.google.genai.types.Content; +import com.google.genai.types.FinishReason; import com.google.genai.types.Part; import io.reactivex.rxjava3.subscribers.TestSubscriber; import java.time.Duration; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.metadata.ChatGenerationMetadata; +import org.springframework.ai.chat.metadata.ChatResponseMetadata; +import org.springframework.ai.chat.metadata.DefaultUsage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.model.Generation; @@ -149,16 +154,229 @@ void testGenerateContentStreaming() { testObserver.awaitDone(5, TimeUnit.SECONDS); testObserver.assertComplete(); testObserver.assertNoErrors(); - testObserver.assertValueCount(3); + // 3 partial chunks + 1 aggregated final response on stream completion + testObserver.assertValueCount(4); List responses = testObserver.values(); - assertThat(responses).hasSize(3); + assertThat(responses).hasSize(4); // Verify each streaming response for (LlmResponse response : responses) { assertThat(response.content()).isPresent(); assertThat(response.content().get().parts()).isPresent(); } + + // The first 3 values are partial chunks; the last value is the aggregated final response + for (int i = 0; i < 3; i++) { + assertThat(responses.get(i).partial()).contains(true); + } + LlmResponse finalResponse = responses.get(3); + assertThat(finalResponse.partial()).contains(false); + assertThat(finalResponse.turnComplete()).contains(true); + assertThat(finalResponse.content().get().parts().get().get(0).text()) + .contains("I'm doing well!"); + } + + @Test + void testGenerateContentStreamingFinalCarriesUsageMetadata() { + // Usage metadata from the stream's last chunk must survive onto the aggregated final + // response (partial events are streamed but never persisted by the Runner). + AssistantMessage chunk1 = new AssistantMessage("Hello"); + AssistantMessage chunk2 = new AssistantMessage(" world"); + ChatResponse lastResponse = + new ChatResponse( + List.of(new Generation(chunk2)), + ChatResponseMetadata.builder() + .id("resp-usage") + .usage(new DefaultUsage(10, 4, 6)) + .build()); + Flux responseFlux = + Flux.just(new ChatResponse(List.of(new Generation(chunk1))), lastResponse); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + testObserver.assertValueCount(3); + + LlmResponse finalResponse = testObserver.values().get(2); + assertThat(finalResponse.partial()).contains(false); + // Assert the mapped counts (not just presence) to catch a mapping regression. + assertThat(finalResponse.usageMetadata()).isPresent(); + assertThat(finalResponse.usageMetadata().get().promptTokenCount()).contains(10); + assertThat(finalResponse.usageMetadata().get().candidatesTokenCount()).contains(4); + assertThat(finalResponse.usageMetadata().get().totalTokenCount()).contains(6); + } + + @Test + void testGenerateContentStreamingEmptyChunksEmitSingleFinal() { + // A stream of empty chunks must still yield exactly one partial=false response: + // without a final event the turn looks unfinished and BaseLlmFlow re-calls the + // model with the same history until maxLlmCalls. + AssistantMessage emptyChunk = new AssistantMessage(""); + Flux responseFlux = + Flux.just( + new ChatResponse(List.of(new Generation(emptyChunk))), + new ChatResponse(List.of(new Generation(emptyChunk)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + // 2 partial chunks + exactly 1 final + testObserver.assertValueCount(3); + List responses = testObserver.values(); + for (int i = 0; i < 2; i++) { + assertThat(responses.get(i).partial()).contains(true); + } + assertThat(responses.get(2).partial()).contains(false); + assertThat(responses.get(2).turnComplete()).contains(true); + } + + @Test + void testGenerateContentStreamingThoughtThenAnswerKeptAsSeparateParts() { + // Reasoning must stay thought-flagged parts separate from the answer (Kotlin parity): + // merged into one text part, reasoning is persisted as answer text and replayed to the + // model on the next turn. + AssistantMessage thoughtChunk = + AssistantMessage.builder() + .content("reasoning step") + .properties(Map.of("isThought", true)) + .build(); + AssistantMessage answerChunk = new AssistantMessage("The answer."); + Flux responseFlux = + Flux.just( + new ChatResponse(List.of(new Generation(thoughtChunk))), + new ChatResponse(List.of(new Generation(answerChunk)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + + LlmResponse finalResponse = testObserver.values().get(testObserver.values().size() - 1); + List finalParts = finalResponse.content().get().parts().get(); + // two separate text parts: thought segment first, answer second + assertThat(finalParts.size()).isGreaterThanOrEqualTo(2); + assertThat(finalParts.get(0).thought()).contains(true); + assertThat(finalParts.get(0).text()).contains("reasoning step"); + assertThat(finalParts.get(1).thought()).isEmpty(); + assertThat(finalParts.get(1).text()).contains("The answer."); + } + + @Test + void testGenerateContentStreamingMaxTokensCarriesErrorOnFinal() { + // A MAX_TOKENS stream must not be persisted as a clean turn: the final carries the + // finish reason and the error code (Kotlin aggregator / Python LiteLLM parity). + AssistantMessage chunk = new AssistantMessage("partial answer"); + ChatGenerationMetadata maxTokens = + ChatGenerationMetadata.builder().finishReason("LENGTH").build(); + Flux responseFlux = + Flux.just(new ChatResponse(List.of(new Generation(chunk, maxTokens)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + + LlmResponse finalResponse = testObserver.values().get(testObserver.values().size() - 1); + assertThat(finalResponse.partial()).contains(false); + assertThat(finalResponse.finishReason()).isPresent(); + assertThat(finalResponse.finishReason().get().knownEnum()) + .isEqualTo(FinishReason.Known.MAX_TOKENS); + assertThat(finalResponse.errorCode()).isPresent(); + } + + @Test + void testGenerateContentStreamingMixedTextAndToolCallTurn() { + // A text + tool-call turn: the aggregated final must carry both the full text and the + // accumulated function call (checked by review), with the tool call intact. + AssistantMessage textChunk = new AssistantMessage("Checking "); + AssistantMessage textChunk2 = new AssistantMessage("the logs now."); + AssistantMessage toolCallChunk = + AssistantMessage.builder() + .content("") + .toolCalls( + List.of( + new AssistantMessage.ToolCall( + "call-1", "function", "queryLog", "{\"kw\":\"x\"}"))) + .build(); + Flux responseFlux = + Flux.just( + new ChatResponse(List.of(new Generation(textChunk))), + new ChatResponse(List.of(new Generation(textChunk2))), + new ChatResponse(List.of(new Generation(toolCallChunk)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + // 3 partial chunks + 1 aggregated final + testObserver.assertValueCount(4); + + LlmResponse finalResponse = testObserver.values().get(3); + assertThat(finalResponse.partial()).contains(false); + List finalParts = finalResponse.content().get().parts().get(); + // full text first, then the accumulated tool call + assertThat(finalParts.get(0).text()).contains("Checking the logs now."); + assertThat(finalParts.get(1).functionCall()).isPresent(); + assertThat(finalParts.get(1).functionCall().get().name()).contains("queryLog"); + } + + @Test + void testGenerateContentStreamingCjkTerminalPunctuationPersistsFinal() { + // Regression for CJK turns: a response ending with CJK terminal punctuation must still + // produce exactly one final (persisted) event, emitted on stream completion. + Flux responseFlux = + Flux.just( + createStreamingChatResponse("今天天气"), + createStreamingChatResponse("不错，适合出门。"), + createStreamingChatResponse("记得带伞。")); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + testObserver.assertValueCount(4); + + List responses = testObserver.values(); + for (int i = 0; i < 3; i++) { + assertThat(responses.get(i).partial()).contains(true); + } + LlmResponse finalResponse = responses.get(3); + assertThat(finalResponse.partial()).contains(false); + assertThat(finalResponse.content().get().parts().get().get(0).text()) + .contains("今天天气不错，适合出门。记得带伞。"); } @Test @@ -273,7 +491,8 @@ void testGenerateContentStreamingBackpressure() { testObserver.awaitDone(10, TimeUnit.SECONDS); testObserver.assertComplete(); testObserver.assertNoErrors(); - testObserver.assertValueCount(1000); + // 1000 partial chunks + 1 aggregated final response on stream completion + testObserver.assertValueCount(1001); } private ChatResponse createStreamingChatResponse(String text) {