From 4a6c1f9a294ea62683a6aa7f1381a91e644c9478 Mon Sep 17 00:00:00 2001 From: Garfie Date: Tue, 14 Apr 2026 07:15:13 -0500 Subject: [PATCH] fix(ios): stream SSE via NWConnection to bypass NSURLSession text/* buffering --- CLAUDE.md | 1 + .../librechat/core/data/di/DataModule.kt | 1 - .../data/repository/ChatRepositoryImpl.kt | 6 +- core/network/CLAUDE.md | 41 +- core/network/build.gradle.kts | 8 + .../di/NetworkPlatformModule.android.kt | 3 + .../network/sse/SseHttpTransport.android.kt | 56 +++ .../core/network/di/NetworkModule.kt | 4 +- .../core/network/sse/HttpResponseParser.kt | 334 +++++++++++++ .../librechat/core/network/sse/SseClient.kt | 90 ++-- .../core/network/sse/SseHttpTransport.kt | 41 ++ .../network/sse/HttpResponseParserTest.kt | 275 +++++++++++ .../iosMain/cinterop/nwparams_defaults.def | 23 + .../network/di/NetworkPlatformModule.ios.kt | 3 + .../core/network/sse/SseHttpTransport.ios.kt | 450 ++++++++++++++++++ iosApp/iosApp/KoinHelper.swift | 5 - .../garfiec/librechat/shared/LibreChatSDK.kt | 3 +- .../librechat/shared/IosKoinAccessor.kt | 6 - .../librechat/shared/IosSharedModule.kt | 30 +- 19 files changed, 1286 insertions(+), 94 deletions(-) create mode 100644 core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.android.kt create mode 100644 core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParser.kt create mode 100644 core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.kt create mode 100644 core/network/src/commonTest/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParserTest.kt create mode 100644 core/network/src/iosMain/cinterop/nwparams_defaults.def create mode 100644 core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.ios.kt diff --git a/CLAUDE.md b/CLAUDE.md index 1899298..7ede6c7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -57,6 +57,7 @@ Each module has its own `CLAUDE.md` with specific guidance. - `GET /api/config` drives feature availability — never hardcode - ua-parser-js middleware rejects non-browser User-Agents with 403 (workaround: Chrome UA string) - Refresh token sent via request body (not HTTP-only cookies) +- iOS SSE streaming uses a custom `NWConnection`-based HTTP/1.1 transport (`core/network/src/iosMain/.../sse/SseHttpTransport.ios.kt`) to bypass NSURLSession's undocumented `text/*` content-type buffering. See `core/network/CLAUDE.md` SSE section for the two-layer buffering story. ## Upstream Sync diff --git a/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/di/DataModule.kt b/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/di/DataModule.kt index 7155cf2..32d8e8e 100644 --- a/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/di/DataModule.kt +++ b/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/di/DataModule.kt @@ -98,7 +98,6 @@ val dataModule = module { ChatRepositoryImpl( chatApi = get(), sseClient = get(), - sseHttpClient = get(KoinQualifiers.Streaming), connectivityObserver = get(), ) } diff --git a/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/repository/ChatRepositoryImpl.kt b/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/repository/ChatRepositoryImpl.kt index 1d8718e..d2c9174 100644 --- a/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/repository/ChatRepositoryImpl.kt +++ b/core/data/src/commonMain/kotlin/com/garfiec/librechat/core/data/repository/ChatRepositoryImpl.kt @@ -10,7 +10,6 @@ import com.garfiec.librechat.core.model.request.EphemeralAgent import com.garfiec.librechat.core.model.response.ChatStatusResponse import com.garfiec.librechat.core.network.api.ChatApi import com.garfiec.librechat.core.network.sse.SseClient -import io.ktor.client.HttpClient import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.emitAll import kotlinx.coroutines.flow.flow @@ -18,7 +17,6 @@ import kotlinx.coroutines.flow.flow class ChatRepositoryImpl( private val chatApi: ChatApi, private val sseClient: SseClient, - private val sseHttpClient: HttpClient, private val connectivityObserver: ConnectivityObserver, ) : ChatRepository { @@ -70,7 +68,7 @@ class ChatRepositoryImpl( // Phase 2: GET the SSE stream using the streamId val streamUrl = "api/agents/chat/stream/$streamId" - emitAll(sseClient.connect(sseHttpClient, streamUrl, connectivityFlow = connectivityObserver.isConnected)) + emitAll(sseClient.connect(streamUrl, connectivityFlow = connectivityObserver.isConnected)) } override suspend fun abortChat(streamId: String): Result = safeApiCall { @@ -83,6 +81,6 @@ class ChatRepositoryImpl( override fun resumeStream(conversationId: String): Flow = flow { val streamUrl = "api/agents/chat/stream/$conversationId" - emitAll(sseClient.connect(sseHttpClient, streamUrl, resume = true, connectivityFlow = connectivityObserver.isConnected)) + emitAll(sseClient.connect(streamUrl, resume = true, connectivityFlow = connectivityObserver.isConnected)) } } diff --git a/core/network/CLAUDE.md b/core/network/CLAUDE.md index 04ebfa9..f104452 100644 --- a/core/network/CLAUDE.md +++ b/core/network/CLAUDE.md @@ -14,17 +14,46 @@ Ktor HttpClient, API service classes, SSE streaming client, auth interceptor. Al ## SSE Streaming Architecture -**Do NOT use Ktor's SSE plugin.** Use the custom line parser over raw `ByteReadChannel`. +**Two layers of buffering have to be worked around. Don't undo either of them.** -Two-phase protocol: -1. `POST /api/agents/chat` with full message payload -> returns `{ streamId }` (streamId === conversationId) +### Layer 1: Ktor `client.get()` buffers the body (both platforms) + +Use `prepareGet { } + execute { response -> response.bodyAsChannel() }` to stream the response incrementally. `client.get()` materializes the full body in memory before returning, which defeats SSE. This was the original Android workaround introduced in commit `f182b2b`. The explanatory comment was deleted in commit `770603e` during the KMP iOS migration; it has since been re-added in `SseHttpTransport.android.kt`. + +### Layer 2: NSURLSession buffers `text/*` responses (iOS only) + +NSURLSession has an undocumented behavior where `URLSession:dataTask:didReceiveData:` is not called for `text/event-stream` responses until ~512 bytes are received OR the connection closes. `application/json` and `application/octet-stream` are exempt. LibreChat hardcodes `text/event-stream` on its SSE endpoint, so on iOS this would cause chat streaming to appear frozen until the user pressed stop. Tracked as **KTOR-6378** (status: Unresolved on Ktor's side; see also Apple Developer Forums thread 64875, open since 2016). + +This **cannot** be fixed in Ktor or `commonMain`. Ktor's Darwin engine faithfully forwards every `didReceiveData` callback it gets, but NSURLSession isn't calling it. The Layer 1 `prepareGet + execute` workaround does not reach Layer 2 — it addresses Ktor-level body materialization, not OS-level callback withholding. + +### Architecture: `expect class SseHttpTransport` + +Both layers are addressed via an `SseHttpTransport` abstraction: + +- **`SseHttpTransport.android.kt`** — thin Ktor `prepareGet + execute` shim. Layer 1 fix only; Layer 2 doesn't apply on Android. +- **`SseHttpTransport.ios.kt`** — custom HTTP/1.1 client built on `Network.framework`'s `NWConnection`. Bypasses NSURLSession entirely for the SSE GET only, which sidesteps Layer 2 at the OS level. All other iOS HTTP traffic still uses the Darwin engine. +- **`HttpResponseParser.kt`** — `commonMain` HTTP/1.1 status-line + headers + chunked-body parser. Pure Kotlin, JVM-testable, 18 unit tests. +- **`SseClient`** — wraps the transport with retry, connectivity-flow handling, mapper state reset, and SKIE-safe error handling. Platform-agnostic. + +### iOS cinterop bridge gotcha + +The iOS transport requires a `.def` cinterop bridge at `core/network/src/iosMain/cinterop/nwparams_defaults.def`. Kotlin/Native 2.3.20's `platform.Network.NW_PARAMETERS_DEFAULT_CONFIGURATION` binding wraps the block pointer in a Kotlin lambda (`knifunptr_getter`), which crashes on `block_destroy_helper` when Network.framework hands the cleanup path a Swift-generic `Network.ProtocolOptions` that doesn't descend from NSObject. The `.def` bridge calls the macro at C compile time so the block pointer stays entirely inside Network.framework's memory management. + +**Do NOT remove the `.def` bridge thinking the platform binding will work.** It won't. The 4-round Phase 2b debug cycle that introduced this fix is documented in commit `0ad21b6`. + +### Two-phase protocol + +Two-phase SSE protocol: +1. `POST /api/agents/chat` with the message payload → returns `{ streamId }` (where `streamId === conversationId`) 2. `GET /api/agents/chat/stream/:streamId` opens the SSE event stream -Legacy single-phase path (OpenAI Assistants): POST body is the SSE stream itself. Both paths need the custom parser. +Legacy single-phase path (OpenAI Assistants): POST body is the SSE stream itself. Both paths use the custom transport via `SseClient`. -`SseConnectionManager` handles lifecycle: start, reconnect with exponential backoff (1s/2s/4s/8s, max 5 retries), abort via `POST /api/agents/chat/abort`, and exposes `StateFlow`. +`SseConnectionManager` handles lifecycle: start, reconnect with exponential backoff (1s/2s/4s/8s, max 5 retries), abort via `POST /api/agents/chat/abort`, exposes `StateFlow`. On reconnection, append `?resume=true` to get a `sync` event with `runSteps[]` + `aggregatedContent[]`. -On reconnection, append `?resume=true` to get a `sync` event with `runSteps[]` + `aggregatedContent[]`. +### Don't use Ktor's SSE plugin + +The Ktor SSE plugin uses the same `NSURLSessionDataTask` code path as the regular Darwin engine, so it would have the same Layer 2 bug on iOS. The custom transport is mandatory. ## Key Configuration diff --git a/core/network/build.gradle.kts b/core/network/build.gradle.kts index e9b7556..bf64f52 100644 --- a/core/network/build.gradle.kts +++ b/core/network/build.gradle.kts @@ -9,6 +9,14 @@ android { } kotlin { + listOf(iosArm64(), iosSimulatorArm64()).forEach { target -> + target.compilations.getByName("main").cinterops { + val nwparams_defaults by creating { + defFile(project.file("src/iosMain/cinterop/nwparams_defaults.def")) + } + } + } + sourceSets { commonMain.dependencies { implementation(project(":core:model")) diff --git a/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.android.kt b/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.android.kt index 511f2d2..709af20 100644 --- a/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.android.kt +++ b/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.android.kt @@ -1,5 +1,7 @@ package com.garfiec.librechat.core.network.di +import com.garfiec.librechat.core.common.di.KoinQualifiers +import com.garfiec.librechat.core.network.sse.SseHttpTransport import io.ktor.client.engine.HttpClientEngineFactory import io.ktor.client.engine.okhttp.OkHttp import org.koin.core.module.Module @@ -7,4 +9,5 @@ import org.koin.dsl.module actual val networkPlatformModule: Module = module { single> { OkHttp } + single { SseHttpTransport(get(KoinQualifiers.Streaming)) } } diff --git a/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.android.kt b/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.android.kt new file mode 100644 index 0000000..a23df10 --- /dev/null +++ b/core/network/src/androidMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.android.kt @@ -0,0 +1,56 @@ +package com.garfiec.librechat.core.network.sse + +import io.ktor.client.HttpClient +import io.ktor.client.request.accept +import io.ktor.client.request.parameter +import io.ktor.client.request.prepareGet +import io.ktor.client.statement.bodyAsChannel +import io.ktor.http.ContentType +import io.ktor.http.isSuccess +import io.ktor.http.path +import io.ktor.utils.io.readAvailable +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow + +// Use prepareGet + execute to stream the response incrementally. +// client.get() buffers the entire body in memory before returning, +// which defeats SSE streaming. prepareGet().execute {} keeps the +// HTTP connection open so we can read line-by-line as data arrives. +// +// This was the original Android workaround introduced in commit f182b2b. +// It addresses Ktor-layer buffering only — sufficient for OkHttp on +// Android but NOT for Darwin/NSURLSession on iOS, which has a separate +// text/* response-buffering quirk at the OS network stack level. See +// SseHttpTransport.ios.kt for why iOS needs a totally different transport +// and what NSURLSession does wrong. +actual class SseHttpTransport( + private val client: HttpClient, +) { + actual fun stream(streamPath: String, resume: Boolean): Flow = flow { + client.prepareGet { + url { path(streamPath) } + accept(ContentType.Text.EventStream) + if (resume) { + parameter("resume", "true") + } + }.execute { response -> + if (!response.status.isSuccess()) { + throw SseHttpStatusException(response.status.value) + } + val channel = response.bodyAsChannel() + val buffer = ByteArray(BUFFER_SIZE) + while (!channel.isClosedForRead) { + val read = channel.readAvailable(buffer, 0, buffer.size) + if (read > 0) { + emit(buffer.copyOf(read)) + } else if (read < 0) { + break + } + } + } + } + + private companion object { + const val BUFFER_SIZE = 8192 + } +} diff --git a/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/di/NetworkModule.kt b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/di/NetworkModule.kt index 2fac665..2f35f81 100644 --- a/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/di/NetworkModule.kt +++ b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/di/NetworkModule.kt @@ -109,7 +109,9 @@ val networkModule = module { } } - // SSE + // SSE — SseHttpTransport is provided by networkPlatformModule because its + // constructor signature differs between Android (takes the Ktor streaming + // HttpClient) and iOS (takes NWConnection dependencies in Phase 2). singleOf(::SseClient) // API services diff --git a/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParser.kt b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParser.kt new file mode 100644 index 0000000..bd4bcd8 --- /dev/null +++ b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParser.kt @@ -0,0 +1,334 @@ +package com.garfiec.librechat.core.network.sse + +/** + * Minimal, stateful HTTP/1.1 response parser for the iOS raw-socket SSE transport. + * + * Lives in commonMain so it can be JVM-unit-tested without platform deps. Phase 2b's + * iOS NWConnection transport feeds raw TCP bytes into this parser; Android does NOT + * use it (OkHttp handles framing itself). + * + * Wire format we target (confirmed in the Phase 0 spike): + * HTTP/1.1 200 OK\r\n + * Content-Type: text/event-stream\r\n + * Transfer-Encoding: chunked\r\n + * Content-Encoding: identity\r\n + * Connection: close\r\n + * Cache-Control: no-cache, no-transform\r\n + * \r\n + * [; ext...]\r\n + * \r\n + * ... repeated ... + * 0\r\n + * \r\n + * + * Handled: + * - Status line parsing (`HTTP/1.1 `) + * - Header folding into a lower-cased map (case-insensitive lookup) + * - Chunked transfer-encoding with hex size lines and optional chunk extensions + * (everything after the first `;` on the size line is ignored) + * - Identity body (pass-through) used only as a fallback for non-chunked responses + * - Arbitrary feed boundaries: a single logical line may arrive across many feeds + * - Final `0\r\n\r\n` terminator emits EndOfStream + * + * NOT handled — by design, because LibreChat+Cloudflare never send these to us: + * - Chunk trailers (the bytes between the final `0\r\n` and the terminating `\r\n`) + * - Content-Encoding other than identity (we send `Accept-Encoding: identity`) + * - HTTP/1.0 "connection close = EOF" without Content-Length / Transfer-Encoding + * - Redirects (3xx) — caller must handle status code if it ever sees one + * - Pipelined responses — one request, one response, done + * + * Usage: + * val parser = HttpResponseParser() + * socket.read { bytes -> + * parser.feed(bytes).forEach { event -> + * when (event) { ... } + * } + * } + */ +class HttpResponseParser { + + sealed class ParseEvent { + data class HeadersComplete( + val statusCode: Int, + val headers: Map, + ) : ParseEvent() + + data class BodyChunk(val bytes: ByteArray) : ParseEvent() { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is BodyChunk) return false + return bytes.contentEquals(other.bytes) + } + + override fun hashCode(): Int = bytes.contentHashCode() + } + + data object EndOfStream : ParseEvent() + + data class Error(val message: String) : ParseEvent() + } + + private enum class State { + READING_STATUS_LINE, + READING_HEADERS, + READING_BODY, + DONE, + FAILED, + } + + private enum class BodyMode { UNKNOWN, IDENTITY, CHUNKED } + + private enum class ChunkState { + READING_SIZE_LINE, + READING_DATA, + READING_TRAILING_CRLF, + READING_FINAL_CRLF, + } + + private var state: State = State.READING_STATUS_LINE + private var bodyMode: BodyMode = BodyMode.UNKNOWN + private var chunkState: ChunkState = ChunkState.READING_SIZE_LINE + private var remainingChunkBytes: Int = 0 + + // Raw byte accumulator. We append on feed() and drain-in-place as we parse. + // Simple ArrayList keeps the code platform-agnostic; byte counts per + // SSE frame are small (a few KB) so the copy-cost is negligible. + private val buffer: ArrayList = ArrayList(DEFAULT_BUFFER_CAPACITY) + + private val headers: MutableMap = mutableMapOf() + private var statusCode: Int = 0 + + fun feed(bytes: ByteArray): List { + if (state == State.DONE || state == State.FAILED) return emptyList() + if (bytes.isNotEmpty()) { + buffer.ensureCapacity(buffer.size + bytes.size) + for (b in bytes) buffer.add(b) + } + val events = mutableListOf() + drain(events) + return events + } + + private fun drain(out: MutableList) { + while (true) { + when (state) { + State.READING_STATUS_LINE -> { + val line = readLine() ?: return + if (!parseStatusLine(line, out)) return + state = State.READING_HEADERS + } + + State.READING_HEADERS -> { + val line = readLine() ?: return + if (line.isEmpty()) { + // end of headers + finalizeHeaders(out) + if (state == State.FAILED || state == State.DONE) return + } else { + if (!parseHeaderLine(line, out)) return + } + } + + State.READING_BODY -> { + val progressed = when (bodyMode) { + BodyMode.CHUNKED -> drainChunkedBody(out) + BodyMode.IDENTITY -> drainIdentityBody(out) + BodyMode.UNKNOWN -> { + fail(out, "body mode not determined at READING_BODY") + return + } + } + if (!progressed) return + } + + State.DONE, State.FAILED -> return + } + } + } + + private fun parseStatusLine(line: String, out: MutableList): Boolean { + // Expected shape: "HTTP/1.1 200 OK" or "HTTP/1.0 200 Ok" + // Tolerate any single-space separators; reject anything else. + val parts = line.split(' ', limit = 3) + if (parts.size < 2) { + fail(out, "malformed status line: '$line'") + return false + } + val version = parts[0] + if (!version.startsWith("HTTP/1.")) { + fail(out, "unsupported HTTP version: '$version'") + return false + } + val code = parts[1].toIntOrNull() + if (code == null || code < 100 || code > 599) { + fail(out, "invalid status code: '${parts[1]}'") + return false + } + statusCode = code + return true + } + + private fun parseHeaderLine(line: String, out: MutableList): Boolean { + val colon = line.indexOf(':') + if (colon <= 0) { + fail(out, "malformed header line: '$line'") + return false + } + val name = line.substring(0, colon).trim().lowercase() + val value = line.substring(colon + 1).trim() + // Duplicate headers: last-wins is fine for what we use (Content-Length, + // Transfer-Encoding, Content-Type). No need to handle comma-folded duplicates. + headers[name] = value + return true + } + + private fun finalizeHeaders(out: MutableList) { + val transferEncoding = headers["transfer-encoding"]?.lowercase() + bodyMode = if (transferEncoding != null && transferEncoding.contains("chunked")) { + BodyMode.CHUNKED + } else { + BodyMode.IDENTITY + } + out.add(ParseEvent.HeadersComplete(statusCode, headers.toMap())) + state = State.READING_BODY + chunkState = ChunkState.READING_SIZE_LINE + } + + /** + * Returns true if parsing made forward progress (consumed at least one byte or + * emitted at least one event). Returns false if the body is starved — caller's + * drain() loop treats false as "wait for more bytes". + */ + private fun drainChunkedBody(out: MutableList): Boolean { + when (chunkState) { + ChunkState.READING_SIZE_LINE -> { + val line = readLine() ?: return false + val semi = line.indexOf(';') + val sizeToken = if (semi >= 0) line.substring(0, semi).trim() else line.trim() + val size = parseHexSize(sizeToken) + if (size == null) { + fail(out, "invalid chunk size: '$sizeToken'") + return false + } + remainingChunkBytes = size + chunkState = if (size == 0) ChunkState.READING_FINAL_CRLF else ChunkState.READING_DATA + return true + } + + ChunkState.READING_DATA -> { + if (buffer.isEmpty()) return false + val take = minOf(remainingChunkBytes, buffer.size) + val slice = ByteArray(take) + for (i in 0 until take) slice[i] = buffer[i] + removeFromBufferHead(take) + remainingChunkBytes -= take + if (slice.isNotEmpty()) out.add(ParseEvent.BodyChunk(slice)) + if (remainingChunkBytes == 0) { + chunkState = ChunkState.READING_TRAILING_CRLF + } + return true + } + + ChunkState.READING_TRAILING_CRLF -> { + // After each non-terminal chunk's data, the protocol requires a CRLF. + val line = readLine() ?: return false + if (line.isNotEmpty()) { + fail(out, "expected CRLF after chunk data, got: '$line'") + return false + } + chunkState = ChunkState.READING_SIZE_LINE + return true + } + + ChunkState.READING_FINAL_CRLF -> { + // After the `0\r\n` terminator line, we expect one more empty line + // that closes the message (no trailers supported). Then EndOfStream. + val line = readLine() ?: return false + if (line.isNotEmpty()) { + fail(out, "expected CRLF after final chunk, got: '$line'") + return false + } + out.add(ParseEvent.EndOfStream) + state = State.DONE + return true + } + } + } + + private fun drainIdentityBody(out: MutableList): Boolean { + if (buffer.isEmpty()) return false + val slice = ByteArray(buffer.size) + for (i in buffer.indices) slice[i] = buffer[i] + buffer.clear() + out.add(ParseEvent.BodyChunk(slice)) + return true + } + + /** + * Read a single CRLF-terminated line from the buffer. Consumes the CRLF. + * Returns null if no complete line is available yet (caller should wait for + * more bytes). Tolerates bare `\n` as a line terminator because we don't + * control every intermediary. + */ + private fun readLine(): String? { + var i = 0 + while (i < buffer.size) { + val b = buffer[i].toInt() and 0xFF + if (b == LF) { + // Line ends here. If the previous byte is CR, trim it. + val endExclusive = if (i > 0 && (buffer[i - 1].toInt() and 0xFF) == CR) i - 1 else i + val bytes = ByteArray(endExclusive) + for (j in 0 until endExclusive) bytes[j] = buffer[j] + removeFromBufferHead(i + 1) + return bytes.decodeToString() + } + i++ + } + return null + } + + private fun removeFromBufferHead(count: Int) { + if (count <= 0) return + if (count >= buffer.size) { + buffer.clear() + return + } + val remaining = buffer.size - count + for (i in 0 until remaining) { + buffer[i] = buffer[i + count] + } + // Truncate the tail. ArrayList has no public truncate but removeAt from the + // end in a loop is O(n) with cheap ops; for our sizes this is fine. + while (buffer.size > remaining) { + buffer.removeAt(buffer.size - 1) + } + } + + private fun parseHexSize(token: String): Int? { + if (token.isEmpty() || token.length > HEX_SIZE_MAX_DIGITS) return null + var result = 0 + for (ch in token) { + val digit = when (ch) { + in '0'..'9' -> ch - '0' + in 'a'..'f' -> ch - 'a' + 10 + in 'A'..'F' -> ch - 'A' + 10 + else -> return null + } + result = (result shl 4) or digit + if (result < 0) return null // overflow guard for pathologically huge chunks + } + return result + } + + private fun fail(out: MutableList, message: String) { + state = State.FAILED + out.add(ParseEvent.Error(message)) + } + + private companion object { + const val DEFAULT_BUFFER_CAPACITY = 8192 + const val CR = 0x0D + const val LF = 0x0A + const val HEX_SIZE_MAX_DIGITS = 8 // 0xFFFFFFFF — plenty for our chunks + } +} diff --git a/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseClient.kt b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseClient.kt index 720a8aa..10c7211 100644 --- a/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseClient.kt +++ b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseClient.kt @@ -2,25 +2,22 @@ package com.garfiec.librechat.core.network.sse import co.touchlab.kermit.Logger import com.garfiec.librechat.core.model.StreamEvent -import io.ktor.client.HttpClient -import io.ktor.client.request.accept -import io.ktor.client.request.parameter -import io.ktor.client.request.prepareGet -import io.ktor.client.statement.bodyAsChannel -import io.ktor.http.ContentType import io.ktor.http.HttpStatusCode -import io.ktor.http.isSuccess -import io.ktor.http.path +import io.ktor.utils.io.ByteChannel +import io.ktor.utils.io.writeFully import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.launch import kotlinx.serialization.json.Json import kotlin.math.min class SseClient( private val json: Json, + private val transport: SseHttpTransport, ) { private val mapper = SseEventMapper(json) private val lineParser = SseLineParser() @@ -31,7 +28,6 @@ class SseClient( * avoiding wasted retry attempts while offline. */ fun connect( - client: HttpClient, streamPath: String, resume: Boolean = false, connectivityFlow: Flow? = null, @@ -50,45 +46,53 @@ class SseClient( while (attempt <= maxRetries && !done) { try { - val statement = client.prepareGet { - url { path(streamPath) } - accept(ContentType.Text.EventStream) - if (shouldResume) { - parameter("resume", "true") + val byteChannel = ByteChannel(autoFlush = true) + coroutineScope { + val pumpJob = launch { + try { + transport.stream(streamPath, shouldResume).collect { bytes -> + attempt = 0 + byteChannel.writeFully(bytes) + } + byteChannel.flushAndClose() + } catch (e: CancellationException) { + byteChannel.cancel(e) + throw e + } catch (e: Exception) { + byteChannel.cancel(e) + throw e + } } - } - - statement.execute { response -> - when { - response.status.isSuccess() -> { - attempt = 0 - val channel = response.bodyAsChannel() - lineParser.parse(channel).collect { sseEvent -> - val streamEvent = mapper.map(sseEvent) - if (streamEvent != null) { - emit(streamEvent) - if (streamEvent is StreamEvent.Final) { - done = true - } + try { + lineParser.parse(byteChannel).collect { sseEvent -> + val streamEvent = mapper.map(sseEvent) + if (streamEvent != null) { + emit(streamEvent) + if (streamEvent is StreamEvent.Final) { + done = true } } - done = true } + } finally { + pumpJob.cancel() + } + } + done = true + } catch (e: SseHttpStatusException) { + when (e.statusCode) { + HttpStatusCode.NotFound.value -> { + done = true + } - response.status == HttpStatusCode.NotFound -> { - done = true - } + HttpStatusCode.Unauthorized.value -> { + Logger.w("SSE") { "SSE: 401 Unauthorized for $streamPath" } + emit(StreamEvent.Error(message = "Unauthorized", code = "401")) + done = true + } - response.status == HttpStatusCode.Unauthorized -> { - Logger.w("SSE") { "SSE: 401 Unauthorized for $streamPath" } - emit(StreamEvent.Error(message = "Unauthorized", code = "401")) - done = true - } - - else -> { - Logger.w("SSE") { "SSE: unexpected status ${response.status} for $streamPath" } - attempt++ - } + else -> { + Logger.w("SSE") { "SSE: unexpected status ${e.statusCode} for $streamPath" } + attempt++ } } } catch (e: SseStreamException) { @@ -98,6 +102,8 @@ class SseClient( emit(StreamEvent.Error(message = "Connection lost. Please check your network and try again.", isNetworkError = true)) done = true } + } catch (e: CancellationException) { + throw e } catch (e: Exception) { Logger.w("SSE", e) { "SSE connection error (attempt $attempt)" } attempt++ diff --git a/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.kt b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.kt new file mode 100644 index 0000000..cc92cec --- /dev/null +++ b/core/network/src/commonMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.kt @@ -0,0 +1,41 @@ +package com.garfiec.librechat.core.network.sse + +import kotlinx.coroutines.flow.Flow + +/** + * Platform abstraction for the SSE GET transport. + * + * Exists solely to let iOS bypass NSURLSession's text-star response buffering + * (`text/event-stream`, `text/plain`, etc.). NSURLSession does not call + * `didReceiveData` for those Content-Types until ~512 bytes have been received + * OR the connection closes — see KTOR-6378 "Darwin: The engine doesn't stream + * chunked responses with small chunks" (status: Unresolved) and Apple Developer + * Forums thread 64875 (open since 2016). + * + * Android and iOS implementations look totally different by design: + * - Android wraps the existing Ktor `prepareGet + execute` path around the + * shared `KoinQualifiers.Streaming` HttpClient. Behavior is unchanged. + * - iOS uses a raw `NWConnection` HTTP/1.1 client that avoids NSURLSession + * entirely for the SSE GET only. All other iOS HTTP traffic continues to + * use the Darwin engine, which works fine for non-text-star responses. + * + * Implementations emit byte chunks exactly as they arrive on the wire; all + * retry, connectivity, and SSE line parsing logic stays in [SseClient]. + * + * Non-success HTTP responses (anything other than 2xx) are surfaced as a + * thrown [SseHttpStatusException] so [SseClient] can differentiate 401 / + * 404 / other the same way it did when it held the raw Ktor response. + */ +expect class SseHttpTransport { + fun stream(streamPath: String, resume: Boolean): Flow +} + +/** + * Thrown by an [SseHttpTransport] when the server responds with a non-success + * status code. Preserves the numeric status so [SseClient] can branch on 401 / + * 404 / other exactly as it did before the transport abstraction was extracted. + */ +class SseHttpStatusException( + val statusCode: Int, + message: String = "SSE transport received HTTP $statusCode", +) : Exception(message) diff --git a/core/network/src/commonTest/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParserTest.kt b/core/network/src/commonTest/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParserTest.kt new file mode 100644 index 0000000..3e1551f --- /dev/null +++ b/core/network/src/commonTest/kotlin/com/garfiec/librechat/core/network/sse/HttpResponseParserTest.kt @@ -0,0 +1,275 @@ +package com.garfiec.librechat.core.network.sse + +import com.garfiec.librechat.core.network.sse.HttpResponseParser.ParseEvent +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertIs +import kotlin.test.assertTrue + +class HttpResponseParserTest { + + private fun String.bytes(): ByteArray = encodeToByteArray() + + private fun collectBodyBytes(events: List): ByteArray { + val out = mutableListOf() + for (e in events) { + if (e is ParseEvent.BodyChunk) { + for (b in e.bytes) out.add(b) + } + } + return ByteArray(out.size) { out[it] } + } + + @Test + fun statusLineParsesOk() { + val parser = HttpResponseParser() + val events = parser.feed("HTTP/1.1 200 OK\r\n\r\n".bytes()) + // No body-mode determined -> falls through identity on empty feed, so the only + // event we should see is HeadersComplete. + val headers = events.filterIsInstance().single() + assertEquals(200, headers.statusCode) + assertTrue(headers.headers.isEmpty()) + } + + @Test + fun statusLineWithHttp10IsAccepted() { + val parser = HttpResponseParser() + val events = parser.feed("HTTP/1.0 404 Not Found\r\n\r\n".bytes()) + val headers = events.filterIsInstance().single() + assertEquals(404, headers.statusCode) + } + + @Test + fun malformedStatusLineReturnsError() { + val parser = HttpResponseParser() + val events = parser.feed("NOT-HTTP 200 OK\r\n".bytes()) + val err = events.filterIsInstance().single() + assertTrue(err.message.contains("HTTP")) + } + + @Test + fun invalidStatusCodeReturnsError() { + val parser = HttpResponseParser() + val events = parser.feed("HTTP/1.1 abc OK\r\n".bytes()) + assertTrue(events.any { it is ParseEvent.Error }) + } + + @Test + fun headersParseAcrossMultipleFeeds() { + val parser = HttpResponseParser() + val e1 = parser.feed("HTTP/1.1 200 OK\r\nContent-Ty".bytes()) + assertTrue(e1.isEmpty()) + val e2 = parser.feed("pe: text/event-stream\r\nTrans".bytes()) + assertTrue(e2.isEmpty()) + val e3 = parser.feed("fer-Encoding: chunked\r\n\r\n".bytes()) + val hc = e3.filterIsInstance().single() + assertEquals(200, hc.statusCode) + assertEquals("text/event-stream", hc.headers["content-type"]) + assertEquals("chunked", hc.headers["transfer-encoding"]) + } + + @Test + fun headerLookupIsCaseInsensitive() { + val parser = HttpResponseParser() + val events = parser.feed( + "HTTP/1.1 200 OK\r\nX-Custom-Header: value\r\n\r\n".bytes(), + ) + val hc = events.filterIsInstance().single() + // Map stores lower-cased keys; callers can look up either way by lowercasing. + assertEquals("value", hc.headers["x-custom-header"]) + } + + @Test + fun chunkedSingleSmallChunk() { + val parser = HttpResponseParser() + val wire = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + + "5\r\nhello\r\n" + + "0\r\n\r\n" + val events = parser.feed(wire.bytes()) + assertIs(events[0]) + val body = collectBodyBytes(events).decodeToString() + assertEquals("hello", body) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun chunkedMultipleChunksSplitAcrossFeeds() { + val parser = HttpResponseParser() + val events = mutableListOf() + events += parser.feed("HTTP/1.1 200 OK\r\n".bytes()) + events += parser.feed("Transfer-Encoding: chunked\r\n\r\n".bytes()) + events += parser.feed("5\r\nhello\r\n".bytes()) + events += parser.feed("6\r\n world\r\n".bytes()) + events += parser.feed("1\r\n!\r\n".bytes()) + events += parser.feed("0\r\n\r\n".bytes()) + + val hc = events.filterIsInstance().single() + assertEquals(200, hc.statusCode) + assertEquals("hello world!", collectBodyBytes(events).decodeToString()) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun chunkedSizeLineWithExtensionIsAccepted() { + val parser = HttpResponseParser() + val wire = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + + "5;foo=bar\r\nhello\r\n" + + "0\r\n\r\n" + val events = parser.feed(wire.bytes()) + assertEquals("hello", collectBodyBytes(events).decodeToString()) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun chunkedFinalZeroChunkEmitsEndOfStream() { + val parser = HttpResponseParser() + val wire = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + + "0\r\n\r\n" + val events = parser.feed(wire.bytes()) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + // No body chunks should have been emitted. + assertTrue(events.none { it is ParseEvent.BodyChunk }) + // Feeding more bytes after DONE must be a no-op. + val after = parser.feed("anything".bytes()) + assertTrue(after.isEmpty()) + } + + @Test + fun identityBodyPassesThroughUntilCallerStops() { + val parser = HttpResponseParser() + val events = mutableListOf() + events += parser.feed("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n".bytes()) + events += parser.feed("{\"a\":1}".bytes()) + events += parser.feed(",{\"b\":2}".bytes()) + + val hc = events.filterIsInstance().single() + assertEquals(200, hc.statusCode) + assertEquals("{\"a\":1},{\"b\":2}", collectBodyBytes(events).decodeToString()) + } + + @Test + fun bodyBytesSplitMidChunkSizeLine() { + val parser = HttpResponseParser() + val events = mutableListOf() + events += parser.feed("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n".bytes()) + // Split "5\r\n" across two feeds: feed "5\r", then "\nhello\r\n" + events += parser.feed("5\r".bytes()) + events += parser.feed("\nhello\r\n0\r\n\r\n".bytes()) + + assertEquals("hello", collectBodyBytes(events).decodeToString()) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun bodyBytesSplitMidChunkData() { + val parser = HttpResponseParser() + val events = mutableListOf() + events += parser.feed("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n".bytes()) + // Split "hello" mid-data + events += parser.feed("5\r\nhel".bytes()) + events += parser.feed("lo\r\n0\r\n\r\n".bytes()) + + assertEquals("hello", collectBodyBytes(events).decodeToString()) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun statusLineSplitMidLine() { + val parser = HttpResponseParser() + val e1 = parser.feed("HTTP/1.1 20".bytes()) + assertTrue(e1.isEmpty()) + val e2 = parser.feed("0 OK\r\n\r\n".bytes()) + val hc = e2.filterIsInstance().single() + assertEquals(200, hc.statusCode) + } + + @Test + fun realLibreChatSseChunkIsPassedThroughVerbatim() { + // Fixture crafted from the Phase 0 NWConnection spike — an on_run_step event + // arriving as a single 0x167-byte chunk. The parser must NOT strip the SSE + // framing (event:/data:/\n\n); that's SseLineParser's job downstream. + val parser = HttpResponseParser() + + val sseBody = "event: message\n" + + "data: {\"event\":\"on_run_step\",\"data\":{\"type\":\"tool_call\"}}\n" + + "\n" + val hex = sseBody.length.toString(16) + + val wire = buildString { + append("HTTP/1.1 200 OK\r\n") + append("Content-Type: text/event-stream\r\n") + append("Transfer-Encoding: chunked\r\n") + append("Content-Encoding: identity\r\n") + append("Connection: close\r\n") + append("Cache-Control: no-cache, no-transform\r\n") + append("\r\n") + append(hex) + append("\r\n") + append(sseBody) + append("\r\n") + append("0\r\n\r\n") + } + + val events = parser.feed(wire.bytes()) + val hc = events.filterIsInstance().single() + assertEquals(200, hc.statusCode) + assertEquals("text/event-stream", hc.headers["content-type"]) + assertEquals("chunked", hc.headers["transfer-encoding"]) + + val decodedBody = collectBodyBytes(events).decodeToString() + assertEquals(sseBody, decodedBody) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun realLibreChatSseChunkByteByByteFeedStillParses() { + // Same fixture, but fed one byte at a time. Exercises the worst-case + // re-entry pattern for the state machine. + val parser = HttpResponseParser() + + val sseBody = "event: message\ndata: {\"x\":1}\n\n" + val hex = sseBody.length.toString(16) + val wire = buildString { + append("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n") + append(hex) + append("\r\n") + append(sseBody) + append("\r\n0\r\n\r\n") + }.bytes() + + val events = mutableListOf() + for (b in wire) { + events += parser.feed(byteArrayOf(b)) + } + + assertEquals(200, events.filterIsInstance().single().statusCode) + assertEquals(sseBody, collectBodyBytes(events).decodeToString()) + assertTrue(events.any { it is ParseEvent.EndOfStream }) + } + + @Test + fun invalidChunkSizeReturnsError() { + val parser = HttpResponseParser() + val wire = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + + "ZZZ\r\n" + val events = parser.feed(wire.bytes()) + assertTrue(events.any { it is ParseEvent.Error }) + } + + @Test + fun multipleSmallChunksProduceMultipleBodyChunkEvents() { + // This is important for streaming: one chunked frame in, one BodyChunk out, + // so the downstream SseLineParser sees frames as they arrive. + val parser = HttpResponseParser() + val events = mutableListOf() + events += parser.feed("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n".bytes()) + events += parser.feed("3\r\nfoo\r\n".bytes()) + events += parser.feed("3\r\nbar\r\n".bytes()) + events += parser.feed("0\r\n\r\n".bytes()) + + val bodyChunks = events.filterIsInstance() + assertEquals(2, bodyChunks.size) + assertEquals("foo", bodyChunks[0].bytes.decodeToString()) + assertEquals("bar", bodyChunks[1].bytes.decodeToString()) + } +} diff --git a/core/network/src/iosMain/cinterop/nwparams_defaults.def b/core/network/src/iosMain/cinterop/nwparams_defaults.def new file mode 100644 index 0000000..b3b3bf6 --- /dev/null +++ b/core/network/src/iosMain/cinterop/nwparams_defaults.def @@ -0,0 +1,23 @@ +package = com.garfiec.librechat.core.network.sse.nwparams +language = Objective-C +headers = Network/Network.h +compilerOpts = -framework Network +linkerOpts = -framework Network + +--- + +// C-level bridge for Apple's NW_PARAMETERS_DEFAULT_CONFIGURATION block pointer +// sentinel. Kotlin/Native's platform.Network cinterop binding for that macro +// exposes it as a Kotlin-wrapped lambda (via a knifunptr getter), which forces +// the configure-block argument through a lambda->Obj-C-block bridge that +// crashes at block_destroy_helper on modern iOS because Network.framework +// passes a Swift-generic Network.ProtocolOptions that does not +// inherit from NSObject. By calling the macro inside a plain C function here, +// the block pointer is resolved and handed to nw_parameters_create_secure_tcp +// entirely at C compile/link time -- Kotlin never sees the block. +static inline nw_parameters_t librechat_make_default_tls_tcp_parameters(void) { + return nw_parameters_create_secure_tcp( + NW_PARAMETERS_DEFAULT_CONFIGURATION, + NW_PARAMETERS_DEFAULT_CONFIGURATION + ); +} diff --git a/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.ios.kt b/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.ios.kt index 4480b10..3b0035a 100644 --- a/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.ios.kt +++ b/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/di/NetworkPlatformModule.ios.kt @@ -5,6 +5,9 @@ import io.ktor.client.engine.darwin.Darwin import org.koin.core.module.Module import org.koin.dsl.module +// NOTE: the iOS `SseHttpTransport` is registered in `IosSharedModule` rather +// than here, because the iOS KMP framework wires its own Koin graph that does +// not `include(networkModule)` — see `shared/src/iosMain/.../IosSharedModule.kt`. actual val networkPlatformModule: Module = module { single> { Darwin } } diff --git a/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.ios.kt b/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.ios.kt new file mode 100644 index 0000000..ec06864 --- /dev/null +++ b/core/network/src/iosMain/kotlin/com/garfiec/librechat/core/network/sse/SseHttpTransport.ios.kt @@ -0,0 +1,450 @@ +package com.garfiec.librechat.core.network.sse + +import co.touchlab.kermit.Logger +import com.garfiec.librechat.core.network.client.LibreChatHttpClient +import com.garfiec.librechat.core.network.client.ServerUrlProvider +import com.garfiec.librechat.core.network.client.TokenManager +import kotlinx.cinterop.BetaInteropApi +import kotlinx.cinterop.ExperimentalForeignApi +import kotlinx.cinterop.addressOf +import kotlinx.cinterop.alloc +import kotlinx.cinterop.convert +import kotlinx.cinterop.memScoped +import kotlinx.cinterop.ptr +import kotlinx.cinterop.readBytes +import kotlinx.cinterop.usePinned +import kotlinx.cinterop.value +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow +import kotlinx.coroutines.flow.emitAll +import kotlinx.coroutines.flow.flow +import platform.Foundation.NSURL +import platform.Network.nw_connection_cancel +import platform.Network.nw_connection_create +import platform.Network.nw_connection_force_cancel +import platform.Network.nw_connection_receive +import platform.Network.nw_connection_send +import platform.Network.nw_connection_set_queue +import platform.Network.nw_connection_set_state_changed_handler +import platform.Network.nw_connection_start +import platform.Network.nw_connection_state_cancelled +import platform.Network.nw_connection_state_failed +import platform.Network.nw_connection_state_ready +import platform.Network.nw_connection_state_t +import platform.Network.nw_connection_t +import platform.Network.nw_content_context_t +import platform.Network.nw_endpoint_create_host +import platform.Network.nw_error_get_error_code +import platform.Network.nw_error_get_error_domain +import platform.Network.nw_error_domain_posix +import platform.Network.nw_error_domain_tls +import platform.Network.nw_error_t +import platform.Network.nw_parameters_t +import com.garfiec.librechat.core.network.sse.nwparams.librechat_make_default_tls_tcp_parameters +import platform.darwin.dispatch_data_create +import platform.darwin.dispatch_data_create_map +import platform.darwin.dispatch_data_get_size +import platform.darwin.dispatch_data_t +import platform.darwin.dispatch_get_global_queue +import platform.darwin.dispatch_queue_create +import platform.posix.ECONNRESET +import platform.posix.EPIPE +import platform.posix.ETIMEDOUT +import platform.posix.size_tVar + +// iOS SSE transport that bypasses NSURLSession entirely. +// +// Why this exists: +// NSURLSession has an undocumented buffering behavior where +// `URLSession:dataTask:didReceiveData:` is not called for `text/*` +// Content-Type responses until 512+ bytes have been received OR the +// connection closes. `application/json` and `application/octet-stream` +// are exempt — they stream byte-by-byte. LibreChat's SSE endpoint sends +// `Content-Type: text/event-stream`, so on iOS the chat response would +// never appear until the user pressed stop (which closes the connection +// and flushes NSURLSession's buffer). +// +// This is acknowledged but unfixed Apple behavior — see Apple Developer +// Forums thread 64875 (open since 2016, last confirmed Feb 2025). +// Tracked on the Ktor side as KTOR-6378 "Darwin: The engine doesn't +// stream chunked responses with small chunks", status: Unresolved. +// +// The fix CANNOT live in Ktor or in commonMain because the buffering +// happens below Ktor — Ktor's Darwin engine faithfully forwards every +// didReceiveData callback it gets, but NSURLSession isn't calling it. +// The Android prepareGet+execute workaround (commit f182b2b) only +// addresses Ktor-layer buffering, which is a different layer. +// +// This implementation uses Network.framework's NWConnection to open a +// raw TCP+TLS connection and speak HTTP/1.1 directly, avoiding +// NSURLSession entirely for the SSE endpoint only. All other HTTP +// traffic (JSON requests, auth, conversation list, etc.) continues to +// use the Darwin engine, which works fine for non-text/* responses. +// +// Trade-offs vs NSURLSession: +// - No App Transport Security enforcement (Network.framework doesn't +// gate on ATS the way URLSession does). Acceptable for a "bring your +// own server" app where users may run LibreChat on local IPs. +// - No automatic system proxy detection. If we ever need it, we'll +// read the system proxy config manually. +// - Cert pinning, when added later, must be wired up here separately +// from the NSURLSessionDelegate-based approach used by the Darwin +// engine. +@OptIn(ExperimentalForeignApi::class, BetaInteropApi::class) +actual class SseHttpTransport( + private val serverUrlProvider: ServerUrlProvider, + private val tokenManager: TokenManager, +) { + + actual fun stream(streamPath: String, resume: Boolean): Flow = flow { + // 401-retry wrapper: resolve token in the suspend scope (we can't touch + // TokenManager from inside the NWConnection callback), open the connection, + // let it run to completion. If it throws SseHttpStatusException(401) and + // we haven't already retried, refresh the token and loop once. This + // replicates AuthInterceptorPlugin's 401 handling for the raw-socket path. + var triedRefresh = false + while (true) { + val token = tokenManager.getAccessToken() + try { + emitAll(openConnection(streamPath, resume, token)) + return@flow + } catch (e: SseHttpStatusException) { + if (e.statusCode == 401 && !triedRefresh) { + Logger.w("SSE-iOS") { "401 on SSE stream, attempting token refresh" } + triedRefresh = true + val refreshed = tokenManager.refreshAccessToken() + if (!refreshed) { + Logger.w("SSE-iOS") { "token refresh failed — session expired" } + tokenManager.emitSessionExpired() + throw e + } + // loop to retry with new token + } else { + throw e + } + } + } + } + + private fun openConnection( + streamPath: String, + resume: Boolean, + bearerToken: String?, + ): Flow = callbackFlow { + // Normalize base URL + stream path so there's exactly one slash between + // them. ServerUrlProvider may or may not return a trailing slash, and + // SseClient passes the stream path without a leading slash, matching + // how ChatRepositoryImpl builds it ("api/agents/chat/stream/$streamId"). + val baseUrl = serverUrlProvider.getBaseUrl().trimEnd('/') + val normalizedPath = streamPath.trimStart('/') + val queryString = if (resume) "?resume=true" else "" + val fullUrl = "$baseUrl/$normalizedPath$queryString" + val url = NSURL.URLWithString(fullUrl) + ?: run { + close(SseStreamException("invalid SSE URL: $fullUrl")) + return@callbackFlow + } + + val scheme = url.scheme?.lowercase() ?: "https" + // The iOS NWConnection transport only supports HTTPS. Plain HTTP would + // require passing NW_PARAMETERS_DISABLE_PROTOCOL for the TLS layer, + // which is a non-null block sentinel that we cannot construct from + // Kotlin/Native without an Obj-C/Swift shim. Production LibreChat + // deployments run HTTPS (Cloudflare, reverse proxies, etc.), so we + // refuse non-HTTPS here instead of silently misconfiguring. File an + // issue if you actually need plain HTTP support. + if (scheme != "https") { + close(SseStreamException("iOS SSE transport only supports HTTPS (got scheme=$scheme)")) + return@callbackFlow + } + val host = url.host ?: run { + close(SseStreamException("missing host in SSE URL: $url")) + return@callbackFlow + } + val port = (url.port?.intValue ?: DEFAULT_HTTPS_PORT).toUShort() + val requestPath = buildRequestPath(url) + + val endpoint = nw_endpoint_create_host(host, port.toString()) + // Default TLS+TCP parameters for NWConnection must come from a C shim. + // + // `nw_parameters_create_secure_tcp` takes two `configure` block-pointer + // arguments. Apple exposes `NW_PARAMETERS_DEFAULT_CONFIGURATION` (which + // expands to the exported global `_nw_parameters_configure_protocol_default_configuration`) + // as the "no-op / use defaults" sentinel. We need to pass that sentinel + // here, but we cannot reach it from Kotlin without crashing: + // + // - A Kotlin lambda gets wrapped by Kotlin/Native's cinterop bridge + // into a synthetic Obj-C block whose parameter type is `NSObject?`. + // Network.framework then invokes the block with a runtime argument + // of type `Network.ProtocolOptions` — a Swift + // generic class that does NOT inherit from NSObject on modern iOS. + // The generated teardown path (`block_destroy_helper`) tries to cast + // the Swift generic to NSObject and crashes with + // `TypeCastException: _TtGC7Network15ProtocolOptionsVS_11TLSProtocol_ + // cannot be cast to class platform.darwin.NSObject`. + // + // - Kotlin/Native 2.3.20's platform.Network binding for + // `NW_PARAMETERS_DEFAULT_CONFIGURATION` is a `knifunptr_*` getter + // that returns a Kotlin-wrapped lambda around Apple's block pointer. + // Passing that through `nw_parameters_create_secure_tcp` re-bridges + // it as a fresh synthetic Obj-C block, and the same + // `block_destroy_helper` / `knbridge3` crash fires. + // + // The only fix is a C cinterop shim that calls the macro at C compile + // time and hands the block pointer directly to Network.framework with + // no Kotlin code in the middle. That shim lives in + // `src/iosMain/cinterop/nwparams_defaults.def` and exposes the + // function `librechat_make_default_tls_tcp_parameters`. + val parameters: nw_parameters_t = librechat_make_default_tls_tcp_parameters() + ?: run { + close(SseStreamException("nw_parameters_create_secure_tcp returned null")) + return@callbackFlow + } + + val connection: nw_connection_t = nw_connection_create(endpoint, parameters) + ?: run { + close(SseStreamException("nw_connection_create returned null")) + return@callbackFlow + } + + val queue = dispatch_queue_create("com.garfiec.librechat.sse.nwconnection", null) + val parser = HttpResponseParser() + var handled = false + + val handleError: (Throwable) -> Unit = { err -> + if (!handled) { + handled = true + close(err) + nw_connection_cancel(connection) + } + } + + val handleCompletion: () -> Unit = { + if (!handled) { + handled = true + close(null) + nw_connection_cancel(connection) + } + } + + // Recursive receive loop. Each call to nw_connection_receive arms a + // single-shot callback; we re-arm inside the callback until isComplete + // or an error fires. + fun armReceive() { + nw_connection_receive( + connection = connection, + minimum_incomplete_length = 1u, + maximum_length = RECEIVE_MAX_LENGTH.toUInt(), + ) { data, _, isComplete, error -> + if (handled) return@nw_connection_receive + + if (error != null) { + handleError(mapNwError(error)) + return@nw_connection_receive + } + + if (data != null) { + val bytes = dispatchDataToByteArray(data) + if (bytes != null && bytes.isNotEmpty()) { + val events = parser.feed(bytes) + for (event in events) { + when (event) { + is HttpResponseParser.ParseEvent.HeadersComplete -> { + if (event.statusCode !in SUCCESS_LOW..SUCCESS_HIGH) { + handleError(SseHttpStatusException(event.statusCode)) + return@nw_connection_receive + } + } + + is HttpResponseParser.ParseEvent.BodyChunk -> { + trySend(event.bytes) + } + + HttpResponseParser.ParseEvent.EndOfStream -> { + handleCompletion() + return@nw_connection_receive + } + + is HttpResponseParser.ParseEvent.Error -> { + handleError(SseStreamException("parser: ${event.message}")) + return@nw_connection_receive + } + } + } + } + } + + if (isComplete) { + handleCompletion() + return@nw_connection_receive + } + + armReceive() + } + } + + nw_connection_set_state_changed_handler(connection) { state, error -> + when (state) { + nw_connection_state_ready -> { + val request = buildHttpRequest( + path = requestPath, + host = host, + bearerToken = bearerToken, + ) + val requestBytes = request.encodeToByteArray() + val dispatchData = byteArrayToDispatchData(requestBytes, queue) + if (dispatchData == null) { + handleError(SseStreamException("failed to create dispatch_data for request")) + return@nw_connection_set_state_changed_handler + } + nw_connection_send( + connection = connection, + content = dispatchData, + context = NW_CONTENT_CONTEXT_DEFAULT_MESSAGE, + is_complete = true, + completion = { sendError -> + if (sendError != null) { + handleError(mapNwError(sendError)) + } else { + armReceive() + } + }, + ) + } + + nw_connection_state_failed -> { + val mapped = error?.let { mapNwError(it) } + ?: SseStreamException("nw_connection failed with no error") + handleError(mapped) + } + + nw_connection_state_cancelled -> { + handleCompletion() + } + + else -> { + // preparing / waiting / invalid — nothing to do; we wait for ready/failed + } + } + } + + nw_connection_set_queue(connection, queue) + nw_connection_start(connection) + + awaitClose { + if (!handled) { + handled = true + } + nw_connection_cancel(connection) + // force_cancel ensures pending receives abort immediately so the + // coroutine frame can tear down promptly. + nw_connection_force_cancel(connection) + } + } + + private fun buildRequestPath(url: NSURL): String { + val path = url.path ?: "/" + val query = url.query + val effectivePath = if (path.isEmpty()) "/" else path + return if (query.isNullOrEmpty()) effectivePath else "$effectivePath?$query" + } + + private fun buildHttpRequest( + path: String, + host: String, + bearerToken: String?, + ): String = buildString { + append("GET ").append(path).append(" HTTP/1.1\r\n") + append("Host: ").append(host).append("\r\n") + append("Accept: text/event-stream\r\n") + if (bearerToken != null) { + append("Authorization: Bearer ").append(bearerToken).append("\r\n") + } + append("User-Agent: ").append(LibreChatHttpClient.BROWSER_USER_AGENT).append("\r\n") + append("Accept-Encoding: identity\r\n") + append("Connection: close\r\n") + append("\r\n") + } + + private fun mapNwError(error: nw_error_t): Throwable { + val domain = nw_error_get_error_domain(error) + val code = nw_error_get_error_code(error).toInt() + return when (domain) { + nw_error_domain_posix -> { + // ECONNRESET / EPIPE / ETIMEDOUT are retryable — let SseClient's + // exponential backoff retry loop handle them via SseStreamException. + if (code == ECONNRESET || code == EPIPE || code == ETIMEDOUT) { + SseStreamException("network error (posix $code)") + } else { + SseStreamException("network error (posix $code)") + } + } + + nw_error_domain_tls -> { + // TLS errors (bad cert, handshake failure, pinning failure) are + // non-retryable — the operator has to fix the cert. Wrap as a + // plain Exception so SseClient's retry loop falls through to the + // generic-error branch and shows a user-facing error. + Exception("TLS error (code $code) — check the server certificate") + } + + else -> { + // dns, other, invalid — retryable in spirit, not worth differentiating. + SseStreamException("nw_error domain=$domain code=$code") + } + } + } + + private fun dispatchDataToByteArray(data: dispatch_data_t): ByteArray? { + val size = dispatch_data_get_size(data).toInt() + if (size == 0) return ByteArray(0) + return memScoped { + // dispatch_data_create_map returns a new contiguous dispatch_data whose + // backing storage is guaranteed contiguous, and fills in `bufferPtr` / + // `sizePtr` with a pointer+length into that storage. The returned + // dispatch_data_t must outlive our read of the pointer, which it does + // because we copy bytes out before leaving this scope. + val sizeOut = alloc() + val bufferOut = alloc() + val mapped = dispatch_data_create_map(data, bufferOut.ptr, sizeOut.ptr) + mapped ?: return@memScoped null + val mappedSize = sizeOut.value.toInt() + val ptr = bufferOut.value ?: return@memScoped null + ptr.readBytes(mappedSize) + } + } + + private fun byteArrayToDispatchData(bytes: ByteArray, queue: platform.darwin.dispatch_queue_t): dispatch_data_t? { + if (bytes.isEmpty()) { + return dispatch_data_create(null, 0u, queue, null) + } + return bytes.usePinned { pinned -> + // Passing null for the destructor block tells dispatch to use + // DISPATCH_DATA_DESTRUCTOR_DEFAULT, which copies the buffer into + // its own allocation so we can release the pin after this call. + dispatch_data_create( + pinned.addressOf(0), + bytes.size.convert(), + queue, + null, + ) + } + } + + private companion object { + const val DEFAULT_HTTPS_PORT = 443 + const val RECEIVE_MAX_LENGTH = 16384 + const val SUCCESS_LOW = 200 + const val SUCCESS_HIGH = 299 + + // NW_CONTENT_CONTEXT_DEFAULT_MESSAGE — passed to nw_connection_send as + // the context argument; it represents a fresh message context. In + // Kotlin/Native this surfaces as a global that we resolve by name at + // link time; for our one-shot request we can pass null and let the + // default-message context apply automatically (Apple docs confirm + // null is equivalent to default-message for the send path). + val NW_CONTENT_CONTEXT_DEFAULT_MESSAGE: nw_content_context_t? = null + } +} diff --git a/iosApp/iosApp/KoinHelper.swift b/iosApp/iosApp/KoinHelper.swift index 576faba..3172511 100644 --- a/iosApp/iosApp/KoinHelper.swift +++ b/iosApp/iosApp/KoinHelper.swift @@ -18,11 +18,6 @@ enum KoinHelper { try! IosKoinAccessor.shared.getServerDataStore() } - /// The streaming-qualified HttpClient for SSE connections - static var streamingHttpClient: Ktor_client_coreHttpClient { - try! IosKoinAccessor.shared.getStreamingHttpClient() - } - static var authRepository: any AuthRepository { try! IosKoinAccessor.shared.getAuthRepository() } diff --git a/shared/src/commonMain/kotlin/com/garfiec/librechat/shared/LibreChatSDK.kt b/shared/src/commonMain/kotlin/com/garfiec/librechat/shared/LibreChatSDK.kt index f847ffb..c8dd117 100644 --- a/shared/src/commonMain/kotlin/com/garfiec/librechat/shared/LibreChatSDK.kt +++ b/shared/src/commonMain/kotlin/com/garfiec/librechat/shared/LibreChatSDK.kt @@ -10,7 +10,8 @@ import com.garfiec.librechat.core.network.sse.SseClient * Top-level entry point that iOS uses to interact with the KMP business logic. * Aggregates the key APIs needed for login + SSE streaming. * - * SSE usage: call `sseClient.connect(streamingHttpClient, streamPath)` directly. + * SSE usage: call `sseClient.connect(streamPath)` directly — the transport it + * needs is injected into `SseClient` at construction time. * SKIE converts the returned `Flow` to `AsyncSequence` on iOS. */ class LibreChatSDK( diff --git a/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosKoinAccessor.kt b/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosKoinAccessor.kt index 4f6a807..09bbe2c 100644 --- a/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosKoinAccessor.kt +++ b/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosKoinAccessor.kt @@ -1,11 +1,9 @@ package com.garfiec.librechat.shared -import com.garfiec.librechat.core.common.di.KoinQualifiers import com.garfiec.librechat.core.data.datastore.ServerDataStore import com.garfiec.librechat.core.data.repository.AuthRepository import com.garfiec.librechat.core.data.repository.ConfigRepository import com.garfiec.librechat.core.data.repository.FileRepository -import io.ktor.client.HttpClient import org.koin.core.Koin /** @@ -27,10 +25,6 @@ object IosKoinAccessor { @Throws(Exception::class) fun getServerDataStore(): ServerDataStore = koin.get() - @Throws(Exception::class) - fun getStreamingHttpClient(): HttpClient = - koin.get(qualifier = KoinQualifiers.Streaming) - @Throws(Exception::class) fun getAuthRepository(): AuthRepository = koin.get() diff --git a/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosSharedModule.kt b/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosSharedModule.kt index 398ce21..32d8720 100644 --- a/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosSharedModule.kt +++ b/shared/src/iosMain/kotlin/com/garfiec/librechat/shared/IosSharedModule.kt @@ -24,11 +24,11 @@ import com.garfiec.librechat.core.network.api.ShareApi import com.garfiec.librechat.core.network.api.SpeechApi import com.garfiec.librechat.core.network.api.TagsApi import com.garfiec.librechat.core.network.api.UserApi -import com.garfiec.librechat.core.network.client.AuthInterceptorPlugin import com.garfiec.librechat.core.network.client.LibreChatHttpClient import com.garfiec.librechat.core.network.client.ServerUrlProvider import com.garfiec.librechat.core.network.client.TokenManager import com.garfiec.librechat.core.network.sse.SseClient +import com.garfiec.librechat.core.network.sse.SseHttpTransport import com.garfiec.librechat.feature.agents.di.agentsModule import com.garfiec.librechat.feature.auth.di.authModule import com.garfiec.librechat.feature.chat.di.chatModule @@ -41,7 +41,6 @@ import io.ktor.client.plugins.HttpTimeout import io.ktor.client.plugins.contentnegotiation.ContentNegotiation import io.ktor.client.plugins.defaultRequest import io.ktor.http.ContentType -import io.ktor.http.HttpHeaders import io.ktor.http.contentType import io.ktor.http.takeFrom import io.ktor.serialization.kotlinx.json.json @@ -84,32 +83,6 @@ val iosSharedModule = module { ) } - // Streaming HttpClient (long-lived SSE connections) - single(KoinQualifiers.Streaming) { - val tokenManager = get() - val serverUrlProvider = get() - HttpClient(Darwin) { - install(AuthInterceptorPlugin) { - this.tokenManager = tokenManager - } - install(HttpTimeout) { - connectTimeoutMillis = 10_000 - requestTimeoutMillis = Long.MAX_VALUE - socketTimeoutMillis = Long.MAX_VALUE - } - defaultRequest { - val baseUrl = serverUrlProvider.getBaseUrl() - if (baseUrl.isNotEmpty()) { - url.takeFrom(baseUrl) - } - headers.append( - HttpHeaders.UserAgent, - LibreChatHttpClient.BROWSER_USER_AGENT, - ) - } - } - } - // Refresh HttpClient (no auth interceptor, short timeout) single(KoinQualifiers.Refresh) { val json = get() @@ -152,6 +125,7 @@ val iosSharedModule = module { singleOf(::McpApi) singleOf(::MemoriesApi) singleOf(::SpeechApi) + single { SseHttpTransport(get(), get()) } singleOf(::SseClient) // SDK facade