From e3cea71245fd28ff638ece0403fa33c4c8a29b90 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Sat, 1 Aug 2026 13:59:55 +0700 Subject: [PATCH 1/4] fix(streaming): rationalize thinking retrieval vs model budget createObjectStreamWithThinking always enables thinking format instructions and extraction, independent of Thinking.withTokenBudget. Object-only streams omit thinking format. Document the split. Fixes #1799. Signed-off-by: arimu1 <19286898+arimu1@users.noreply.github.com> --- .../common/streaming/StreamingPromptRunner.kt | 6 ++ .../support/OperationContextDelegate.kt | 28 +++++++- .../StreamingChatClientOperations.kt | 13 +++- .../streaming/StreamingLlmOperationsImpl.kt | 12 +++- .../StreamingChatClientOperationsTest.kt | 67 ++++++++++++++++++- .../asciidoc/reference/streaming/page.adoc | 32 +++++++++ 6 files changed, 153 insertions(+), 5 deletions(-) diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt index 498e275ee..6c96e9354 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt @@ -87,6 +87,12 @@ interface StreamingPromptRunner : PromptRunner { * Create a reactive stream with both objects and thinking content. * Provides access to the LLM's reasoning process alongside the results. * + * Always enables thinking format instructions for this stream so the model is + * asked to emit reasoning blocks. This is independent of any model thinking + * budget on [com.embabel.common.ai.model.LlmOptions]: configure + * `Thinking.withTokenBudget(...)` only when the provider requires a budget + * (e.g. Anthropic extended thinking), not as a prerequisite for this API. + * * @param itemClass The class of objects to create * @return Flux emitting StreamingEvent instances for objects and thinking */ diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt index d353bb541..f9e023654 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt @@ -357,7 +357,9 @@ internal data class OperationContextDelegate( val streamingLlmOperations = streamingFactory().createStreamingOperations(llm) return streamingLlmOperations.createObjectStreamWithThinking( messages = messages, - interaction = streamingInteraction(), + // Enable application-level thinking extraction (format instructions + extractThinking) + // when needed, while preserving any caller-configured model thinking budget. + interaction = streamingInteractionForThinkingIfNecessary(), outputClass = itemClass, agentProcess = context.processContext.agentProcess, action = action, @@ -403,6 +405,30 @@ internal data class OperationContextDelegate( ) } + /** + * Streaming interaction for [createObjectStreamWithThinking]. + * + * Ensures application-level thinking extraction is enabled on [LlmOptions] (same idea as + * non-streaming [thinkingInteraction]) without requiring callers to pre-configure a thinking + * token budget. An existing budget or extraction config is preserved via + * [Thinking.applyExtraction]. + * + * This is *not* LLM-native reasoning (provider-specific thinking channels); those are out of + * scope here (see docs + issue #1716). Application-level thinking uses prompt format + * instructions so the model emits parseable reasoning blocks in the main content stream. + * + * Model thinking budget remains optional and independent: set it with + * `LlmOptions.withThinking(Thinking.withTokenBudget(...))` when the provider needs it. + */ + private fun streamingInteractionForThinkingIfNecessary(): LlmInteraction { + val base = streamingInteraction() + val thinking = when (val existing = llm.thinking) { + null, Thinking.NONE -> Thinking.withExtraction() + else -> existing.applyExtraction() + } + return base.copy(llm = llm.withThinking(thinking)) + } + private fun streamingFactory(): StreamingLlmOperationsFactory { val llmOperations = context.agentPlatform().platformServices.llmOperations return llmOperations as? StreamingLlmOperationsFactory diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt index 529ecd971..42d1a5772 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt @@ -239,6 +239,8 @@ internal class StreamingChatClientOperations( llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, + // Object-only stream: do not ask the model for thinking blocks (they would be discarded). + includeThinkingFormat = false, ) .filter { it.isObject() } .map { (it as StreamingEvent.Object).item } @@ -300,6 +302,11 @@ internal class StreamingChatClientOperations( llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, + // Explicit *WithThinking API: always include application-level thinking format + // instructions (prompt-instructed reasoning blocks). This is not LLM-native + // reasoning (provider thinking channels — see #1716). Model-side thinking budget + // (LlmOptions.thinking tokenBudget) remains independent. + includeThinkingFormat = true, ) } @@ -330,6 +337,9 @@ internal class StreamingChatClientOperations( * **Performance Characteristics:** * - Streaming-friendly: no blocking operations * + * @param includeThinkingFormat when true, prompt the model for `` blocks + * (used by [doTransformObjectStreamWithThinking]). Independent of any model thinking budget + * configured via [com.embabel.common.ai.model.LlmOptions.thinking]. * @return Unified Flux> that public methods can filter as needed */ private fun doTransformObjectStreamInternal( @@ -340,6 +350,7 @@ internal class StreamingChatClientOperations( llmRequestEvent: LlmRequestEvent?, agentProcess: AgentProcess?, action: Action?, + includeThinkingFormat: Boolean, ): Flux> { // Common setup - delegate to ChatClientLlmOperations for LLM setup val llm = chatClientLlmOperations.getLlm(interaction) @@ -357,7 +368,7 @@ internal class StreamingChatClientOperations( clazz = outputClassAny, objectMapper = chatClientLlmOperations.objectMapper, fieldFilter = interaction.fieldFilter, - thinkingEnabled = interaction.llm.thinking?.enabled ?: false, + thinkingEnabled = includeThinkingFormat, ) as StreamingJacksonOutputConverter // signature compatibility for downstream Flux/StreamingEvent uses // Build prompt using helper methods, including streaming format instructions diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt index aefdd863d..548d7c95f 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt @@ -157,6 +157,8 @@ internal class StreamingLlmOperationsImpl( llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, + // Object-only stream: do not ask the model for thinking blocks (they would be discarded). + includeThinkingFormat = false, ) .filter { it.isObject() } .map { (it as StreamingEvent.Object).item } @@ -177,6 +179,9 @@ internal class StreamingLlmOperationsImpl( llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, + // Explicit *WithThinking API: always include thinking format instructions. + // Model-side thinking budget (LlmOptions.thinking tokenBudget) remains independent. + includeThinkingFormat = true, ) } @@ -191,6 +196,10 @@ internal class StreamingLlmOperationsImpl( * 1. Raw LLM chunks from [LlmMessageStreamer] * 2. Line buffering via [rawChunksToLines] * 3. Event generation via [StreamingJacksonOutputConverter] + * + * @param includeThinkingFormat when true, prompt the model for `` blocks + * (used by [doTransformObjectStreamWithThinking]). Independent of any model thinking budget + * configured via [com.embabel.common.ai.model.LlmOptions.thinking]. */ private fun doTransformObjectStreamInternal( messages: List, @@ -200,6 +209,7 @@ internal class StreamingLlmOperationsImpl( llmRequestEvent: LlmRequestEvent?, agentProcess: AgentProcess?, action: Action?, + includeThinkingFormat: Boolean, ): Flux> { // Create converter for JSONL parsing. // Spring AI 2.0's StreamingJacksonOutputConverter requires T : Any; @@ -211,7 +221,7 @@ internal class StreamingLlmOperationsImpl( clazz = outputClassAny, objectMapper = objectMapper, fieldFilter = interaction.fieldFilter, - thinkingEnabled = interaction.llm.thinking?.enabled ?: false, + thinkingEnabled = includeThinkingFormat, ) as StreamingJacksonOutputConverter // Build prompt contributions with streaming format instructions diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt index 584f4a811..269261b8e 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt @@ -22,9 +22,12 @@ import com.embabel.agent.core.internal.streaming.StreamingLlmOperations import com.embabel.agent.spi.support.springai.ChatClientLlmOperations import com.embabel.agent.spi.support.springai.SpringAiLlmService import com.embabel.chat.UserMessage +import com.embabel.common.ai.model.LlmOptions import tools.jackson.module.kotlin.jacksonObjectMapper +import io.mockk.CapturingSlot import io.mockk.every import io.mockk.mockk +import io.mockk.slot import io.mockk.verify import org.junit.jupiter.api.Assertions.* import org.junit.jupiter.api.BeforeEach @@ -468,16 +471,76 @@ class StreamingChatClientOperationsTest { } - private fun mockChatClientForStreaming(chunkFlux: Flux) { + private fun mockChatClientForStreaming(chunkFlux: Flux): CapturingSlot { val mockRequestSpec = mockk(relaxed = true) val mockContentStreamSpec = mockk(relaxed = true) + val promptSlot = slot() - every { mockChatClient.prompt(any()) } returns mockRequestSpec + every { mockChatClient.prompt(capture(promptSlot)) } returns mockRequestSpec every { mockRequestSpec.tools(any>()) } returns mockRequestSpec every { mockRequestSpec.options(any()) } returns mockRequestSpec every { mockRequestSpec.stream() } returns mockContentStreamSpec every { mockContentStreamSpec.content() } returns chunkFlux + return promptSlot + } + + @Nested + inner class ThinkingFormatInstructionTests { + + @Test + fun `createObjectStreamWithThinking includes thinking format without LlmOptions thinking config`() { + // Given: no thinking budget / extraction on LlmOptions (the awkward pre-req from #1799) + every { mockInteraction.llm } returns LlmOptions() + val promptSlot = mockChatClientForStreaming( + Flux.just("{\"name\":\"Item1\",\"value\":1}\n") + ) + + // When + streamingOperations.createObjectStreamWithThinking( + messages = listOf(UserMessage("test")), + interaction = mockInteraction, + outputClass = TestItem::class.java, + agentProcess = mockAgentProcess, + action = mockAction + ).collectList().block(Duration.ofSeconds(2)) + + // Then: format instructions still ask for blocks + assertTrue(promptSlot.isCaptured, "expected ChatClient.prompt to be called") + val promptText = promptSlot.captured.contents + assertTrue( + promptText.contains(""), + "createObjectStreamWithThinking should inject thinking format without Thinking.withTokenBudget" + ) + } + + @Test + fun `createObjectStream omits thinking format instructions`() { + // Given: even if model thinking budget is configured, object-only stream should not + // push the model to emit thinking blocks (they would be discarded) + every { mockInteraction.llm } returns LlmOptions() + .withThinking(com.embabel.common.ai.model.Thinking.withTokenBudget(8000)) + val promptSlot = mockChatClientForStreaming( + Flux.just("{\"name\":\"Item1\",\"value\":1}\n") + ) + + // When + streamingOperations.createObjectStream( + messages = listOf(UserMessage("test")), + interaction = mockInteraction, + outputClass = TestItem::class.java, + agentProcess = mockAgentProcess, + action = mockAction + ).collectList().block(Duration.ofSeconds(2)) + + // Then + assertTrue(promptSlot.isCaptured, "expected ChatClient.prompt to be called") + val promptText = promptSlot.captured.contents + assertFalse( + promptText.contains(""), + "createObjectStream should not inject thinking format instructions" + ) + } } /** diff --git a/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc b/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc index f69b5c278..9a613511d 100644 --- a/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc +++ b/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc @@ -37,10 +37,42 @@ Provider integrations override `LlmMessageStreamer.streamInference` by emitting the assembled assistant message. The completed message must include provider-assigned tool-call IDs, names, and complete arguments. Provider adapters must not execute those calls themselves. +==== Thinking: application-level retrieval vs model budget + +Embabel streaming currently deals with *application-level thinking*: the model is prompted +(via thinking *format instructions*) to emit reasoning blocks that Embabel parses into +``StreamingEvent.Thinking`` entries. This is separate from *LLM-native reasoning* (for example +Anthropic extended thinking / other providers' separate reasoning channels), which is usually +returned as a dedicated response property and is **not** extracted by Embabel today +(see https://github.com/embabel/embabel-agent/issues/1716[#1716]). + +For application-level thinking, streaming exposes two independent concerns (same split as +non-streaming thinking APIs): + +* *Retrieving thinking events* — Call ``createObjectStreamWithThinking(...)``. + Embabel always enables extraction and injects thinking format instructions for that stream. + You do *not* need to set a thinking token budget just to use this API. +* *Model thinking budget / thoroughness* — Optionally configure + ``LlmOptions.withThinking(Thinking.withTokenBudget(n))`` when the provider requires (or + benefits from) an explicit budget. Use this when you want the model to reason more deeply, + even if you do not consume thinking events, or together with ``createObjectStreamWithThinking``. + +For models where thinking is always on and not budget-configurable, call +``createObjectStreamWithThinking`` without a token budget; any application-level thinking the +model emits is still delivered as ``StreamingEvent`` thinking entries. + +Object-only streams (``createObjectStream`` / ``generateStream``) do not inject thinking +format instructions, so they will not push the model to emit application-level reasoning blocks. + ==== Example - Simple Thinking and Object Streaming with Callbacks [source,java] ---- + // Optional: only when the provider needs a thinking budget (e.g. Anthropic). + // Not required just to receive thinking StreamingEvents. + // LlmOptions options = new LlmOptions().withThinking(Thinking.withTokenBudget(8000)); + // PromptRunner runner = ai.withLlm(options).withToolObject(Tooling.class); + PromptRunner runner = ai.withDefaultLlm() // Example uses qwen3:latest .withToolObject(Tooling.class); From 2dc25c05795ac32df3630ecbbce96aabe2ca7291 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Sun, 2 Aug 2026 17:10:31 +0700 Subject: [PATCH 2/4] test(streaming): drop unnecessary FQNs in StreamingChatClientOperationsTest Address review feedback to use imports instead of fully qualified names. Signed-off-by: arimu1 <19286898+arimu1@users.noreply.github.com> --- .../streaming/StreamingChatClientOperationsTest.kt | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt index 269261b8e..6aca93b0e 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt @@ -40,6 +40,9 @@ import org.springframework.ai.tool.ToolCallback import reactor.core.publisher.Flux import reactor.test.StepVerifier import java.time.Duration +import com.embabel.common.ai.model.OptionsConverter +import com.embabel.common.ai.model.Thinking +import com.embabel.common.ai.prompt.PromptContributor /** * Unit tests for StreamingChatClientOperations. @@ -82,7 +85,7 @@ class StreamingChatClientOperationsTest { every { mockChatClientLlmOperations.createChatClient(mockLlm) } returns mockChatClient every { mockInteraction.promptContributors } returns emptyList() every { mockLlm.promptContributors } returns emptyList() - val mockOptionsConverter = mockk(relaxed = true) + val mockOptionsConverter = mockk(relaxed = true) every { mockLlm.optionsConverter } returns mockOptionsConverter every { mockOptionsConverter.convertOptions(any(), any()) } returns mockk(relaxed = true) every { mockInteraction.llm } returns mockk(relaxed = true) @@ -519,7 +522,7 @@ class StreamingChatClientOperationsTest { // Given: even if model thinking budget is configured, object-only stream should not // push the model to emit thinking blocks (they would be discarded) every { mockInteraction.llm } returns LlmOptions() - .withThinking(com.embabel.common.ai.model.Thinking.withTokenBudget(8000)) + .withThinking(Thinking.withTokenBudget(8000)) val promptSlot = mockChatClientForStreaming( Flux.just("{\"name\":\"Item1\",\"value\":1}\n") ) @@ -583,7 +586,7 @@ class StreamingChatClientOperationsTest { @Test fun `should prepend prompt contributions as system message`() { // Given - val mockContributor = mockk() + val mockContributor = mockk() every { mockContributor.contribution() } returns "System contribution" every { mockInteraction.promptContributors } returns listOf(mockContributor) From 87ab47eafc96cf71a1a49f42ff656a8bd0c000d1 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Mon, 3 Aug 2026 08:05:11 +0700 Subject: [PATCH 3/4] fix(streaming): drive thinking format from Thinking.extractThinking Remove the separate includeThinkingFormat SPI parameter. Application-level prompt format now follows Thinking.extractThinking on the Interaction (enabled via withExtraction/applyExtraction); provider tokenBudget remains independent (Thinking.enabled). WithThinking paths ensure extractThinking is set so format propagates through Interaction only. Addresses igordayen design feedback on #1853 / #1799. Signed-off-by: arimu1 <19286898+arimu1@users.noreply.github.com> --- .../common/streaming/StreamingPromptRunner.kt | 11 +++-- .../support/OperationContextDelegate.kt | 19 ++++---- .../StreamingChatClientOperations.kt | 46 +++++++++++++------ .../streaming/StreamingLlmOperationsImpl.kt | 36 ++++++++++----- .../StreamingChatClientOperationsTest.kt | 33 +++++++------ .../asciidoc/reference/streaming/page.adoc | 38 +++++++-------- 6 files changed, 112 insertions(+), 71 deletions(-) diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt index 6c96e9354..e98631407 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt @@ -87,11 +87,12 @@ interface StreamingPromptRunner : PromptRunner { * Create a reactive stream with both objects and thinking content. * Provides access to the LLM's reasoning process alongside the results. * - * Always enables thinking format instructions for this stream so the model is - * asked to emit reasoning blocks. This is independent of any model thinking - * budget on [com.embabel.common.ai.model.LlmOptions]: configure - * `Thinking.withTokenBudget(...)` only when the provider requires a budget - * (e.g. Anthropic extended thinking), not as a prerequisite for this API. + * Enables application-level thinking on the Interaction + * ([com.embabel.common.ai.model.Thinking.withExtraction] / `extractThinking`) so the + * stream injects prompt format instructions and returns reasoning blocks. This is + * independent of any provider model budget (`Thinking.withTokenBudget(...)`), which is + * only needed when the provider requires a budget (e.g. Anthropic extended thinking) — + * not as a prerequisite for this API. * * @param itemClass The class of objects to create * @return Flux emitting StreamingEvent instances for objects and thinking diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt index f9e023654..6a7cf3984 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt @@ -408,23 +408,22 @@ internal data class OperationContextDelegate( /** * Streaming interaction for [createObjectStreamWithThinking]. * - * Ensures application-level thinking extraction is enabled on [LlmOptions] (same idea as - * non-streaming [thinkingInteraction]) without requiring callers to pre-configure a thinking - * token budget. An existing budget or extraction config is preserved via - * [Thinking.applyExtraction]. + * Turns on application-level thinking on the Interaction via [Thinking.extractThinking] + * (same idea as non-streaming [thinkingInteraction] / [Thinking.withExtraction]), without + * requiring a provider token budget. Existing budget is preserved with [Thinking.applyExtraction]. * - * This is *not* LLM-native reasoning (provider-specific thinking channels); those are out of - * scope here (see docs + issue #1716). Application-level thinking uses prompt format - * instructions so the model emits parseable reasoning blocks in the main content stream. + * SPI streaming then reads [Thinking.extractThinking] to decide whether to inject prompt + * format instructions — no separate "thinking format" flag. Propagation is entirely through + * [LlmInteraction.llm.thinking]. * - * Model thinking budget remains optional and independent: set it with - * `LlmOptions.withThinking(Thinking.withTokenBudget(...))` when the provider needs it. + * This is *not* LLM-native reasoning (provider thinking channels; see #1716). + * Provider budget remains optional: `LlmOptions.withThinking(Thinking.withTokenBudget(...))`. */ private fun streamingInteractionForThinkingIfNecessary(): LlmInteraction { val base = streamingInteraction() val thinking = when (val existing = llm.thinking) { null, Thinking.NONE -> Thinking.withExtraction() - else -> existing.applyExtraction() + else -> if (existing.extractThinking) existing else existing.applyExtraction() } return base.copy(llm = llm.withThinking(thinking)) } diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt index 42d1a5772..97172165e 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt @@ -33,6 +33,7 @@ import com.embabel.agent.spi.support.springai.toSpringAiMessage import com.embabel.agent.spi.support.springai.toSpringToolCallbacks import com.embabel.chat.Message import com.embabel.common.ai.converters.streaming.StreamingJacksonOutputConverter +import com.embabel.common.ai.model.Thinking import com.embabel.common.core.streaming.StreamingEvent import org.slf4j.LoggerFactory import org.springframework.ai.chat.messages.SystemMessage @@ -234,13 +235,13 @@ internal class StreamingChatClientOperations( ): Flux { return doTransformObjectStreamInternal( messages = messages, + // Object-only stream: leave Interaction thinking as-is. Format instructions follow + // Thinking.extractThinking (application-level), not provider tokenBudget (Thinking.enabled). interaction = interaction, outputClass = outputClass, llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, - // Object-only stream: do not ask the model for thinking blocks (they would be discarded). - includeThinkingFormat = false, ) .filter { it.isObject() } .map { (it as StreamingEvent.Object).item } @@ -297,19 +298,36 @@ internal class StreamingChatClientOperations( ): Flux> { return doTransformObjectStreamInternal( messages = messages, - interaction = interaction, + // *WithThinking*: ensure Interaction carries application-level Thinking + // (extractThinking). Format instructions follow that flag — no separate SPI param. + interaction = withApplicationLevelThinkingIfNecessary(interaction), outputClass = outputClass, llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, - // Explicit *WithThinking API: always include application-level thinking format - // instructions (prompt-instructed reasoning blocks). This is not LLM-native - // reasoning (provider thinking channels — see #1716). Model-side thinking budget - // (LlmOptions.thinking tokenBudget) remains independent. - includeThinkingFormat = true, ) } + /** + * Ensure [Thinking.extractThinking] is set on the interaction for application-level + * (prompt-instructed) thinking streams. Preserves any existing provider budget + * ([Thinking.enabled] / [Thinking.tokenBudget]) via [Thinking.applyExtraction]. + * + * This is *not* LLM-native reasoning (provider thinking channels — see #1716). + */ + private fun withApplicationLevelThinkingIfNecessary(interaction: LlmInteraction): LlmInteraction { + val existing = interaction.llm.thinking + val thinking = when (existing) { + null, Thinking.NONE -> Thinking.withExtraction() + else -> if (existing.extractThinking) existing else existing.applyExtraction() + } + return if (thinking === existing) { + interaction + } else { + interaction.copy(llm = interaction.llm.withThinking(thinking)) + } + } + /** * Internal unified streaming implementation - workhorse -that handles the complete transformation pipeline. * @@ -337,9 +355,9 @@ internal class StreamingChatClientOperations( * **Performance Characteristics:** * - Streaming-friendly: no blocking operations * - * @param includeThinkingFormat when true, prompt the model for `` blocks - * (used by [doTransformObjectStreamWithThinking]). Independent of any model thinking budget - * configured via [com.embabel.common.ai.model.LlmOptions.thinking]. + * Prompt thinking format follows [Thinking.extractThinking] on [interaction] (application-level). + * Provider model budget remains [Thinking.enabled] / [Thinking.tokenBudget] and is independent. + * * @return Unified Flux> that public methods can filter as needed */ private fun doTransformObjectStreamInternal( @@ -350,7 +368,6 @@ internal class StreamingChatClientOperations( llmRequestEvent: LlmRequestEvent?, agentProcess: AgentProcess?, action: Action?, - includeThinkingFormat: Boolean, ): Flux> { // Common setup - delegate to ChatClientLlmOperations for LLM setup val llm = chatClientLlmOperations.getLlm(interaction) @@ -359,6 +376,9 @@ internal class StreamingChatClientOperations( // Chat Options, additional potential option "streaming" val chatOptions = requireSpringAiLlm(llm).convertOptions(interaction.llm) + // Application-level thinking format: Thinking.extractThinking (not provider tokenBudget / enabled). + val includeApplicationLevelThinking = interaction.llm.thinking?.extractThinking == true + // Spring AI 2.0's StreamingJacksonOutputConverter requires T : Any; // erase O via Class for the construction, cast result back at use sites. @Suppress("UNCHECKED_CAST") @@ -368,7 +388,7 @@ internal class StreamingChatClientOperations( clazz = outputClassAny, objectMapper = chatClientLlmOperations.objectMapper, fieldFilter = interaction.fieldFilter, - thinkingEnabled = includeThinkingFormat, + thinkingEnabled = includeApplicationLevelThinking, ) as StreamingJacksonOutputConverter // signature compatibility for downstream Flux/StreamingEvent uses // Build prompt using helper methods, including streaming format instructions diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt index 548d7c95f..28cd06290 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt @@ -37,6 +37,7 @@ import com.embabel.agent.spi.support.guardrails.validateUserInput import com.embabel.chat.Message import com.embabel.chat.UserMessage import com.embabel.common.ai.converters.streaming.StreamingJacksonOutputConverter +import com.embabel.common.ai.model.Thinking import com.embabel.common.core.streaming.StreamingEvent import tools.jackson.databind.ObjectMapper import org.slf4j.LoggerFactory @@ -152,13 +153,12 @@ internal class StreamingLlmOperationsImpl( ): Flux { return doTransformObjectStreamInternal( messages = messages, + // Object-only: format instructions follow Thinking.extractThinking on Interaction. interaction = interaction, outputClass = outputClass, llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, - // Object-only stream: do not ask the model for thinking blocks (they would be discarded). - includeThinkingFormat = false, ) .filter { it.isObject() } .map { (it as StreamingEvent.Object).item } @@ -174,17 +174,32 @@ internal class StreamingLlmOperationsImpl( ): Flux> { return doTransformObjectStreamInternal( messages = messages, - interaction = interaction, + // *WithThinking*: ensure Interaction has application-level Thinking.extractThinking. + interaction = withApplicationLevelThinkingIfNecessary(interaction), outputClass = outputClass, llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, action = action, - // Explicit *WithThinking API: always include thinking format instructions. - // Model-side thinking budget (LlmOptions.thinking tokenBudget) remains independent. - includeThinkingFormat = true, ) } + /** + * Ensure [Thinking.extractThinking] is set for application-level (prompt-instructed) thinking. + * Preserves provider budget via [Thinking.applyExtraction]. Not LLM-native reasoning (#1716). + */ + private fun withApplicationLevelThinkingIfNecessary(interaction: LlmInteraction): LlmInteraction { + val existing = interaction.llm.thinking + val thinking = when (existing) { + null, Thinking.NONE -> Thinking.withExtraction() + else -> if (existing.extractThinking) existing else existing.applyExtraction() + } + return if (thinking === existing) { + interaction + } else { + interaction.copy(llm = interaction.llm.withThinking(thinking)) + } + } + // ======================================== // Internal implementation // ======================================== @@ -197,9 +212,8 @@ internal class StreamingLlmOperationsImpl( * 2. Line buffering via [rawChunksToLines] * 3. Event generation via [StreamingJacksonOutputConverter] * - * @param includeThinkingFormat when true, prompt the model for `` blocks - * (used by [doTransformObjectStreamWithThinking]). Independent of any model thinking budget - * configured via [com.embabel.common.ai.model.LlmOptions.thinking]. + * Prompt thinking format follows [Thinking.extractThinking] on [interaction]. + * Provider model budget remains [Thinking.enabled] / [Thinking.tokenBudget]. */ private fun doTransformObjectStreamInternal( messages: List, @@ -209,11 +223,11 @@ internal class StreamingLlmOperationsImpl( llmRequestEvent: LlmRequestEvent?, agentProcess: AgentProcess?, action: Action?, - includeThinkingFormat: Boolean, ): Flux> { // Create converter for JSONL parsing. // Spring AI 2.0's StreamingJacksonOutputConverter requires T : Any; // erase O via Class for the construction, cast back for downstream Flux/StreamingEvent. + val includeApplicationLevelThinking = interaction.llm.thinking?.extractThinking == true @Suppress("UNCHECKED_CAST") val outputClassAny = outputClass as Class @Suppress("UNCHECKED_CAST") @@ -221,7 +235,7 @@ internal class StreamingLlmOperationsImpl( clazz = outputClassAny, objectMapper = objectMapper, fieldFilter = interaction.fieldFilter, - thinkingEnabled = includeThinkingFormat, + thinkingEnabled = includeApplicationLevelThinking, ) as StreamingJacksonOutputConverter // Build prompt contributions with streaming format instructions diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt index 6aca93b0e..f6139ff46 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt @@ -15,6 +15,7 @@ */ package com.embabel.agent.spi.support.springai.streaming +import com.embabel.agent.api.common.InteractionId import com.embabel.agent.core.Action import com.embabel.agent.core.AgentProcess import com.embabel.agent.core.support.LlmInteraction @@ -167,8 +168,8 @@ class StreamingChatClientOperationsTest { mockAction ) - // Then - verify { mockChatClientLlmOperations.getLlm(mockInteraction) } + // Then: Interaction may be a copy with Thinking.extractThinking enabled + verify { mockChatClientLlmOperations.getLlm(any()) } } @Test @@ -493,8 +494,12 @@ class StreamingChatClientOperationsTest { @Test fun `createObjectStreamWithThinking includes thinking format without LlmOptions thinking config`() { - // Given: no thinking budget / extraction on LlmOptions (the awkward pre-req from #1799) - every { mockInteraction.llm } returns LlmOptions() + // Given: real Interaction with no thinking budget / extraction (the #1799 pre-req). + // SPI enables Thinking.extractThinking on the Interaction for *WithThinking. + val interaction = LlmInteraction( + id = InteractionId("test-thinking-format"), + llm = LlmOptions(), + ) val promptSlot = mockChatClientForStreaming( Flux.just("{\"name\":\"Item1\",\"value\":1}\n") ) @@ -502,13 +507,13 @@ class StreamingChatClientOperationsTest { // When streamingOperations.createObjectStreamWithThinking( messages = listOf(UserMessage("test")), - interaction = mockInteraction, + interaction = interaction, outputClass = TestItem::class.java, agentProcess = mockAgentProcess, action = mockAction ).collectList().block(Duration.ofSeconds(2)) - // Then: format instructions still ask for blocks + // Then: format instructions still ask for blocks (driven by extractThinking) assertTrue(promptSlot.isCaptured, "expected ChatClient.prompt to be called") val promptText = promptSlot.captured.contents assertTrue( @@ -518,11 +523,13 @@ class StreamingChatClientOperationsTest { } @Test - fun `createObjectStream omits thinking format instructions`() { - // Given: even if model thinking budget is configured, object-only stream should not - // push the model to emit thinking blocks (they would be discarded) - every { mockInteraction.llm } returns LlmOptions() - .withThinking(Thinking.withTokenBudget(8000)) + fun `createObjectStream omits thinking format for provider budget only`() { + // Provider budget (Thinking.enabled + tokenBudget) alone must not inject + // application-level format instructions — that follows extractThinking only. + val interaction = LlmInteraction( + id = InteractionId("test-budget-only"), + llm = LlmOptions().withThinking(Thinking.withTokenBudget(8000)), + ) val promptSlot = mockChatClientForStreaming( Flux.just("{\"name\":\"Item1\",\"value\":1}\n") ) @@ -530,7 +537,7 @@ class StreamingChatClientOperationsTest { // When streamingOperations.createObjectStream( messages = listOf(UserMessage("test")), - interaction = mockInteraction, + interaction = interaction, outputClass = TestItem::class.java, agentProcess = mockAgentProcess, action = mockAction @@ -541,7 +548,7 @@ class StreamingChatClientOperationsTest { val promptText = promptSlot.captured.contents assertFalse( promptText.contains(""), - "createObjectStream should not inject thinking format instructions" + "tokenBudget alone should not inject application-level thinking format" ) } } diff --git a/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc b/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc index 9a613511d..9f439be35 100644 --- a/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc +++ b/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc @@ -41,28 +41,28 @@ names, and complete arguments. Provider adapters must not execute those calls th Embabel streaming currently deals with *application-level thinking*: the model is prompted (via thinking *format instructions*) to emit reasoning blocks that Embabel parses into -``StreamingEvent.Thinking`` entries. This is separate from *LLM-native reasoning* (for example -Anthropic extended thinking / other providers' separate reasoning channels), which is usually -returned as a dedicated response property and is **not** extracted by Embabel today +``StreamingEvent.Thinking`` entries. Control is through ``Thinking`` on the Interaction / +``LlmOptions``: + +* *Application-level* — ``Thinking.withExtraction()`` sets ``extractThinking``. That flag + turns on prompt format instructions and block extraction. ``createObjectStreamWithThinking`` + enables this automatically (no separate SPI "format" parameter). +* *Provider model budget* — ``Thinking.withTokenBudget(n)`` sets ``enabled`` + budget for + providers that need it (e.g. Anthropic). Independent of application-level extraction; + combine via ``applyExtraction()`` when you want both. + +This is separate from *LLM-native reasoning* (provider-specific reasoning channels returned as +a dedicated response property), which Embabel does **not** extract today (see https://github.com/embabel/embabel-agent/issues/1716[#1716]). -For application-level thinking, streaming exposes two independent concerns (same split as -non-streaming thinking APIs): - * *Retrieving thinking events* — Call ``createObjectStreamWithThinking(...)``. - Embabel always enables extraction and injects thinking format instructions for that stream. - You do *not* need to set a thinking token budget just to use this API. -* *Model thinking budget / thoroughness* — Optionally configure - ``LlmOptions.withThinking(Thinking.withTokenBudget(n))`` when the provider requires (or - benefits from) an explicit budget. Use this when you want the model to reason more deeply, - even if you do not consume thinking events, or together with ``createObjectStreamWithThinking``. - -For models where thinking is always on and not budget-configurable, call -``createObjectStreamWithThinking`` without a token budget; any application-level thinking the -model emits is still delivered as ``StreamingEvent`` thinking entries. - -Object-only streams (``createObjectStream`` / ``generateStream``) do not inject thinking -format instructions, so they will not push the model to emit application-level reasoning blocks. + You do *not* need a token budget just to use this API. +* *Model thinking budget* — Optionally ``LlmOptions.withThinking(Thinking.withTokenBudget(n))`` + when the provider requires it. + +Object-only streams (``createObjectStream`` / ``generateStream``) only inject thinking format +instructions when ``extractThinking`` is already set on the Interaction (a token budget alone +does not). ==== Example - Simple Thinking and Object Streaming with Callbacks [source,java] From b497c7de32e89a37a1ab19bc0463ac716909e228 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Tue, 4 Aug 2026 06:30:52 +0700 Subject: [PATCH 4/4] test(streaming): cover thinking retrieval vs budget in Ollama IT MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add LLMOllamaStreamingBuilderIT coverage for createObjectStreamWithThinking without tokenBudget, with tokenBudget (applyExtraction), and object-only stream with budget alone — validates application-level extractThinking is independent of provider Thinking.withTokenBudget (#1799 / #1853). Signed-off-by: arimu1 <19286898+arimu1@users.noreply.github.com> --- .../ollama/LLMOllamaStreamingBuilderIT.java | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) diff --git a/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java b/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java index 675eaa55f..c937a86d6 100644 --- a/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java +++ b/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java @@ -22,6 +22,8 @@ import com.embabel.agent.autoconfigure.models.ollama.AgentOllamaAutoConfiguration; import com.embabel.agent.spi.LlmService; import com.embabel.agent.spi.support.springai.SpringAiLlmService; +import com.embabel.common.ai.model.LlmOptions; +import com.embabel.common.ai.model.Thinking; import com.embabel.common.core.streaming.StreamingEvent; import org.junit.jupiter.api.Test; import org.slf4j.Logger; @@ -218,6 +220,155 @@ void realStreamingOllamaIntegrationWithReactiveCallbacks() { logger.info("Integration streaming test completed successfully with {} total events", receivedEvents.size()); } + /** + * Validates application-level thinking retrieval vs provider token budget (#1799 / #1853). + *

