MessageChatMemoryAdvisor persists malformed history on streaming + tool calls (next turn Bad Request 400)

Bug description

With streaming + MessageChatMemoryAdvisor + ToolCallingAdvisor.streamToolCallResponses(true) (needed if you want intermediate toolCalls chunks to reach the downstream Flux<ChatResponse>, e.g. to emit tool_call SSE events to a chat UI), a tool-calling turn writes an AssistantMessage to memory that contains both the final text and the intermediate tool_calls. The next turn replays it and DeepSeek / OpenAI rejects with 400:

An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'.

Root cause looks like MessageAggregator#aggregate (~L174-189): it concatenates all text and accumulates all toolCalls from every chunk on the flux into a single AssistantMessage, ignoring the boundary between the tool-call round and the final-text round.

Environment

• Spring AI 2.0.0-RC1
• Spring Boot 4.0.3, Java 25
• DeepSeek deepseek-v4-flash via spring-ai-starter-model-deepseek (any OpenAI-compatible backend should reproduce)

Steps to reproduce

1. ChatClient with default advisors ToolCallingAdvisor.streamToolCallResponses(true) + MessageChatMemoryAdvisor, one no-arg @Tool the model will call.
2. Round 1: stream "What time is it?", drain the flux.
3. Round 2: stream "Thanks" with the same conversation id → HTTP 400.

Expected behavior

After round 1, the persisted final AssistantMessage should not carry tool_calls unless it is followed by the matching ToolResponseMessage. Two reasonable shapes:

UserMessage("What time is it?")
AssistantMessage(content="It's ...", toolCalls=[])           // final reply only

// or, preserving the tool round-trip:
UserMessage("What time is it?")
AssistantMessage(content="", toolCalls=[getDateTime])
ToolResponseMessage(getDateTime -> "...")
AssistantMessage(content="It's ...", toolCalls=[])

Minimal Complete Reproducible example

package com.example;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.client.advisor.MessageChatMemoryAdvisor;
import org.springframework.ai.chat.client.advisor.ToolCallingAdvisor;
import org.springframework.ai.chat.memory.ChatMemory;
import org.springframework.ai.chat.memory.InMemoryChatMemoryRepository;
import org.springframework.ai.chat.memory.MessageWindowChatMemory;
import org.springframework.ai.deepseek.DeepSeekChatModel;
import org.springframework.ai.deepseek.DeepSeekChatOptions;
import org.springframework.ai.deepseek.api.DeepSeekApi;
import org.springframework.ai.tool.annotation.Tool;
import org.springframework.web.reactive.function.client.WebClientResponseException;

import java.time.LocalDateTime;

import static org.assertj.core.api.Assertions.assertThatThrownBy;

@EnabledIfEnvironmentVariable(named = "DEEPSEEK_API_KEY", matches = ".+")
class ReproTest {

    static class Clock {
        @Tool(name = "getDateTime", description = "ISO-8601 local date-time")
        String getDateTime() { return LocalDateTime.now().toString(); }
    }

    @Test
    void repro() {
        ChatMemory mem = MessageWindowChatMemory.builder()
                .chatMemoryRepository(new InMemoryChatMemoryRepository()).build();

        ChatClient client = ChatClient.builder(DeepSeekChatModel.builder()
                        .deepSeekApi(DeepSeekApi.builder().apiKey(System.getenv("DEEPSEEK_API_KEY")).build())
                        .options(DeepSeekChatOptions.builder().model("deepseek-v4-flash").build())
                        .build())
                .defaultAdvisors(
                        ToolCallingAdvisor.builder().streamToolCallResponses(true).build(),
                        MessageChatMemoryAdvisor.builder(mem).build())
                .defaultTools(new Clock())
                .build();

        client.prompt()
                .advisors(a -> a.param(ChatMemory.CONVERSATION_ID, "c1"))
                .user("What time is it?")
                .stream().chatResponse()
                .doOnNext(r -> {
                    var g = r.getResult();
                    System.out.println(" chunk: finishReason=" + (g.getMetadata() == null ? null : g.getMetadata().getFinishReason())
                            + " toolCalls=" + g.getOutput().getToolCalls()
                            + " text=" + g.getOutput().getText());
                })
                .blockLast();

        System.out.println("memory: " + mem.get("c1"));

        assertThatThrownBy(() -> client.prompt()
                .advisors(a -> a.param(ChatMemory.CONVERSATION_ID, "c1"))
                .user("Thanks")
                .stream().chatResponse().blockLast())
                .isInstanceOf(WebClientResponseException.BadRequest.class);
    }
}

Stdout (trimmed):

 chunk: finishReason=null toolCalls=[ToolCall[id=call_..., name=getDateTime, arguments={}]] text=
 chunk: finishReason=null toolCalls=[] text=It's
 ... (N text chunks)
 chunk: finishReason=STOP toolCalls=[] text=.
memory: [UserMessage(...),
         AssistantMessage(toolCalls=[ToolCall[...getDateTime...]], textContent=It's ..., ...)]
                          ^^^ orphan tool_calls, merged with final text into one message

Body of the round-2 400:

{"error":{"message":"An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. (insufficient tool messages following tool_calls message)","type":"invalid_request_error","code":"invalid_request_error"}}

With streamToolCallResponses(false) the bug is hidden because ToolCallingAdvisor filters those chunks before MessageChatMemoryAdvisor sees them, but that filtering kills the use case (UI never sees tool calls).

Start at MessageAggregator#aggregate around lines 174-189, then trace how MessageChatMemoryAdvisor receives streamed responses from ToolCallingAdvisor. Run the supplied ReproTest with DEEPSEEK_API_KEY set; done means the persisted history has valid tool-call sequencing and the second stream no longer returns HTTP 400.