fix(#4882,#4883): streaming surfaces provider HTTP errors + Kimi region modes

CHANGELOG.md

@@ -4,6 +4,24 @@ All notable changes to Agents.KT are documented here. The format follows [Keep a
 
 ## [Unreleased]
 
+### Fixed — streaming now surfaces provider HTTP errors instead of swallowing them (#4882)
+
+`OpenAiClient.chatStream` — and every OpenAI-compatible subclass (OpenAI, DeepSeek, Kimi, OpenRouter,
+Perplexity) — previously returned the raw response body on a **non-2xx** streaming response without
+checking the status. An error body has no `data:` lines, so the SSE parser emitted a lone terminal
+`End`: a **silent, empty, success-looking stream**. A stream started with an expired key, a `429`, or a
+provider `5xx` returned nothing instead of raising. `sendChatStream` now checks `statusCode()` and throws
+`LlmProviderException` (HTTP status + provider label + a bounded slice of the error body), matching the
+non-streaming `chat()` contract. Kimi's region-hint wrapping (#4511) still applies on auth errors.
+
+### Added — Kimi region modes: `KimiRegion.China` / `KimiRegion.International` (#4883)
+
+Moonshot/Kimi runs two independent platforms with **non-interchangeable** keys. A new `KimiRegion` enum
+plus a `model { kimi("moonshot-v1-8k", region = KimiRegion.INTERNATIONAL) }` DSL overload make the region
+an explicit, typed choice instead of a raw `baseUrl` string. Additive: `kimi("...")` with no region is
+byte-identical to before (China default preserved); `KimiRegion.INTERNATIONAL.baseUrl` is also usable as
+the `KimiClient(baseUrl = …)` argument directly.
+
 ### Changed — x402 buyer trust hardening: guardrails are now mandatory and bind more (#4528)
 
 An external audit flagged that the "guardrails-first" buyer had **optional** guardrails (an empty

docs/providers.md

@@ -123,6 +123,27 @@ Credentials load from `.secrets/gemini-key` / `GEMINI_API_KEY`. Live tests are t
 (run in the default suite when a key is present, skip otherwise; they also skip on free-tier
 `RESOURCE_EXHAUSTED` rate limits). Verified end-to-end against `gemini-2.5-flash`.
 
+## Kimi regions (China / International)
+
+Moonshot/Kimi runs **two independent platforms** with **non-interchangeable** keys — a key from one is
+rejected by the other with `Invalid Authentication`:
+
+| Region | Base URL | Keys from |
+|---|---|---|
+| `KimiRegion.CHINA` (default) | `https://api.moonshot.cn` | platform.moonshot.cn |
+| `KimiRegion.INTERNATIONAL` | `https://api.moonshot.ai` | platform.moonshot.ai |
+
+Pick it explicitly in the DSL (#4883):
+
+```kotlin
+model { kimi("moonshot-v1-8k", region = KimiRegion.INTERNATIONAL) }   // international key
+model { kimi("moonshot-v1-8k") }                                       // China default (unchanged)
+```
+
+`KimiRegion.INTERNATIONAL.baseUrl` is also usable directly as `KimiClient(baseUrl = …)`. On a region
+mismatch the auth error names the *other* endpoint and how to switch (#4511). Credentials load from
+`.secrets/kimi-key` / `KIMI_API_KEY`.
+
 ## Updating this matrix
 
 The four columns here track the OpenAI-family `ModelProvider.entries`; `GEMINI` is the fifth wire

src/main/kotlin/agents_engine/model/KimiRegion.kt

@@ -0,0 +1,16 @@
+package agents_engine.model
+
+/**
+ * #4883 — Moonshot/Kimi runs two independent platforms with **non-interchangeable** keys: a key
+ * issued by one is rejected by the other with `Invalid Authentication` (see [KimiClient.regionAwareError]).
+ * Pick the region explicitly — `model { kimi("moonshot-v1-8k", region = KimiRegion.INTERNATIONAL) }` —
+ * instead of passing a raw `baseUrl` string. The [baseUrl] is also usable directly:
+ * `KimiClient(apiKey, model, baseUrl = KimiRegion.INTERNATIONAL.baseUrl)`.
+ */
+enum class KimiRegion(val baseUrl: String) {
+    /** platform.moonshot.cn — the historical default ([KimiClient.CHINA_BASE_URL]). */
+    CHINA(KimiClient.CHINA_BASE_URL),
+
+    /** platform.moonshot.ai ([KimiClient.INTERNATIONAL_BASE_URL]). */
+    INTERNATIONAL(KimiClient.INTERNATIONAL_BASE_URL),
+}

src/main/kotlin/agents_engine/model/ModelBuilder.kt

@@ -80,9 +80,12 @@ class ModelBuilder {
      * is available. Model names follow Moonshot's naming, e.g.
      * `"moonshot-v1-8k"`, `"moonshot-v1-32k"`, `"moonshot-v1-128k"`.
      */
-    fun kimi(modelName: String) {
+    fun kimi(modelName: String, region: KimiRegion? = null) {
         name = modelName
         provider = ModelProvider.KIMI
+        // #4883 — only override the base URL when a region is explicitly chosen, so existing
+        // `kimi("...")` calls (and any manual `kimiBaseUrl = ...`) keep the China default untouched.
+        if (region != null) kimiBaseUrl = region.baseUrl
     }
 
     /**

src/main/kotlin/agents_engine/model/OpenAiClient.kt

@@ -167,6 +167,19 @@ open class OpenAiClient(
             .POST(HttpRequest.BodyPublishers.ofString(body))
         headers.forEach { (k, v) -> builder.header(k, v) }
         val response = http.send(builder.build(), HttpResponse.BodyHandlers.ofInputStream())
+        // #4882 — mirror the non-streaming contract (sendChat -> HttpModelClientSupport.sendBounded
+        // throws on 4xx/5xx). A streaming error MUST surface as an exception, not be swallowed by
+        // parseSseStream (which skips every non-`data:` line and emits a lone End) into a silent,
+        // empty, success-looking stream. Read a bounded slice of the error body for the message.
+        if (response.statusCode() !in 200..299) {
+            val errorBody = response.body().use { stream ->
+                val cap = minOf(maxResponseBytes, STREAM_ERROR_BODY_CAP).toInt()
+                String(stream.readNBytes(cap), Charsets.UTF_8).trim()
+            }
+            throw LlmProviderException(
+                "$providerLabel streaming request failed with HTTP ${response.statusCode()}: $errorBody",
+            )
+        }
         return response.body()
     }
 
@@ -459,6 +472,11 @@ open class OpenAiClient(
         val DEFAULT_CONNECT_TIMEOUT: Duration = 10.seconds
         const val DEFAULT_MAX_RESPONSE_BYTES: Long = 16L * 1024 * 1024
         const val DEFAULT_MAX_TOKENS: Int = 4096
+
+        // #4882 — cap the error-body slice read when a streaming request returns non-2xx.
+        // Provider error envelopes are small; 64 KiB is ample and bounds the read against a
+        // hostile/misbehaving endpoint.
+        private const val STREAM_ERROR_BODY_CAP: Long = 64L * 1024
     }
 }
 

src/main/resources/internals-agent/model/OpenAiClient.md

@@ -63,6 +63,8 @@ When `chat(messages, jsonSchema)` receives a non-null `JsonSchema`, the request
 
 OpenAI's `{"error": {"message": "...", "type": "...", ...}}` surfaces as `LlmProviderException` (#702 — same contract as the other adapters).
 
+Streaming honors the same contract (#4882): `sendChatStream` checks `response.statusCode()` and, on a non-2xx, throws `LlmProviderException` (HTTP status + provider label + a bounded slice of the error body) instead of handing the error body to `parseSseStream` — which would skip the non-`data:` lines and emit a lone `End`, i.e. a silent empty stream. Applies to every OpenAI-compatible subclass (DeepSeek/Kimi/OpenRouter/Perplexity).
+
 ## Test seam
 
 `sendChat(body, headers)` is `open`. Used by `OpenAiClientTest` / `OpenAiClientStreamingTest`.

src/test/kotlin/agents_engine/model/KimiRegionModeTest.kt

@@ -0,0 +1,43 @@
+package agents_engine.model
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+// #4883 — Kimi/Moonshot runs two independent platforms; the region is a first-class, typed choice.
+class KimiRegionModeTest {
+
+    @Test
+    fun `KimiRegion exposes both platform base URLs`() {
+        assertEquals("https://api.moonshot.cn", KimiRegion.CHINA.baseUrl)
+        assertEquals("https://api.moonshot.ai", KimiRegion.INTERNATIONAL.baseUrl)
+    }
+
+    @Test
+    fun `region source values match the KimiClient constants`() {
+        assertEquals(KimiClient.CHINA_BASE_URL, KimiRegion.CHINA.baseUrl)
+        assertEquals(KimiClient.INTERNATIONAL_BASE_URL, KimiRegion.INTERNATIONAL.baseUrl)
+    }
+
+    @Test
+    fun `kimi DSL without a region preserves the China default`() {
+        val builder = ModelBuilder()
+        builder.kimi("moonshot-v1-8k")
+        assertEquals(KimiClient.DEFAULT_BASE_URL, builder.kimiBaseUrl, "no region => China default unchanged")
+        assertEquals(ModelProvider.KIMI, builder.provider)
+        assertEquals("moonshot-v1-8k", builder.name)
+    }
+
+    @Test
+    fun `kimi DSL with INTERNATIONAL region selects the moonshot ai base URL`() {
+        val builder = ModelBuilder()
+        builder.kimi("moonshot-v1-8k", region = KimiRegion.INTERNATIONAL)
+        assertEquals("https://api.moonshot.ai", builder.kimiBaseUrl)
+    }
+
+    @Test
+    fun `kimi DSL with CHINA region selects the moonshot cn base URL`() {
+        val builder = ModelBuilder()
+        builder.kimi("moonshot-v1-128k", region = KimiRegion.CHINA)
+        assertEquals("https://api.moonshot.cn", builder.kimiBaseUrl)
+    }
+}

src/test/kotlin/agents_engine/model/OpenAiClientStreamErrorTest.kt

@@ -0,0 +1,64 @@
+package agents_engine.model
+
+import com.sun.net.httpserver.HttpServer
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import java.net.InetSocketAddress
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+// #4882 — the streaming path must surface a non-2xx HTTP response as LlmProviderException,
+// matching the non-streaming contract (sendChat -> HttpModelClientSupport.sendBounded throws on
+// 4xx/5xx). Previously `sendChatStream` returned `response.body()` unchecked, so an error body
+// (no `data:` lines) was swallowed by parseSseStream into a silent, empty, success-looking stream
+// (a lone End) — no exception, no observability. Confirmed against a real 401 from api.moonshot.cn.
+class OpenAiClientStreamErrorTest {
+
+    private var server: HttpServer? = null
+
+    private fun serverReturning(status: Int, body: String): String {
+        val srv = HttpServer.create(InetSocketAddress("127.0.0.1", 0), 0).apply {
+            createContext("/v1/chat/completions") { ex ->
+                val bytes = body.toByteArray()
+                ex.sendResponseHeaders(status, bytes.size.toLong())
+                ex.responseBody.use { it.write(bytes) }
+                ex.close()
+            }
+            executor = null
+            start()
+        }
+        server = srv
+        return "http://localhost:${srv.address.port}"
+    }
+
+    @AfterTest fun stop() {
+        server?.stop(0)
+    }
+
+    @Test
+    fun `streaming surfaces a 401 as LlmProviderException with status and body`() = runBlocking {
+        val baseUrl = serverReturning(401, """{"error":{"message":"Invalid Authentication"}}""")
+        val client = OpenAiClient(apiKey = "sk-bad", model = "gpt-4o-mini", baseUrl = baseUrl)
+
+        val ex = assertFailsWith<LlmProviderException> {
+            client.chatStream(listOf(LlmMessage("user", "hi"))).toList()
+        }
+        val msg = ex.message.orEmpty()
+        assertTrue("401" in msg, "message must name the HTTP status: $msg")
+        assertTrue("Invalid Authentication" in msg, "message must include the provider error body: $msg")
+        assertTrue("OpenAI" in msg, "message must name the provider label: $msg")
+    }
+
+    @Test
+    fun `streaming surfaces a 5xx as LlmProviderException`() = runBlocking {
+        val baseUrl = serverReturning(503, """{"error":{"message":"upstream unavailable"}}""")
+        val client = OpenAiClient(apiKey = "sk", model = "gpt-4o-mini", baseUrl = baseUrl)
+
+        val ex = assertFailsWith<LlmProviderException> {
+            client.chatStream(listOf(LlmMessage("user", "hi"))).toList()
+        }
+        assertTrue("503" in ex.message.orEmpty(), "message must name the 5xx status: ${ex.message}")
+    }
+}