fix(streaming): rationalize thinking retrieval vs model budget

embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt

@@ -87,6 +87,13 @@ interface StreamingPromptRunner : PromptRunner {
          * Create a reactive stream with both objects and thinking content.
          * Provides access to the LLM's reasoning process alongside the results.
          *
+         * 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
          */

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,29 @@ internal data class OperationContextDelegate(
         )
     }
 
+    /**
+     * Streaming interaction for [createObjectStreamWithThinking].
+     *
+     * 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].
+     *
+     * 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].
+     *
+     * 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 -> if (existing.extractThinking) existing else existing.applyExtraction()
+        }
+        return base.copy(llm = llm.withThinking(thinking))
+    }
+
     private fun streamingFactory(): StreamingLlmOperationsFactory {
         val llmOperations = context.agentPlatform().platformServices.llmOperations
         return llmOperations as? StreamingLlmOperationsFactory

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,6 +235,8 @@ internal class StreamingChatClientOperations(
     ): Flux<O> {
         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,
@@ -295,14 +298,36 @@ internal class StreamingChatClientOperations(
     ): Flux<StreamingEvent<O>> {
         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,
         )
     }
 
+    /**
+     * 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.
      *
@@ -330,6 +355,9 @@ internal class StreamingChatClientOperations(
      * **Performance Characteristics:**
      * - Streaming-friendly: no blocking operations
      *
+     * Prompt thinking format follows [Thinking.extractThinking] on [interaction] (application-level).
+     * Provider model budget remains [Thinking.enabled] / [Thinking.tokenBudget] and is independent.
+     *
      * @return Unified Flux<StreamingEvent<O>> that public methods can filter as needed
      */
     private fun <O> doTransformObjectStreamInternal(
@@ -348,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<Any> for the construction, cast result back at use sites.
         @Suppress("UNCHECKED_CAST")
@@ -357,7 +388,7 @@ internal class StreamingChatClientOperations(
             clazz = outputClassAny,
             objectMapper = chatClientLlmOperations.objectMapper,
             fieldFilter = interaction.fieldFilter,
-            thinkingEnabled = interaction.llm.thinking?.enabled ?: false,
+            thinkingEnabled = includeApplicationLevelThinking,
         ) as StreamingJacksonOutputConverter<O>  // signature compatibility for downstream Flux<O>/StreamingEvent<O> uses
 
         // Build prompt using helper methods, including streaming format instructions

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,6 +153,7 @@ internal class StreamingLlmOperationsImpl(
     ): Flux<O> {
         return doTransformObjectStreamInternal(
             messages = messages,
+            // Object-only: format instructions follow Thinking.extractThinking on Interaction.
             interaction = interaction,
             outputClass = outputClass,
             llmRequestEvent = llmRequestEvent,
@@ -172,14 +174,32 @@ internal class StreamingLlmOperationsImpl(
     ): Flux<StreamingEvent<O>> {
         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,
         )
     }
 
+    /**
+     * 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
     // ========================================
@@ -191,6 +211,9 @@ internal class StreamingLlmOperationsImpl(
      * 1. Raw LLM chunks from [LlmMessageStreamer]
      * 2. Line buffering via [rawChunksToLines]
      * 3. Event generation via [StreamingJacksonOutputConverter]
+     *
+     * Prompt thinking format follows [Thinking.extractThinking] on [interaction].
+     * Provider model budget remains [Thinking.enabled] / [Thinking.tokenBudget].
      */
     private fun <O> doTransformObjectStreamInternal(
         messages: List<Message>,
@@ -204,14 +227,15 @@ internal class StreamingLlmOperationsImpl(
         // Create converter for JSONL parsing.
         // Spring AI 2.0's StreamingJacksonOutputConverter requires T : Any;
         // erase O via Class<Any> for the construction, cast back for downstream Flux<O>/StreamingEvent<O>.
+        val includeApplicationLevelThinking = interaction.llm.thinking?.extractThinking == true
         @Suppress("UNCHECKED_CAST")
         val outputClassAny = outputClass as Class<Any>
         @Suppress("UNCHECKED_CAST")
         val streamingConverter = StreamingJacksonOutputConverter<Any>(
             clazz = outputClassAny,
             objectMapper = objectMapper,
             fieldFilter = interaction.fieldFilter,
-            thinkingEnabled = interaction.llm.thinking?.enabled ?: false,
+            thinkingEnabled = includeApplicationLevelThinking,
         ) as StreamingJacksonOutputConverter<O>
 
         // Build prompt contributions with streaming format instructions

embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt

@@ -15,16 +15,20 @@
  */
 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
 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
@@ -37,6 +41,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.
@@ -79,7 +86,7 @@ class StreamingChatClientOperationsTest {
         every { mockChatClientLlmOperations.createChatClient(mockLlm) } returns mockChatClient
         every { mockInteraction.promptContributors } returns emptyList()
         every { mockLlm.promptContributors } returns emptyList()
-        val mockOptionsConverter = mockk<com.embabel.common.ai.model.OptionsConverter>(relaxed = true)
+        val mockOptionsConverter = mockk<OptionsConverter>(relaxed = true)
         every { mockLlm.optionsConverter } returns mockOptionsConverter
         every { mockOptionsConverter.convertOptions(any(), any()) } returns mockk(relaxed = true)
         every { mockInteraction.llm } returns mockk(relaxed = true)
@@ -161,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
@@ -468,16 +475,82 @@ class StreamingChatClientOperationsTest {
     }
 
 
-    private fun mockChatClientForStreaming(chunkFlux: Flux<String>) {
+    private fun mockChatClientForStreaming(chunkFlux: Flux<String>): CapturingSlot<Prompt> {
         val mockRequestSpec = mockk<ChatClient.ChatClientRequestSpec>(relaxed = true)
         val mockContentStreamSpec = mockk<ChatClient.StreamResponseSpec>(relaxed = true)
+        val promptSlot = slot<Prompt>()
 
-        every { mockChatClient.prompt(any<Prompt>()) } returns mockRequestSpec
+        every { mockChatClient.prompt(capture(promptSlot)) } returns mockRequestSpec
         every { mockRequestSpec.tools(any<List<ToolCallback>>()) } 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: 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")
+            )
+
+            // When
+            streamingOperations.createObjectStreamWithThinking(
+                messages = listOf(UserMessage("test")),
+                interaction = interaction,
+                outputClass = TestItem::class.java,
+                agentProcess = mockAgentProcess,
+                action = mockAction
+            ).collectList().block(Duration.ofSeconds(2))
+
+            // Then: format instructions still ask for <think> blocks (driven by extractThinking)
+            assertTrue(promptSlot.isCaptured, "expected ChatClient.prompt to be called")
+            val promptText = promptSlot.captured.contents
+            assertTrue(
+                promptText.contains("<think>"),
+                "createObjectStreamWithThinking should inject thinking format without Thinking.withTokenBudget"
+            )
+        }
+
+        @Test
+        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")
+            )
+
+            // When
+            streamingOperations.createObjectStream(
+                messages = listOf(UserMessage("test")),
+                interaction = interaction,
+                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("<think>"),
+                "tokenBudget alone should not inject application-level thinking format"
+            )
+        }
     }
 
     /**
@@ -520,7 +593,7 @@ class StreamingChatClientOperationsTest {
         @Test
         fun `should prepend prompt contributions as system message`() {
             // Given
-            val mockContributor = mockk<com.embabel.common.ai.prompt.PromptContributor>()
+            val mockContributor = mockk<PromptContributor>()
             every { mockContributor.contribution() } returns "System contribution"
             every { mockInteraction.promptContributors } returns listOf(mockContributor)