fix(http): parse complete Java SSE data events

http-client/http-client-java/src/main/kotlin/ai/koog/http/client/java/JavaKoogHttpClient.kt

@@ -31,6 +31,8 @@ import kotlin.time.Duration.Companion.minutes
  * This client provides enhanced logging, flexible request and response handling, and supports
  * configurability for underlying Java HttpClient instances.
  *
+ * SSE dispatches complete UTF-8 data events, preserving multiline payloads and ignoring event metadata.
+ *
  * @property clientName The name of the client, used for logging and traceability.
  * @property logger A logging instance of type KLogger for recording client-related events and errors.
  * @property httpClient The configured Java HttpClient instance used for making HTTP requests.
@@ -203,24 +205,27 @@ public class JavaKoogHttpClient internal constructor(
 
             logger.debug { "SSE connection opened for $clientName" }
 
-            // Process the stream of lines
-            response.body().forEach { line ->
+            val data = StringBuilder()
+            var firstLine = true
+            response.body().forEach { rawLine ->
                 try {
-                    val dataPrefix = "data: "
-                    // SSE format: "data: <content>"
-                    val data = if (line.startsWith(dataPrefix)) {
-                        line.substring(dataPrefix.length)
-                    } else if (line.isNotEmpty() && !line.startsWith(":")) {
-                        line
-                    } else {
-                        null
-                    }
-
-                    if (data != null && dataFilter(data)) {
-                        data.trim()
-                            .let(decodeStreamingResponse)
-                            .let(processStreamingChunk)
-                            ?.let { trySend(it) }
+                    val line = if (firstLine) rawLine.removePrefix("\uFEFF") else rawLine
+                    firstLine = false
+                    if (line.isEmpty()) {
+                        if (data.isNotEmpty()) {
+                            val payload = data.dropLast(1).toString()
+                            data.setLength(0)
+                            if (dataFilter(payload)) {
+                                processStreamingChunk(decodeStreamingResponse(payload))?.let { trySend(it) }
+                            }
+                        }
+                    } else if (!line.startsWith(":")) {
+                        val colon = line.indexOf(':')
+                        val field = if (colon < 0) line else line.substring(0, colon)
+                        if (field == "data") {
+                            val value = if (colon < 0) "" else line.substring(colon + 1).removePrefix(" ")
+                            data.append(value).append('\n')
+                        }
                     }
                 } catch (e: CancellationException) {
                     throw e
@@ -235,6 +240,7 @@ public class JavaKoogHttpClient internal constructor(
                 }
             }
 
+            // Pending data without a terminating empty line belongs to an incomplete event.
             logger.debug { "SSE connection closed for $clientName" }
             close()
         } catch (e: CancellationException) {

http-client/http-client-java/src/test/kotlin/ai/koog/http/client/java/JavaKoogHttpClientSseFramingTest.kt

@@ -0,0 +1,97 @@
+package ai.koog.http.client.java
+
+import ai.koog.http.client.KoogHttpClient
+import ai.koog.http.client.test.MockWebServer
+import io.ktor.http.ContentType
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class JavaKoogHttpClientSseFramingTest {
+    @Test
+    fun testMultilineDataDispatchesAsOneEvent() = runTest {
+        assertEquals(listOf("first\nsecond"), collect("data: first\ndata:second\n\n"))
+    }
+
+    @Test
+    fun testMetadataAndCommentsAreIgnored() = runTest {
+        assertEquals(listOf("payload"), collect(": heartbeat\nevent: message\nid: 7\nretry: 100\nfuture: ignored\nraw text\ndata: payload\n\n"))
+    }
+
+    @Test
+    fun testLeadingBomAndAllLineEndings() = runTest {
+        assertEquals(listOf("one", "two", "three"), collect("\uFEFFdata: one\r\rdata: two\r\n\r\ndata: three\n\n"))
+    }
+
+    @Test
+    fun testDataWhitespaceIsPreserved() = runTest {
+        assertEquals(listOf(" leading  \n\ttab "), collect("data:  leading  \ndata:\ttab \n\n"))
+    }
+
+    @Test
+    fun testEmptyDataAndColonlessDataDispatch() = runTest {
+        assertEquals(listOf("", ""), collect("\nevent: ignored\n\ndata:\n\ndata\n\n"))
+    }
+
+    @Test
+    fun testIncompleteEventIsDiscardedAtEof() = runTest {
+        assertEquals(listOf("complete"), collect("data: complete\n\ndata: incomplete\n"))
+    }
+
+    @Test
+    fun testFilterAndDecoderReceiveCompleteDataEvents() = runTest {
+        val filtered = mutableListOf<String?>()
+        withResponse("data: first\ndata: second\n\ndata: skip\n\ndata: keep\n\n") { client ->
+            val result = client.sse(
+                path = "/stream",
+                requestBody = "{}",
+                requestBodyType = String::class,
+                dataFilter = {
+                    filtered.add(it)
+                    it != "skip"
+                },
+                decodeStreamingResponse = { it.uppercase() },
+                processStreamingChunk = { it.takeUnless { it == "FIRST\nSECOND" } },
+            ).flowOn(Dispatchers.IO).toList()
+            assertEquals(listOf<String?>("first\nsecond", "skip", "keep"), filtered)
+            assertEquals(listOf("KEEP"), result)
+        }
+    }
+
+    private suspend fun collect(response: String): List<String> {
+        var result = emptyList<String>()
+        withResponse(response) { client ->
+            result = client.sse(
+                path = "/stream",
+                requestBody = "{}",
+                requestBodyType = String::class,
+                decodeStreamingResponse = { it },
+                processStreamingChunk = { it },
+            ).flowOn(Dispatchers.IO).toList()
+        }
+        return result
+    }
+
+    private suspend fun withResponse(response: String, block: suspend (KoogHttpClient) -> Unit) {
+        val server = MockWebServer()
+        server.start(
+            postEndpoints = listOf(
+                MockWebServer.PostEndpointConfig(
+                    path = "/stream",
+                    responseBody = response,
+                    contentType = ContentType.Text.EventStream,
+                )
+            )
+        )
+        val client = JavaKoogHttpClient.Factory().create(clientName = "java-sse", baseUrl = server.url(""))
+        try {
+            block(client)
+        } finally {
+            client.close()
+            server.stop()
+        }
+    }
+}