+ * {@code createObjectStreamWithThinking} must enable thinking extraction/format via + * {@code Thinking.extractThinking} without requiring {@code Thinking.withTokenBudget}. + * Object-only streaming with a provider budget alone must still complete (budget is orthogonal). + */ + @Test + void createObjectStreamWithThinkingRetrievalIndependentOfTokenBudget() { + reactor.util.Loggers.useVerboseConsoleLoggers(); + + // --- Path A: *WithThinking* without any LlmOptions thinking / token budget --- + // Format instructions and extractThinking are applied by createObjectStreamWithThinking. + PromptRunner extractionOnlyRunner = ai.withLlm("qwen3:latest") + .withToolObject(new Tooling()); + assertTrue(extractionOnlyRunner.supportsStreaming(), "Test LLM should support streaming"); + + List extractionEvents = new CopyOnWriteArrayList<>(); + AtomicReference extractionError = new AtomicReference<>(); + AtomicBoolean extractionCompleted = new AtomicBoolean(false); + + String thinkingPrompt = + "What are exactly two of the hottest months in Florida and their highest temperatures. " + + "Think step by step before returning the JSONL objects."; + + Flux> extractionStream = new StreamingPromptRunnerBuilder(extractionOnlyRunner) + .streaming() + .withPrompt(thinkingPrompt) + .createObjectStreamWithThinking(MonthItem.class); + + extractionStream + .timeout(Duration.ofSeconds(150)) + .doOnNext(event -> { + if (event.isThinking()) { + extractionEvents.add("THINKING: " + event.getThinking()); + logger.info("Extraction-only path thinking: {}", event.getThinking()); + } else if (event.isObject()) { + MonthItem obj = event.getObject(); + extractionEvents.add("OBJECT: " + obj.getName()); + logger.info("Extraction-only path object: {}", obj.getName()); + } + }) + .doOnError(error -> { + extractionError.set(error); + logger.error("Extraction-only stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> extractionCompleted.set(true)) + .blockLast(Duration.ofSeconds(6000)); + + assertNull(extractionError.get(), "createObjectStreamWithThinking must work without tokenBudget"); + assertTrue(extractionCompleted.get(), "Extraction-only stream should complete"); + assertFalse(extractionEvents.isEmpty(), "Should receive at least one streaming event without tokenBudget"); + assertTrue( + extractionEvents.stream().anyMatch(e -> e.startsWith("OBJECT:")), + "Extraction-only path should still produce object events" + ); + logger.info( + "Extraction-only createObjectStreamWithThinking completed with {} events (thinking events may vary by model)", + extractionEvents.size() + ); + + // --- Path B: *WithThinking* with provider token budget (budget preserved + extractThinking applied) --- + PromptRunner budgetAndExtractionRunner = ai.withLlm( + LlmOptions.withModel("qwen3:latest") + .withThinking(Thinking.withTokenBudget(100))) + .withToolObject(new Tooling()); + assertTrue(budgetAndExtractionRunner.supportsStreaming(), "Budget-configured LLM should support streaming"); + + List budgetEvents = new CopyOnWriteArrayList<>(); + AtomicReference budgetError = new AtomicReference<>(); + AtomicBoolean budgetCompleted = new AtomicBoolean(false); + + Flux> budgetStream = new StreamingPromptRunnerBuilder(budgetAndExtractionRunner) + .streaming() + .withPrompt(thinkingPrompt) + .createObjectStreamWithThinking(MonthItem.class); + + budgetStream + .timeout(Duration.ofSeconds(150)) + .doOnNext(event -> { + if (event.isThinking()) { + budgetEvents.add("THINKING: " + event.getThinking()); + logger.info("Budget+extraction path thinking: {}", event.getThinking()); + } else if (event.isObject()) { + MonthItem obj = event.getObject(); + budgetEvents.add("OBJECT: " + obj.getName()); + logger.info("Budget+extraction path object: {}", obj.getName()); + } + }) + .doOnError(error -> { + budgetError.set(error); + logger.error("Budget+extraction stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> budgetCompleted.set(true)) + .blockLast(Duration.ofSeconds(6000)); + + assertNull(budgetError.get(), "createObjectStreamWithThinking must work with tokenBudget (applyExtraction)"); + assertTrue(budgetCompleted.get(), "Budget+extraction stream should complete"); + assertFalse(budgetEvents.isEmpty(), "Should receive events when tokenBudget coexists with extractThinking"); + assertTrue( + budgetEvents.stream().anyMatch(e -> e.startsWith("OBJECT:")), + "Budget+extraction path should still produce object events" + ); + logger.info( + "Budget+extraction createObjectStreamWithThinking completed with {} events", + budgetEvents.size() + ); + + // --- Path C: object-only stream with provider budget only (no *WithThinking*) --- + // Token budget alone must not require thinking events; stream of typed objects completes. + PromptRunner budgetOnlyRunner = ai.withLlm( + LlmOptions.withModel("qwen3:latest") + .withThinking(Thinking.withTokenBudget(100))) + .withToolObject(new Tooling()); + + List objectOnlyEvents = new CopyOnWriteArrayList<>(); + AtomicReference objectOnlyError = new AtomicReference<>(); + AtomicBoolean objectOnlyCompleted = new AtomicBoolean(false); + + String objectOnlyPrompt = + "Return exactly two JSONL objects for the hottest months in Florida with name and temperature."; + + Flux objectOnlyStream = new StreamingPromptRunnerBuilder(budgetOnlyRunner) + .streaming() + .withPrompt(objectOnlyPrompt) + .createObjectStream(MonthItem.class); + + objectOnlyStream + .timeout(Duration.ofSeconds(150)) + .doOnNext(obj -> { + objectOnlyEvents.add("OBJECT: " + obj.getName()); + logger.info("Object-only + budget path object: {}", obj.getName()); + }) + .doOnError(error -> { + objectOnlyError.set(error); + logger.error("Object-only stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> objectOnlyCompleted.set(true)) + .blockLast(Duration.ofSeconds(6000)); + + assertNull(objectOnlyError.get(), "Object-only stream with tokenBudget should not error"); + assertTrue(objectOnlyCompleted.get(), "Object-only stream should complete"); + assertFalse(objectOnlyEvents.isEmpty(), "Object-only path should receive typed objects"); + logger.info( + "Object-only stream with tokenBudget completed with {} objects (thinking retrieval independent of budget)", + objectOnlyEvents.size() + ); + } + @Test void rawTextStreamingOllamaIntegrationWithReactiveCallbacks() { // Enable Reactor debugging