From b5bde38d732c5406f892687073c898a1dde27fe1 Mon Sep 17 00:00:00 2001 From: skobeltsyn Date: Fri, 3 Jul 2026 12:12:28 +0300 Subject: [PATCH] fix(#4882,#4883): streaming surfaces provider HTTP errors + Kimi region modes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #4882 — OpenAI-compatible streaming no longer swallows non-2xx responses. `OpenAiClient.sendChatStream` returned `response.body()` without checking the status; an error body (no `data:` lines) was swallowed by `parseSseStream` into a lone terminal `End` — a silent, empty, success-looking stream. Now checks `statusCode()` and throws `LlmProviderException` (status + provider label + bounded error-body slice), matching the non-streaming `chat()` contract. Fixes silent-empty-output on expired-key / 429 / 5xx for OpenAI, DeepSeek, Kimi, OpenRouter, Perplexity. Kimi's region-hint wrapping (#4511) still applies. TDD: local 401/503 HttpServer → assert LlmProviderException. #4883 — Kimi region modes. New `KimiRegion` enum (CHINA / INTERNATIONAL) + a `model { kimi("model", region = KimiRegion.INTERNATIONAL) }` DSL overload. Additive: `kimi("model")` with no region keeps the China default byte-identical. `KimiRegion.INTERNATIONAL.baseUrl` also usable as `KimiClient(baseUrl = …)`. Docs: OpenAiClient.md adjunct (streaming error boundary), providers.md (Kimi regions section), CHANGELOG. Full ./gradlew build green (detekt, checkOneTypePerFile, DocsConsistency, all guards); sole test red was a flaky live-cloud-api Perplexity call, passed on rerun. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 18 ++++++ docs/providers.md | 21 ++++++ .../kotlin/agents_engine/model/KimiRegion.kt | 16 +++++ .../agents_engine/model/ModelBuilder.kt | 5 +- .../agents_engine/model/OpenAiClient.kt | 18 ++++++ .../internals-agent/model/OpenAiClient.md | 2 + .../agents_engine/model/KimiRegionModeTest.kt | 43 +++++++++++++ .../model/OpenAiClientStreamErrorTest.kt | 64 +++++++++++++++++++ 8 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 src/main/kotlin/agents_engine/model/KimiRegion.kt create mode 100644 src/test/kotlin/agents_engine/model/KimiRegionModeTest.kt create mode 100644 src/test/kotlin/agents_engine/model/OpenAiClientStreamErrorTest.kt diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c1f6986..4ebde065 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/providers.md b/docs/providers.md index 25c5c297..22ef5865 100644 --- a/docs/providers.md +++ b/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 diff --git a/src/main/kotlin/agents_engine/model/KimiRegion.kt b/src/main/kotlin/agents_engine/model/KimiRegion.kt new file mode 100644 index 00000000..7054d651 --- /dev/null +++ b/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), +} diff --git a/src/main/kotlin/agents_engine/model/ModelBuilder.kt b/src/main/kotlin/agents_engine/model/ModelBuilder.kt index ec060022..091c1725 100644 --- a/src/main/kotlin/agents_engine/model/ModelBuilder.kt +++ b/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 } /** diff --git a/src/main/kotlin/agents_engine/model/OpenAiClient.kt b/src/main/kotlin/agents_engine/model/OpenAiClient.kt index 37ed9b31..078b9bc4 100644 --- a/src/main/kotlin/agents_engine/model/OpenAiClient.kt +++ b/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 } } diff --git a/src/main/resources/internals-agent/model/OpenAiClient.md b/src/main/resources/internals-agent/model/OpenAiClient.md index 15a69458..3bdc0299 100644 --- a/src/main/resources/internals-agent/model/OpenAiClient.md +++ b/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`. diff --git a/src/test/kotlin/agents_engine/model/KimiRegionModeTest.kt b/src/test/kotlin/agents_engine/model/KimiRegionModeTest.kt new file mode 100644 index 00000000..e30918e5 --- /dev/null +++ b/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) + } +} diff --git a/src/test/kotlin/agents_engine/model/OpenAiClientStreamErrorTest.kt b/src/test/kotlin/agents_engine/model/OpenAiClientStreamErrorTest.kt new file mode 100644 index 00000000..c85aa434 --- /dev/null +++ b/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 { + 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 { + client.chatStream(listOf(LlmMessage("user", "hi"))).toList() + } + assertTrue("503" in ex.message.orEmpty(), "message must name the 5xx status: ${ex.message}") + } +}