Streaming review

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

@@ -16,6 +16,7 @@
 package com.embabel.agent.api.common.streaming
 
 import com.embabel.agent.api.common.PromptRunner
+import com.embabel.agent.core.internal.streaming.toThinkingEvents
 import com.embabel.chat.Message
 import com.embabel.common.core.streaming.StreamingEvent
 import reactor.core.publisher.Flux
@@ -99,6 +100,18 @@ interface StreamingPromptRunner : PromptRunner {
          */
         fun generateStream(): Flux<String>
 
+        /**
+         * Generate a reactive thinking-only stream from newline-delimited text output.
+         *
+         * Object creation remains the responsibility of [createObjectStream]
+         * and [createObjectStreamWithThinking]. JSON values and bare code fences
+         * are omitted from this stream.
+         *
+         * @return Flux emitting thinking events
+         */
+        fun generateStreamWithThinking(): Flux<StreamingEvent<String>> =
+            generateStream().toThinkingEvents()
+
     }
 }
 

embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/DelegatingStreaming.kt

@@ -38,6 +38,9 @@ internal data class DelegatingStreaming(
         return delegate.generateStream()
     }
 
+    override fun generateStreamWithThinking(): Flux<StreamingEvent<String>> =
+        delegate.generateStreamWithThinking()
+
     override fun <T> createObjectStream(itemClass: Class<T>): Flux<T> =
         delegate.createObjectStream(itemClass)
 

embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt

@@ -341,6 +341,17 @@ internal data class OperationContextDelegate(
         )
     }
 
