KG-616 fix streaming + tool call issue (#1262)

Related to [KG-616](https://youtrack.jetbrains.com/issue/KG-616)

## Motivation and Context
This PR addresses critical issues encountered with SSE streaming for
both Google and OpenRouter clients, ensuring robust functionality and
better error handling.

1.  **Google Client Fix:**
* **Problem:** The
[GoogleLLMClient](cci:2://file:///Users/ku76uh/Developer/jetbrains/fork/koog/prompt/prompt-executor/prompt-executor-clients/prompt-executor-google-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/google/GoogleLLMClient.kt:98:0-797:1)
was ignoring the `tools` parameter during streaming requests
([executeStreaming](cci:1://file:///Users/ku76uh/Developer/jetbrains/fork/koog/prompt/prompt-executor/prompt-executor-model/src/commonMain/kotlin/ai/koog/prompt/executor/model/PromptExecutor.kt:36:4-48:24)),
explicitly passing `emptyList()` instead. This broke tool usage for
streaming calls.
* **Solution:** We corrected this to specificially pass the `tools`
parameter to
[createGoogleRequest](cci:1://file:///Users/ku76uh/Developer/jetbrains/fork/koog/prompt/prompt-executor/prompt-executor-clients/prompt-executor-google-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/google/GoogleLLMClient.kt:277:4-446:5).

2.  **OpenRouter Integration Test Fix:**
* **Problem:** Integration tests for OpenRouter streaming were failing
with `Socket is not connected` errors. This was traced back to the
default `Apache5` HTTP client engine used in tests, which has known
connection pooling issues with OpenRouter's non-standard Keep-Alive
comments (e.g., `: OPENROUTER PROCESSING`).
* **Solution:** We switched the integration test runtime dependency from
`ktor-client-apache5` to `ktor-client-cio`. The CIO (Coroutine-based
I/O) engine handles these SSE keep-alive comments correctly, resolving
the socket closure issues.

3.  **Enhanced SSE Error Handling:**
* **Problem:** When a `SSEClientException` occurred (e.g., due to a 400
Bad Request or streaming interruption), the error handling logic in
[KtorKoogHttpClient](cci:2://file:///Users/ku76uh/Developer/jetbrains/fork/koog/http-client/http-client-ktor/src/commonMain/kotlin/ai/koog/http/client/ktor/KtorKoogHttpClient.kt:49:0-191:1)
would attempt to read `e.response?.bodyAsText()` directly. If the
response body was already consumed or unavailable (common in streaming
errors), this would throw a `DoubleReceiveException`, masking the
original error.
* **Solution:** We wrapped the `bodyAsText()` call in a `try-catch`
block. This allows us to successfully capture the error body for
handshake failures (like 400 Bad Request) while gracefully handling
cases where the body is unavailable (falling back to `null`), ensuring
the original `SSEClientException` is always propagated with maximum
available context.

## Breaking Changes
None. These are internal fixes and test configuration updates.

---

#### Type of the changes
- [ ] New feature (non-breaking change which adds functionality)
- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [x] Tests improvement
- [ ] Refactoring

#### Checklist
- [x] The pull request has a description of the proposed change
- [x] I read the [Contributing
Guidelines](https://github.com/JetBrains/koog/blob/main/CONTRIBUTING.md)
before opening the pull request
- [x] The pull request uses **`develop`** as the base branch
- [x] Tests for the changes have been added
- [x] All new and existing tests passed

##### Additional steps for pull requests adding a new feature
- [x] An issue describing the proposed change exists
- [x] The pull request includes a link to the issue
- [ ] The change was discussed and approved in the issue
- [ ] Docs have been added / updated

---------

Co-authored-by: Malte Heuser <[email protected]>

Author: mltheuser

http-client/http-client-ktor/src/commonMain/kotlin/ai/koog/http/client/ktor/KtorKoogHttpClient.kt

@@ -161,10 +161,17 @@ public class KtorKoogHttpClient internal constructor(
                 }
             }
         } catch (e: SSEClientException) {
+            val errorBody = try {
+                e.response?.bodyAsText()
+            } catch (ignored: Exception) {
+                logger.debug(ignored) { "Unable to read SSE error response body (may already be consumed)" }
+                null
+            }
             throw KoogHttpClientException(
                 clientName = clientName,
                 statusCode = e.response?.status?.value,
-                errorBody = e.response?.bodyAsText(),
+                errorBody = errorBody,
+                message = e.message,
                 cause = e
             )
         } catch (e: CancellationException) {

integration-tests/build.gradle.kts

@@ -17,7 +17,7 @@ kotlin {
                 implementation(libs.testcontainers)
                 implementation(libs.ktor.server.netty)
                 implementation(kotlin("test-junit5"))
-                runtimeOnly(libs.ktor.client.apache5)
+                runtimeOnly(libs.ktor.client.cio)
                 runtimeOnly(libs.slf4j.simple)
             }
         }

integration-tests/src/jvmTest/kotlin/ai/koog/integration/tests/executor/ExecutorIntegrationTestBase.kt

@@ -89,7 +89,7 @@ import org.junit.jupiter.api.Assumptions.assumeTrue
 import org.junit.jupiter.api.BeforeAll
 import java.nio.file.Path
 import java.nio.file.Paths
-import java.util.Base64
+import java.util.*
 import kotlin.io.path.pathString
 import kotlin.io.path.readBytes
 import kotlin.io.path.readText
@@ -1119,6 +1119,59 @@ abstract class ExecutorIntegrationTestBase {
             answer.shouldContain("20")
         }
     }
+
+    open fun integration_testExecuteStreamingWithTools(model: LLModel) = runTest(timeout = 300.seconds) {
+        Models.assumeAvailable(model.provider)
+        assumeTrue(model.capabilities.contains(LLMCapability.Tools), "Model $model does not support tools")
+
+        val executor = getExecutor(model)
+
+        val prompt = Prompt.build("test-streaming", LLMParams(toolChoice = ToolChoice.Required)) {
+            system("You are a helpful assistant.")
+            user("Count three times five")
+        }
+
+        withRetry(times = 3, testName = "integration_testExecuteStreamingWithTools[${model.id}]") {
+            with(StringBuilder()) {
+                val endMessages = mutableListOf<StreamFrame.End>()
+                val toolMessages = mutableListOf<StreamFrame.ToolCall>()
+
+                executor.executeStreamAndCollect(
+                    prompt = prompt,
+                    model = model,
+                    tools = listOf(SimpleCalculatorTool.descriptor),
+                    appendable = this,
+                    endMessages = endMessages,
+                    toolMessages = toolMessages
+                )
+
+                toolMessages.shouldNotBeEmpty()
+                withClue("Expected calculator tool call but got: [$toolMessages]") {
+                    toolMessages.any {
+                        it.name == SimpleCalculatorTool.name &&
+                            it.content.contains(CalculatorOperation.MULTIPLY.name, ignoreCase = true)
+                    } shouldBe true
+                }
+            }
+        }
+    }
+
+    private suspend fun PromptExecutor.executeStreamAndCollect(
+        prompt: Prompt,
+        model: LLModel,
+        tools: List<ToolDescriptor>,
+        appendable: StringBuilder,
+        endMessages: MutableList<StreamFrame.End>,
+        toolMessages: MutableList<StreamFrame.ToolCall>
+    ) {
+        executeStreaming(prompt, model, tools).collect { frame ->
+            when (frame) {
+                is StreamFrame.Append -> appendable.append(frame.text)
+                is StreamFrame.ToolCall -> toolMessages.add(frame)
+                is StreamFrame.End -> endMessages.add(frame)
+            }
+        }
+    }
 }
 
 private suspend fun PromptExecutor.executeStreamAndCollect(

integration-tests/src/jvmTest/kotlin/ai/koog/integration/tests/executor/MultipleLLMPromptExecutorIntegrationTest.kt

@@ -121,6 +121,12 @@ class MultipleLLMPromptExecutorIntegrationTest : ExecutorIntegrationTestBase() {
         super.integration_testExecuteStreamingWithTools(model)
     }
 
+    @ParameterizedTest
+    @MethodSource("allCompletionModels")
+    override fun integration_testExecuteStreamingWithTools(model: LLModel) {
+        super.integration_testExecuteStreamingWithTools(model)
+    }
+
     @ParameterizedTest
     @MethodSource("allCompletionModels")
     override fun integration_testToolWithRequiredParams(model: LLModel) {

integration-tests/src/jvmTest/kotlin/ai/koog/integration/tests/executor/SingleLLMPromptExecutorIntegrationTest.kt

@@ -115,6 +115,12 @@ class SingleLLMPromptExecutorIntegrationTest : ExecutorIntegrationTestBase() {
         super.integration_testExecuteStreamingWithTools(model)
     }
 
+    @ParameterizedTest
+    @MethodSource("allCompletionModels")
+    override fun integration_testExecuteStreamingWithTools(model: LLModel) {
+        super.integration_testExecuteStreamingWithTools(model)
+    }
+
     @ParameterizedTest
     @MethodSource("allCompletionModels")
     override fun integration_testToolWithRequiredParams(model: LLModel) {

prompt/prompt-executor/prompt-executor-clients/prompt-executor-google-client/src/commonMain/kotlin/ai/koog/prompt/executor/clients/google/GoogleLLMClient.kt

@@ -172,7 +172,7 @@ public open class GoogleLLMClient(
             "Model ${model.id} does not support chat completions"
         }
 
-        val request = createGoogleRequest(prompt, model, emptyList())
+        val request = createGoogleRequest(prompt, model, tools)
 
         try {
             httpClient.sse(