From 3105225772fd5ada4e556de73bf67ed8085b6a57 Mon Sep 17 00:00:00 2001 From: kyboi Date: Tue, 29 Sep 2026 17:09:28 +0100 Subject: [PATCH 1/2] fix(openai): preserve Responses streaming tool identity --- .../clients/openai/OpenAILLMClient.kt | 22 +- .../openai/OpenAIPrimaryConstructorTest.kt | 244 ++++++++++++++++++ 2 files changed, 262 insertions(+), 4 deletions(-) diff --git a/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/openai/OpenAILLMClient.kt b/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/openai/OpenAILLMClient.kt index 5eafcdd7f5..53697fadab 100644 --- a/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/openai/OpenAILLMClient.kt +++ b/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/openai/OpenAILLMClient.kt @@ -63,6 +63,8 @@ import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.emitAll +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.withContext import kotlinx.serialization.json.JsonElement import kotlinx.serialization.json.JsonPrimitive @@ -348,7 +350,7 @@ public open class OpenAILLMClient @JvmOverloads constructor( model: LLModel, tools: List, params: OpenAIResponsesParams - ): Flow { + ): Flow = flow { logger.debug { "Executing streaming prompt: $prompt with model: $model" } val llmTools = tools.takeIf { it.isNotEmpty() }?.map { @@ -368,8 +370,9 @@ public open class OpenAILLMClient @JvmOverloads constructor( params = params, stream = true ) + val toolCallsByItemId = mutableMapOf() - return try { + val frames = try { httpClient.sse( path = settings.responsesAPIPath, requestBody = request, @@ -379,6 +382,14 @@ public open class OpenAILLMClient @JvmOverloads constructor( }, processStreamingChunk = { when (it) { + is OpenAIStreamEvent.ResponseOutputItemAdded -> { + val item = it.item + if (item is Item.FunctionToolCall) { + item.id?.let { itemId -> toolCallsByItemId[itemId] = item } + } + null + } + is OpenAIStreamEvent.ResponseOutputTextDelta -> { StreamFrame.TextDelta(text = it.delta, index = it.outputIndex) } @@ -394,9 +405,10 @@ public open class OpenAILLMClient @JvmOverloads constructor( } is OpenAIStreamEvent.ResponseFunctionCallArgumentsDelta -> { + val toolCall = toolCallsByItemId[it.itemId] StreamFrame.ToolCallDelta( - id = it.itemId, - name = null, + id = toolCall?.callId ?: it.itemId, + name = toolCall?.name, content = it.delta, index = it.outputIndex ) @@ -425,6 +437,7 @@ public open class OpenAILLMClient @JvmOverloads constructor( } is Item.FunctionToolCall -> { + item.id?.let { itemId -> toolCallsByItemId.remove(itemId) } StreamFrame.ToolCallComplete( id = item.callId, name = item.name, @@ -464,6 +477,7 @@ public open class OpenAILLMClient @JvmOverloads constructor( cause = e ) } + emitAll(frames) } override suspend fun executeMultipleChoices( diff --git a/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/jvmTest/kotlin/ai/koog/prompt/executor/clients/openai/OpenAIPrimaryConstructorTest.kt b/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/jvmTest/kotlin/ai/koog/prompt/executor/clients/openai/OpenAIPrimaryConstructorTest.kt index 3d2e0ee558..eb92fc3b6b 100644 --- a/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/jvmTest/kotlin/ai/koog/prompt/executor/clients/openai/OpenAIPrimaryConstructorTest.kt +++ b/prompt/prompt-executor/prompt-executor-clients/prompt-executor-openai-client/src/jvmTest/kotlin/ai/koog/prompt/executor/clients/openai/OpenAIPrimaryConstructorTest.kt @@ -17,14 +17,23 @@ import ai.koog.test.utils.CapturingKoogHttpClient import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList import kotlinx.coroutines.test.runTest +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonNamingStrategy import kotlin.reflect.KClass import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertIs class OpenAIPrimaryConstructorTest { + private val streamingJson = Json { + encodeDefaults = true + explicitNulls = false + namingStrategy = JsonNamingStrategy.SnakeCase + } + private val responseJson = """ { "id": "chatcmpl-123", @@ -220,4 +229,239 @@ class OpenAIPrimaryConstructorTest { assertEquals(inputTokens, end.metaInfo.inputTokensCount) assertEquals(outputTokens, end.metaInfo.outputTokensCount) } + + @Test + fun testResponsesStreamingToolCallDeltasUseCanonicalIdentity() = runTest { + val firstItemId = "item_first" + val firstCallId = "call_first" + val firstName = "firstTool" + val secondItemId = "item_second" + val secondCallId = "call_second" + val secondName = "secondTool" + val events = listOf( + OpenAIStreamEvent.ResponseOutputItemAdded( + item = Item.FunctionToolCall( + arguments = "", + callId = firstCallId, + name = firstName, + id = firstItemId, + status = OpenAIInputStatus.IN_PROGRESS + ), + outputIndex = 0, + sequenceNumber = 1 + ), + OpenAIStreamEvent.ResponseFunctionCallArgumentsDelta( + itemId = firstItemId, + outputIndex = 0, + delta = "{\"first\":", + sequenceNumber = 2 + ), + OpenAIStreamEvent.ResponseOutputItemAdded( + item = Item.FunctionToolCall( + arguments = "", + callId = secondCallId, + name = secondName, + id = secondItemId, + status = OpenAIInputStatus.IN_PROGRESS + ), + outputIndex = 1, + sequenceNumber = 3 + ), + OpenAIStreamEvent.ResponseFunctionCallArgumentsDelta( + itemId = secondItemId, + outputIndex = 1, + delta = "{\"second\":true}", + sequenceNumber = 4 + ), + OpenAIStreamEvent.ResponseFunctionCallArgumentsDelta( + itemId = firstItemId, + outputIndex = 0, + delta = "true}", + sequenceNumber = 5 + ), + OpenAIStreamEvent.ResponseOutputItemDone( + item = Item.FunctionToolCall( + arguments = "{\"second\":true}", + callId = secondCallId, + name = secondName, + id = secondItemId, + status = OpenAIInputStatus.COMPLETED + ), + outputIndex = 1, + sequenceNumber = 6 + ), + OpenAIStreamEvent.ResponseOutputItemDone( + item = Item.FunctionToolCall( + arguments = "{\"first\":true}", + callId = firstCallId, + name = firstName, + id = firstItemId, + status = OpenAIInputStatus.COMPLETED + ), + outputIndex = 0, + sequenceNumber = 7 + ), + completedResponseEvent(sequenceNumber = 8) + ) + val client = OpenAILLMClient( + settings = OpenAIClientSettings(baseUrl = "https://unused.test"), + httpClient = streamingTransport { events } + ) + + val frames = client.executeStreaming( + prompt = Prompt( + messages = listOf(Message.User("Use both tools", RequestMetaInfo.Empty)), + id = "test", + params = OpenAIResponsesParams() + ), + model = OpenAIModels.Chat.GPT4o + ).toList() + + assertEquals( + listOf( + StreamFrame.ToolCallDelta(firstCallId, firstName, "{\"first\":", 0), + StreamFrame.ToolCallDelta(secondCallId, secondName, "{\"second\":true}", 1), + StreamFrame.ToolCallDelta(firstCallId, firstName, "true}", 0), + StreamFrame.ToolCallComplete(secondCallId, secondName, "{\"second\":true}", 1), + StreamFrame.ToolCallComplete(firstCallId, firstName, "{\"first\":true}", 0) + ), + frames.dropLast(1) + ) + assertIs(frames.last()) + } + + @Test + fun testResponsesStreamingFallsBackForUnknownAndCompletedItems() = runTest { + val events = listOf( + OpenAIStreamEvent.ResponseOutputItemAdded(Item.Text("ignored"), 0, 1), + OpenAIStreamEvent.ResponseOutputItemAdded(toolCall(id = null), 0, 2), + argumentDelta("unknown", 3), + OpenAIStreamEvent.ResponseOutputItemAdded(toolCall(), 0, 4), + argumentDelta("item", 5), + OpenAIStreamEvent.ResponseOutputItemDone(toolCall(), 0, 6), + argumentDelta("item", 7), + OpenAIStreamEvent.ResponseOutputItemDone(toolCall(id = null), 0, 8), + completedResponseEvent(9) + ) + val frames = toolCallStream(streamingTransport { events }).toList() + + assertEquals( + listOf( + StreamFrame.ToolCallDelta("unknown", null, "{}", 0), + StreamFrame.ToolCallDelta("call", "lookup", "{}", 0), + StreamFrame.ToolCallComplete("call", "lookup", "{}", 0), + StreamFrame.ToolCallDelta("item", null, "{}", 0), + StreamFrame.ToolCallComplete("call", "lookup", "{}", 0) + ), + frames.dropLast(1) + ) + assertIs(frames.last()) + } + + @Test + fun testResponsesStreamingDoesNotRetainIdentityAfterCancelledCollection() = runTest { + var collection = 0 + val stream = toolCallStream( + streamingTransport { + collection++ + if (collection == 1) { + listOf( + OpenAIStreamEvent.ResponseOutputItemAdded(toolCall(), 0, 1), + argumentDelta("item", 2), + completedResponseEvent(3) + ) + } else { + listOf(argumentDelta("item", 1), completedResponseEvent(2)) + } + } + ) + + assertEquals(listOf(StreamFrame.ToolCallDelta("call", "lookup", "{}", 0)), stream.take(1).toList()) + val recollected = stream.toList() + assertEquals(StreamFrame.ToolCallDelta("item", null, "{}", 0), recollected.first()) + assertIs(recollected.last()) + } + + private fun toolCall(id: String? = "item") = Item.FunctionToolCall( + arguments = "{}", + callId = "call", + name = "lookup", + id = id + ) + + private fun argumentDelta(itemId: String, sequenceNumber: Int) = + OpenAIStreamEvent.ResponseFunctionCallArgumentsDelta(itemId, 0, "{}", sequenceNumber) + + private fun toolCallStream(transport: KoogHttpClient): Flow = OpenAILLMClient( + settings = OpenAIClientSettings(baseUrl = "https://unused.test"), + httpClient = transport + ).executeStreaming( + prompt = Prompt( + messages = listOf(Message.User("Use lookup", RequestMetaInfo.Empty)), + id = "test", + params = OpenAIResponsesParams() + ), + model = OpenAIModels.Chat.GPT4o + ) + + private fun streamingTransport(events: () -> List): KoogHttpClient = object : KoogHttpClient { + override val clientName: String = "StreamingOpenAIClient" + + override suspend fun get( + path: String, + responseType: KClass, + parameters: Map, + headers: Map, + ): R = error("GET is not expected in this test") + + override suspend fun post( + path: String, + requestBody: T, + requestBodyType: KClass, + responseType: KClass, + parameters: Map, + headers: Map, + ): R = error("POST is not expected in this test") + + override fun sse( + path: String, + requestBody: T, + requestBodyType: KClass, + dataFilter: (String?) -> Boolean, + decodeStreamingResponse: (String) -> R, + processStreamingChunk: (R) -> O?, + parameters: Map, + headers: Map, + ): Flow = flow { + events().forEach { event -> + val data = streamingJson.encodeToString(OpenAIStreamEvent.serializer(), event) + processStreamingChunk(decodeStreamingResponse(data))?.let { emit(it) } + } + } + + override fun lines( + path: String, + requestBody: T, + requestBodyType: KClass, + parameters: Map, + headers: Map, + ): Flow = error("lines is not expected in this test") + + override fun close(): Unit = Unit + } + + private fun completedResponseEvent(sequenceNumber: Int): OpenAIStreamEvent.ResponseCompleted = + OpenAIStreamEvent.ResponseCompleted( + response = OpenAIResponsesAPIResponse( + created = 1716920005, + id = "response_id", + model = "gpt-5", + output = emptyList(), + parallelToolCalls = true, + status = OpenAIInputStatus.COMPLETED, + text = OpenAITextConfig(), + usage = null + ), + sequenceNumber = sequenceNumber + ) } From 050d91f96d4b7408f8e00d593587a14ec04d7bec Mon Sep 17 00:00:00 2001 From: kyboi Date: Tue, 29 Sep 2026 17:35:51 +0100 Subject: [PATCH 2/2] fix(prompt): preserve interleaved streaming tool calls --- .../streaming/StreamFrameFlowBuilder.kt | 71 +++++---- .../streaming/StreamFrameFlowBuilderTest.kt | 143 +++++++++++++++++- 2 files changed, 180 insertions(+), 34 deletions(-) diff --git a/prompt/prompt-model/src/commonMain/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilder.kt b/prompt/prompt-model/src/commonMain/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilder.kt index 6f8ec00b9d..e7f1c9408c 100644 --- a/prompt/prompt-model/src/commonMain/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilder.kt +++ b/prompt/prompt-model/src/commonMain/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilder.kt @@ -125,7 +125,9 @@ public fun buildStreamFrameFlow(block: suspend StreamFrameFlowBuilder.() -> Unit /** * Represents a wrapper around a [FlowCollector] that provides methods for emitting [StreamFrame] objects. * - * This is mainly used for combining chunked tool calls and only emit completed tool calls. + * Combines chunked tool calls by index or ID, allowing their argument deltas to interleave. + * Tool calls complete in first-seen order when text, reasoning or an end frame arrives, + * or when [tryEmitPendingToolCall] is called explicitly. * * @property flowCollector The underlying [FlowCollector] used for emitting [StreamFrame] objects. */ @@ -134,7 +136,7 @@ public class StreamFrameFlowBuilder( private val flowCollector: FlowCollector, ) { - private val pendingToolCallRef = AtomicReference(null) + private val pendingToolCallsRef = AtomicReference>(emptyList()) private val pendingTextRef = AtomicReference(null) private val pendingReasoningRef = AtomicReference(null) @@ -182,10 +184,13 @@ public class StreamFrameFlowBuilder( } /** - * Updates the coroutine context to signal we're currently combining a tool call, - * this does not emit anything yet, that happens only in [tryEmitPendingToolCall]. + * Emits a tool call delta and accumulates its arguments by index or ID. + * Missing identity fields are filled when a later delta supplies them. + * A delta without an index or ID continues the sole pending call. * - * @throws StreamFrameFlowBuilderError if there is + * @throws StreamFrameFlowBuilderError.NoPartialToolCallToComplete if an anonymous delta has no pending call. + * @throws IllegalStateException if an anonymous delta has multiple possible pending calls. + * @throws IllegalArgumentException if the ID and index conflict with pending call identity. */ public suspend fun emitToolCallDelta( id: String? = null, @@ -196,29 +201,26 @@ public class StreamFrameFlowBuilder( tryEmitPendingText() tryEmitPendingReasoning() val sanitizedId = id?.takeUnless { it.isBlank() } - val previous: PendingToolCall? = pendingToolCallRef.load() - // `id` and `index` are optional per-chunk signals. A new tool call begins only when a - // present signal differs from the pending call, not merely because `id` is non-null: - // some OpenAI-compatible providers send the same `id` on every chunk (see #2002). - val isNewToolCall = - (sanitizedId != null && sanitizedId != previous?.id) || - (index != null && index != previous?.index) - val new: PendingToolCall = if (isNewToolCall) { - tryEmitPendingToolCall() - PendingToolCall(sanitizedId, name, args, index) + val calls = pendingToolCallsRef.load() + // Adapted from Kroog's interleaved tool-call aggregation (Kreoh/kroog@4ef675dc1). + val indexPosition = index?.let { value -> calls.indexOfFirst { it.index == value }.takeIf { it >= 0 } } + val idPosition = sanitizedId?.let { value -> calls.indexOfFirst { it.id == value }.takeIf { it >= 0 } } + require(indexPosition == null || idPosition == null || indexPosition == idPosition) { + "Tool call ID $sanitizedId and index $index identify different pending calls." + } + val position = indexPosition ?: idPosition ?: when { + index != null || sanitizedId != null -> null + calls.isEmpty() -> throw StreamFrameFlowBuilderError.NoPartialToolCallToComplete() + calls.size == 1 -> 0 + else -> error("A tool call delta without an ID or index is ambiguous while multiple calls are pending.") + } + val updated = if (position == null) { + calls + PendingToolCall(sanitizedId, name, args, index) } else { - when { - previous == null -> - throw StreamFrameFlowBuilderError.NoPartialToolCallToComplete() - - previous.index != index -> - throw StreamFrameFlowBuilderError.UnexpectedPartialToolCallIndex(previous.index, index) - - else -> - previous.appendArgumentsDelta(args) - } + val call = calls[position].appendDelta(sanitizedId, name, args, index) + calls.mapIndexed { currentPosition, pending -> if (currentPosition == position) call else pending } } - pendingToolCallRef.store(new) + pendingToolCallsRef.store(updated) flowCollector.emitToolCallDelta(sanitizedId, name, args, index) } @@ -251,11 +253,11 @@ public class StreamFrameFlowBuilder( } /** - * Emits a [pendingToolCallRef] if it exists and then clears it. + * Emits all pending tool calls in first-seen order and then clears them. */ public suspend fun tryEmitPendingToolCall() { - val pendingToolCall = pendingToolCallRef.exchange(null) - if (pendingToolCall != null) { + val calls = pendingToolCallsRef.exchange(emptyList()) + calls.forEach { pendingToolCall -> flowCollector.emitToolCallComplete( id = pendingToolCall.id, name = pendingToolCall.name ?: "", @@ -271,11 +273,16 @@ public class StreamFrameFlowBuilder( val argumentsDelta: String?, val index: Int?, ) { - fun appendArgumentsDelta(argumentsDelta: String?): PendingToolCall { - require(this.index == index) + fun appendDelta(id: String?, name: String?, argumentsDelta: String?, index: Int?): PendingToolCall { + require(this.id == null || id == null || this.id == id) { + "Tool call index ${this.index} has conflicting IDs ${this.id} and $id." + } + require(this.index == null || index == null || this.index == index) { + "Tool call ID ${this.id} has conflicting indices ${this.index} and $index." + } val newArgs = if (argumentsDelta == null) this.argumentsDelta else (this.argumentsDelta ?: "") + argumentsDelta - return copy(argumentsDelta = newArgs) + return copy(id = this.id ?: id, name = this.name ?: name, argumentsDelta = newArgs, index = this.index ?: index) } } diff --git a/prompt/prompt-model/src/commonTest/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilderTest.kt b/prompt/prompt-model/src/commonTest/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilderTest.kt index 8df7df289b..c90b76e394 100644 --- a/prompt/prompt-model/src/commonTest/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilderTest.kt +++ b/prompt/prompt-model/src/commonTest/kotlin/ai/koog/prompt/streaming/StreamFrameFlowBuilderTest.kt @@ -214,9 +214,9 @@ class StreamFrameFlowBuilderTest { listOf( StreamFrame.ToolCallDelta("call_1", "calculator", "{\"a\":", 0), StreamFrame.ToolCallDelta(null, null, " 5}", 0), - StreamFrame.ToolCallComplete("call_1", "calculator", "{\"a\": 5}", 0), StreamFrame.ToolCallDelta("call_2", "calculator", "{\"b\":", 1), StreamFrame.ToolCallDelta(null, null, " 6}", 1), + StreamFrame.ToolCallComplete("call_1", "calculator", "{\"a\": 5}", 0), StreamFrame.ToolCallComplete("call_2", "calculator", "{\"b\": 6}", 1), StreamFrame.End(null, ResponseMetaInfo.Empty), ), @@ -271,9 +271,9 @@ class StreamFrameFlowBuilderTest { listOf( StreamFrame.ToolCallDelta("call_1", "search", "{\"q\":", 0), StreamFrame.ToolCallDelta("call_1", null, " 1}", 0), - StreamFrame.ToolCallComplete("call_1", "search", "{\"q\": 1}", 0), StreamFrame.ToolCallDelta("call_2", "calculator", "{\"a\":", 1), StreamFrame.ToolCallDelta("call_2", null, " 2}", 1), + StreamFrame.ToolCallComplete("call_1", "search", "{\"q\": 1}", 0), StreamFrame.ToolCallComplete("call_2", "calculator", "{\"a\": 2}", 1), StreamFrame.End(null, ResponseMetaInfo.Empty) ), @@ -472,4 +472,143 @@ class StreamFrameFlowBuilderTest { frames ) } + + @Test + fun testInterleavedToolCallsCompleteAtEnd() = runTest { + val frames = buildStreamFrameFlow { + emitToolCallDelta("a", "search", "{\"a\":", 0) + emitToolCallDelta("b", "search", "{\"b\":", 1) + emitToolCallDelta(args = "1}", index = 0) + emitToolCallDelta(args = "2}", index = 1) + emitEnd() + }.toList() + + assertContentEquals( + listOf( + StreamFrame.ToolCallDelta("a", "search", "{\"a\":", 0), + StreamFrame.ToolCallDelta("b", "search", "{\"b\":", 1), + StreamFrame.ToolCallDelta(null, null, "1}", 0), + StreamFrame.ToolCallDelta(null, null, "2}", 1), + StreamFrame.ToolCallComplete("a", "search", "{\"a\":1}", 0), + StreamFrame.ToolCallComplete("b", "search", "{\"b\":2}", 1), + StreamFrame.End(null, ResponseMetaInfo.Empty) + ), + frames + ) + } + + // Late-ID regression scenario from jewoodev's https://github.com/JetBrains/koog/pull/2134. + @Test + fun testToolCallIdArrivingAfterArguments() = runTest { + val frames = buildStreamFrameFlow { + emitToolCallDelta(id = null, name = "search", args = "{\"query\":", index = 0) + emitToolCallDelta(id = "call_1", args = "\"koog\"}", index = 0) + emitEnd() + }.toList() + + assertContentEquals( + listOf( + StreamFrame.ToolCallDelta(null, "search", "{\"query\":", 0), + StreamFrame.ToolCallDelta("call_1", null, "\"koog\"}", 0), + StreamFrame.ToolCallComplete("call_1", "search", "{\"query\":\"koog\"}", 0), + StreamFrame.End(null, ResponseMetaInfo.Empty) + ), + frames + ) + } + + @Test + fun testInterleavedToolCallsWithoutIndices() = runTest { + val frames = buildStreamFrameFlow { + emitToolCallDelta("a", "search", "first") + emitToolCallDelta("b", "search", "second") + emitToolCallDelta(id = "b", args = " result") + emitToolCallDelta(id = "a", args = " result") + emitEnd() + }.toList() + + assertContentEquals( + listOf( + StreamFrame.ToolCallComplete("a", "search", "first result"), + StreamFrame.ToolCallComplete("b", "search", "second result") + ), + frames.filterIsInstance() + ) + } + + @Test + fun testToolCallEnrichesMissingNameAndIndex() = runTest { + val frames = buildStreamFrameFlow { + emitToolCallDelta(id = "a", args = "{") + emitToolCallDelta(id = "a", name = "search", index = 0) + emitToolCallDelta(args = "}") + emitEnd() + }.toList() + + assertContentEquals( + listOf(StreamFrame.ToolCallComplete("a", "search", "{}", 0)), + frames.filterIsInstance() + ) + } + + @Test + fun testPendingToolCallsDrainAtEachBoundaryAndClear() = runTest { + val boundaries: List Unit> = listOf( + { emitTextDelta("text") }, + { emitReasoningDelta(text = "reasoning") }, + { tryEmitPendingToolCall() }, + { emitEnd() } + ) + for (boundary in boundaries) { + val frames = buildStreamFrameFlow { + emitToolCallDelta("b", "search", index = 1) + emitToolCallDelta("a", "search", index = 0) + boundary() + emitToolCallDelta("c", "search", index = 0) + emitEnd() + tryEmitPendingToolCall() + }.toList() + + assertContentEquals( + listOf( + StreamFrame.ToolCallComplete("b", "search", "{}", 1), + StreamFrame.ToolCallComplete("a", "search", "{}", 0), + StreamFrame.ToolCallComplete("c", "search", "{}", 0) + ), + frames.filterIsInstance() + ) + val completions = frames.filterIsInstance() + assertContentEquals(completions.take(2), frames.drop(2).take(2)) + } + } + + @Test + fun testAnonymousDeltaWithMultiplePendingCallsThrows() = runTest { + assertFailsWith { + buildStreamFrameFlow { + emitToolCallDelta("a", "search", index = 0) + emitToolCallDelta("b", "search", index = 1) + emitToolCallDelta(args = "{}") + }.collect() + } + } + + @Test + fun testConflictingToolCallIdentityThrows() = runTest { + val conflictingDeltas = listOf( + StreamFrame.ToolCallDelta("b", null, "{}", 0), + StreamFrame.ToolCallDelta("a", null, "{}", 1), + StreamFrame.ToolCallDelta("c", null, "{}", 0), + StreamFrame.ToolCallDelta("a", null, "{}", 2) + ) + for (delta in conflictingDeltas) { + assertFailsWith { + buildStreamFrameFlow { + emitToolCallDelta("a", "search", index = 0) + emitToolCallDelta("b", "search", index = 1) + emitToolCallDelta(delta.id, delta.name, delta.content, delta.index) + }.collect() + } + } + } }