+    override fun generateStreamWithThinking(): Flux<StreamingEvent<String>> {
+        val streamingLlmOperations = streamingFactory().createStreamingOperations(llm)
+
+        return streamingLlmOperations.generateStreamWithThinking(
+            messages = messages,
+            interaction = streamingInteraction(),
+            agentProcess = context.processContext.agentProcess,
+            action = action,
+        )
+    }
+
     override fun <T> createObjectStream(itemClass: Class<T>): Flux<T> {
         val streamingLlmOperations = streamingFactory().createStreamingOperations(llm)
 

embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/PromptExecutionDelegate.kt

@@ -30,6 +30,7 @@ import com.embabel.agent.api.validation.guardrails.GuardRail
 import com.embabel.agent.core.ToolGroup
 import com.embabel.agent.core.ToolGroupRequirement
 import com.embabel.agent.core.internal.LlmOperations
+import com.embabel.agent.core.internal.streaming.toThinkingEvents
 import com.embabel.agent.core.support.LlmUse
 import com.embabel.agent.spi.loop.ToolInjectionStrategy
 import com.embabel.agent.spi.loop.ToolNotFoundPolicy
@@ -139,6 +140,9 @@ internal interface PromptExecutionDelegate : LlmUse {
 
     fun generateStream(): Flux<String>
 
+    fun generateStreamWithThinking(): Flux<StreamingEvent<String>> =
+        generateStream().toThinkingEvents()
+
     fun <T> createObjectStream(itemClass: Class<T>): Flux<T>
 
     fun <T> createObjectStreamWithThinking(itemClass: Class<T>): Flux<StreamingEvent<T>>

embabel-agent-api/src/main/kotlin/com/embabel/agent/core/internal/streaming/StreamingLlmOperations.kt

@@ -58,18 +58,32 @@ interface StreamingLlmOperations {
     ): Flux<String>
 
     /**
-     * Create a streaming list of objects from JSONL response in the context of an AgentProcess.
-     * Each line in the LLM response should be a valid JSON object matching the output class.
-     * Objects are emitted to the Flux as they are parsed from individual lines.
+     * Generate thinking events from text output.
+     *
+     * Object creation is intentionally excluded from this API.
+     */
+    fun generateStreamWithThinking(
+        messages: List<Message>,
+        interaction: LlmInteraction,
+        agentProcess: AgentProcess,
+        action: Action?,
+    ): Flux<StreamingEvent<String>> =
+        generateStream(messages, interaction, agentProcess, action)
+            .toThinkingEvents()
+
+    /**
+     * Create a streaming list of typed values from JSONL response in the context of an AgentProcess.
+     * Each line in the LLM response should be a valid JSON value matching the output class.
+     * Values are emitted to the Flux as they are parsed from individual lines.
      *
      * Supports the API layer createObjectStream() method.
      *
      * @param messages messages in the conversation so far
      * @param interaction Llm options and tool callbacks to use, plus unique identifier
-     * @param outputClass Class of the output objects
+     * @param outputClass Class of the output values
      * @param agentProcess Agent process we are running within
      * @param action Action we are running within if we are running within an action
-     * @return Flux of typed objects as they are parsed from the response
+     * @return Flux of typed values as they are parsed from the response
      */
     fun <O> createObjectStream(
         messages: List<Message>,
@@ -80,13 +94,13 @@ interface StreamingLlmOperations {
     ): Flux<O>
 
     /**
-     * Try to create a streaming list of objects in the context of an AgentProcess.
-     * Return a Flux that may error if the LLM does not have enough information to create objects.
+     * Try to create a streaming list of typed values in the context of an AgentProcess.
+     * Return a Flux that may error if the LLM does not have enough information to create values.
      * Streaming equivalent of createObjectIfPossible().
      *
      * @param messages messages
      * @param interaction Llm options and tool callbacks to use, plus unique identifier
-     * @param outputClass Class of the output objects
+     * @param outputClass Class of the output values
      * @param agentProcess Agent process we are running within
      * @param action Action we are running within if we are running within an action
      * @return Flux of Result<O> objects, where each Result indicates success/failure for that object
@@ -100,19 +114,19 @@ interface StreamingLlmOperations {
     ): Flux<Result<O>>
 
     /**
-     * Create a streaming list of objects with LLM thinking content from mixed JSONL response.
-     * Supports both JSON object lines and //THINKING: lines in the LLM response.
-     * Returns StreamingEvent objects that can contain either typed objects or thinking content.
+     * Create a streaming list of typed values with LLM thinking content from mixed JSONL response.
+     * Supports both JSON value lines and //THINKING: lines in the LLM response.
+     * Returns StreamingEvent objects that can contain either typed values or thinking content.
      *
      * This enables real-time visibility into LLM reasoning process alongside structured results.
      * Supports the API layer createObjectStreamWithThinking() method.
      *
      * @param messages messages in the conversation so far
      * @param interaction Llm options and tool callbacks to use, plus unique identifier
-     * @param outputClass Class of the output objects
+     * @param outputClass Class of the output values
      * @param agentProcess Agent process we are running within
      * @param action Action we are running within if we are running within an action
-     * @return Flux of StreamingEvent objects containing either objects or thinking content
+     * @return Flux of StreamingEvent objects containing either typed values or thinking content
      */
     fun <O> createObjectStreamWithThinking(
         messages: List<Message>,
@@ -142,18 +156,34 @@ interface StreamingLlmOperations {
         action: Action? = null,
     ): Flux<String>
 
+    /**
+     * Low-level thinking-only stream with optional platform context.
+     *
+     * The default applies the object-stream thinking classification to text
+     * output and deliberately omits object creation.
+     */
+    fun doTransformStreamWithThinking(
+        messages: List<Message>,
+        interaction: LlmInteraction,
+        llmRequestEvent: LlmRequestEvent<String>?,
+        agentProcess: AgentProcess? = null,
+        action: Action? = null,
+    ): Flux<StreamingEvent<String>> =
+        doTransformStream(messages, interaction, llmRequestEvent, agentProcess, action)
+            .toThinkingEvents()
+
     /**
      * Low level object streaming transform with optional platform context.
-     * Streams typed objects as they are parsed from JSONL response.
+     * Streams typed values as they are parsed from JSONL response.
      * When agentProcess is provided, tools are resolved from ToolGroups and decorated.
      *
      * @param messages messages in the conversation so far
      * @param interaction The LLM call options
-     * @param outputClass Class of the output objects
+     * @param outputClass Class of the output values
      * @param llmRequestEvent Event already published for this request if one has been
      * @param agentProcess Optional agent process for tool resolution and decoration
      * @param action Optional action context for tool decoration
-     * @return Flux of typed objects as they are parsed from the response
+     * @return Flux of typed values as they are parsed from the response
      */
     fun <O> doTransformObjectStream(
         messages: List<Message>,
@@ -166,12 +196,12 @@ interface StreamingLlmOperations {
 
     /**
      * Low level mixed content streaming transform with optional platform context.
-     * Streams both typed objects and thinking content from mixed JSONL response.
+     * Streams both typed values and thinking content from mixed JSONL response.
      * When agentProcess is provided, tools are resolved from ToolGroups and decorated.
      *
      * @param messages messages in the conversation so far
      * @param interaction The LLM call options
-     * @param outputClass Class of the output objects
+     * @param outputClass Class of the output values
      * @param llmRequestEvent Event already published for this request if one has been
      * @param agentProcess Optional agent process for tool resolution and decoration
      * @param action Optional action context for tool decoration

embabel-agent-api/src/main/kotlin/com/embabel/agent/core/internal/streaming/ThinkingStreamSupport.kt

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2024-2026 Embabel Pty Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.embabel.agent.core.internal.streaming
+
+import com.embabel.common.ai.converters.streaming.StreamingLineClassifier
+import com.embabel.common.core.streaming.StreamingEvent
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * Applies the same newline-based thinking classification used by object streaming,
+ * while intentionally omitting object creation. Non-JSON lines are emitted as
+ * [StreamingEvent.Thinking]; JSON lines and bare code fences are dropped.
+ *
+ * Line-assembly state is created inside [Flux.defer], keeping repeated and
+ * concurrent subscriptions isolated.
+ */
+internal fun Flux<String>.toThinkingEvents(): Flux<StreamingEvent<String>> =
+    Flux.defer {
+        val buffer = StringBuilder()
+        this@toThinkingEvents
+            .concatMap { chunk ->
+                buffer.append(chunk)
+                val lines = mutableListOf<String>()
+                var newline = buffer.indexOf("\n")
+                while (newline >= 0) {
+                    buffer.substring(0, newline).trim()
+                        .takeIf { it.isNotEmpty() }
+                        ?.let(lines::add)
+                    buffer.delete(0, newline + 1)
+                    newline = buffer.indexOf("\n")
+                }
+                Flux.fromIterable(lines)
+            }
+            .concatWith(
+                Mono.fromSupplier { buffer.toString().trim() }
+                    .filter { it.isNotEmpty() },
+            )
+            .concatMap(StreamingLineClassifier::classify)
+    }

embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt

@@ -113,6 +113,14 @@ internal class StreamingChatClientOperations(
         return doTransformStream(messages, interaction, null, agentProcess, action)
     }
 
+    override fun generateStreamWithThinking(
+        messages: List<Message>,
+        interaction: LlmInteraction,
+        agentProcess: AgentProcess,
+        action: Action?,
+    ): Flux<StreamingEvent<String>> =
+        doTransformStreamWithThinking(messages, interaction, null, agentProcess, action)
+
     override fun <O> createObjectStream(
         messages: List<Message>,
         interaction: LlmInteraction,

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.converters.streaming.StreamingLineClassifier
 import com.embabel.common.core.streaming.StreamingEvent
 import tools.jackson.databind.ObjectMapper
 import org.slf4j.LoggerFactory
@@ -50,7 +51,7 @@ import reactor.core.publisher.Mono
  * LLM framework (Spring AI, LangChain4j, etc.). It delegates raw streaming to
  * [LlmMessageStreamer] and handles:
  * - Line buffering from raw chunks
- * - JSONL parsing to typed objects
+ * - JSONL parsing to typed values
  * - Thinking content extraction
  *
  * @param messageStreamer The streamer for raw LLM content
@@ -80,6 +81,19 @@ internal class StreamingLlmOperationsImpl(
         return doTransformStream(messages, interaction, null, agentProcess, action)
     }
 
+    override fun generateStreamWithThinking(
+        messages: List<Message>,
+        interaction: LlmInteraction,
+        agentProcess: AgentProcess,
+        action: Action?,
+    ): Flux<StreamingEvent<String>> =
+        doTransformStream(messages, interaction, null, agentProcess, action)
+            // Reassemble arbitrary model chunks into the complete lines expected by the
+            // object-stream thinking detector.
+            .transform { rawChunksToLines(it) }
+            // Emits every non-JSON line as Thinking; JSON and formatting fences are dropped.
+            .concatMap(StreamingLineClassifier::classify)
+
     override fun <O> createObjectStream(
         messages: List<Message>,
         interaction: LlmInteraction,
@@ -335,29 +349,30 @@ internal class StreamingLlmOperationsImpl(
      * Convert raw streaming chunks to NDJSON lines.
      * Handles all cases: multiple \n in one chunk, no \n in chunk, line spanning many chunks.
      */
-    private fun rawChunksToLines(raw: Flux<String>): Flux<String> {
-        val buffer = StringBuilder()
-        return raw.concatMap { chunk ->
-            buffer.append(chunk)
-            val lines = mutableListOf<String>()
-            while (true) {
-                val idx = buffer.indexOf('\n')
-                if (idx < 0) break
-                val line = buffer.substring(0, idx).trim()
-                if (line.isNotEmpty()) lines.add(line)
-                buffer.delete(0, idx + 1)
-            }
-            Flux.fromIterable(lines)
-        }.doOnComplete {
-            if (buffer.isNotEmpty()) {
-                val finalLine = buffer.toString().trim()
-                if (finalLine.isNotEmpty()) {
-                    logger.trace("FINAL LINE: '$finalLine'")
+    private fun rawChunksToLines(raw: Flux<String>): Flux<String> =
+        Flux.defer {
+            val buffer = StringBuilder()
+            raw.concatMap { chunk ->
+                buffer.append(chunk)
+                val lines = mutableListOf<String>()
+                while (true) {
+                    val idx = buffer.indexOf('\n')
+                    if (idx < 0) break
+                    val line = buffer.substring(0, idx).trim()
+                    if (line.isNotEmpty()) lines.add(line)
+                    buffer.delete(0, idx + 1)
                 }
-            }
-        }.concatWith(
-            Mono.fromSupplier { buffer.toString().trim() }
-                .filter { it.isNotEmpty() }
-        )
-    }
+                Flux.fromIterable(lines)
+            }.doOnComplete {
+                if (buffer.isNotEmpty()) {
+                    val finalLine = buffer.toString().trim()
+                    if (finalLine.isNotEmpty()) {
+                        logger.trace("FINAL LINE: '$finalLine'")
+                    }
+                }
+            }.concatWith(
+                Mono.fromSupplier { buffer.toString().trim() }
+                    .filter { it.isNotEmpty() }
+            )
+        }
 }