From f23f70a2b12e8603ac2e5f7bdbea74eb77a2ea73 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=AF=97=E9=80=94=E4=B9=90=E6=97=85?= <3296249574@qq.com> Date: Mon, 19 Oct 2026 15:24:30 +0800 Subject: [PATCH 1/2] feat: add bounded SSE event decoding and explicit upstream rejection classification --- README.md | 38 ++++- build.gradle.kts | 2 +- .../OpenCodeZenSseEventReader.java | 143 ++++++++++++++++++ .../OpenCodeZenUpstreamDiagnostics.java | 56 +++++++ .../OpenCodeZenResponseParityTest.java | 104 +++++++++++++ 5 files changed, 336 insertions(+), 7 deletions(-) create mode 100644 opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenSseEventReader.java create mode 100644 opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenUpstreamDiagnostics.java create mode 100644 opencode-zen-spring-boot-autoconfigure/src/test/java/com/sharedlibs/opencodezen/OpenCodeZenResponseParityTest.java diff --git a/README.md b/README.md index 5f0f1ea..c54e31d 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ JVM 共享库，用于与 OpenCode Zen 交互：协议常量、OpenCode 2.x 客 | `opencode-zen-spring-boot-starter` | 聚合入口（消费方只需要这一个坐标） | - group：`com.sharedlibs` -- 当前版本：`1.3.1` +- 当前版本：`1.3.2` --- @@ -46,7 +46,7 @@ repositories { } dependencies { - implementation("com.sharedlibs:opencode-zen-spring-boot-starter:1.3.1") + implementation("com.sharedlibs:opencode-zen-spring-boot-starter:1.3.2") } ``` @@ -113,7 +113,7 @@ $env:OPENCODE_ZEN_LOCAL="E:\project\shared-libs\opencode-zen" # 改完库立 --- -## 1.3.1：根据已可用的 Python V2.4 修复 Java 共享库 +## 1.3.2：根据已可用的 Python V2.4 修复 Java 共享库 **注意：本次修改的是 Java 共享库，不是 Python V2.4 代理，也不是 Python V2.5。** @@ -151,10 +151,10 @@ HttpResponse upstream = HttpClient.newHttpClient().send( - **旧方法仍保留**：`newConversationRequest` / `newEphemeralRequest` 只返回附带 Header 的 `HttpRequest.Builder`。若消费方仍手写 Body，必须自行调用 `applyRequiredEnvelope`，或迁移到新方法。 - Java 库仍只是共享协议/请求工具，并不是完整 HTTP 代理；SSE 拼包、工具调用过滤和上游错误透传由消费方处理。Synthetic 工具**只能作为占位声明，不能执行**。 - 本次用 Python V2.4 作协议参照，**不意味着真实 Zen 网关已经验证成功**。上游 403/429 要尊重、原样处理，不得通过自动轮换网络出口绕过服务端限制。 -- 1.3.1 用于新的 Maven 发布，原有 1.2.0 产物不受影响；下游需要显式升级依赖。 +- 1.3.2 用于新的 Maven 发布，原有 1.2.0 产物不受影响；下游需要显式升级依赖。 -## 1.3.1：Python V2.4 工具声明语义修复（未触碰 Python V2.4） +## 1.3.2：Python V2.4 工具声明语义修复（未触碰 Python V2.4） 这次实际对比了上传的 Python V2.4 中的 `_names_for_tools()` 与其 63 项离线测试。 @@ -172,7 +172,7 @@ Python V2.4 的 `_gateway_rejection()` 会把含 `Detect Abuse Traffic` 的 403 Java `HttpRequest` Factory 并不会配置 `HttpClient` 的代理：`127.0.0.1:7897` 若要用于 Java HTTP 连接，必须在创建 `HttpClient` 时按实际网络需要显式配置。 本次结果只保证离线行为对齐；不代表真实 Zen 推理已经得到 200。 -升级消费者至 `com.sharedlibs:opencode-zen-spring-boot-starter:1.3.1` 前需发布新 Maven 版本。 +升级消费者至 `com.sharedlibs:opencode-zen-spring-boot-starter:1.3.2` 前需发布新 Maven 版本。 ### Java 调用端如何明确选择网络连接（不自动切换出口） @@ -189,3 +189,29 @@ HttpResponse result = http.send(request, HttpResponse.BodyHandlers. `OpenCodeZenHttpClients.direct()` 禁用代理；`system()` 使用 JVM 默认 ProxySelector（不能保证自动读取 HTTPS_PROXY 环境变量）。 如果 Java 项目属于转发网关，可选使用 Factory 的 `newConversationInferenceRequest(..., payload, incomingHeaders)` 或 `newEphemeralInferenceRequest(..., payload, incomingHeaders)` 重载：仅透传项目名、经 Python V2.4 会话 counter 校验通过的 session，以及 Anthropic beta/version；不接受从该 map 覆盖 Authorization / x-api-key。 请勿以 HTTP 403 是否变为 200 来判断工具描述归一化是否正确；先在离线测试中比较 Header 和 JSON Body。 + +## 1.3.2 — V2.4 上游结果判定与 SSE 消费端支持 + +本次仅增加正常、授权请求的结果处理，不改变原有身份 Header、工具指纹、认证逻辑或网络出口，不尝试绕过上游 403 访问限制。 + +- `OpenCodeZenUpstreamDiagnostics.classify(status, responseBytes)`：将 403 Detect Abuse Traffic 判定为 `ABUSE_REJECTED`，普通 FreeTierError 为 `FREE_TIER_REJECTED`，HTTP 429 为 `RATE_LIMITED`，其他错误独立分类。仅检查最多 64 KiB；不会保存响应或敏感 Header。 +- `OpenCodeZenSseEventReader.read(inputStream, eventConsumer)`：增量解析具名 SSE 事件、换行/CRLF、跨缓冲区 UTF-8、多行 JSON 和 `[DONE]`；上游 `error` / `response.failed` fail-closed；总流量限制 32 MiB。 +- 测试不访问真实 Zen 网关，只检查 Python V2.4 中可确认的响应解析和拒绝分类语义。 + +### 结果分类与 SSE 用法 + +```java +HttpResponse response = http.send(request, HttpResponse.BodyHandlers.ofByteArray()); +var result = OpenCodeZenUpstreamDiagnostics.classify(response.statusCode(), response.body()); +if (result == OpenCodeZenUpstreamDiagnostics.Outcome.ABUSE_REJECTED) { + // 尊重服务端限制：停止调用，并通过服务商支持渠道处理。 +} + +// 当上游返回 text/event-stream 时，建议使用 BodyHandlers.ofInputStream() 直接流式处理： +OpenCodeZenSseEventReader.read(response.body(), event -> { + if (event.done()) return; + // 对每个 JsonNode 做模型协议和工具授权校验，再发给下游。 +}); +``` + +注意：第二段示例的 `response` 应由 `HttpResponse` 取得，而非上面的 `byte[]` 响应；具体业务中请选择一种 BodyHandler。不得将 `200 /models` 当作推理授权成功证明。 diff --git a/build.gradle.kts b/build.gradle.kts index 96182c8..8fbb7d7 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -3,7 +3,7 @@ plugins { } group = "com.sharedlibs" -version = "1.3.1" +version = "1.3.2" subprojects { group = rootProject.group diff --git a/opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenSseEventReader.java b/opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenSseEventReader.java new file mode 100644 index 0000000..35e9c9c --- /dev/null +++ b/opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenSseEventReader.java @@ -0,0 +1,143 @@ +package com.sharedlibs.opencodezen; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Locale; +import java.util.Objects; +import java.util.function.Consumer; + +/** + * Streaming SSE reader aligned with Python V2.4's _stream_events, without + * changing HTTP identity, authorization or retry decisions. + * + * This reader never retries an upstream rejection. The client supplies the + * stream from its own authorized HttpClient. + */ +public final class OpenCodeZenSseEventReader { + public static final int MAX_SSE_BYTES = 32 * 1024 * 1024; + private static final ObjectMapper JSON = new ObjectMapper(); + + public record Event(String name, JsonNode payload, boolean done) {} + + public static final class StreamFailure extends IOException { + private final String code; + public StreamFailure(String code, String message) { + super(message); + this.code = code; + } + public StreamFailure(String code, String message, Throwable cause) { + super(message, cause); + this.code = code; + } + public String code() { return code; } + } + + private OpenCodeZenSseEventReader() {} + + /** + * Emit parsed events as they arrive. Consumes at most 32 MiB in total. + * Rejects malformed event JSON, explicit upstream error events, and + * streams ending in the middle of a line. + */ + public static void read(InputStream upstream, Consumer receiver) throws IOException { + Objects.requireNonNull(upstream, "upstream"); + Objects.requireNonNull(receiver, "receiver"); + ByteArrayOutputStream currentLine = new ByteArrayOutputStream(); + ByteArrayOutputStream frameData = new ByteArrayOutputStream(); + byte[] buffer = new byte[8192]; + String[] eventName = {""}; + boolean[] firstData = {true}; + int total = 0; + + for (int count; (count = upstream.read(buffer)) != -1;) { + if (count == 0) { + continue; + } + total += count; + if (total > MAX_SSE_BYTES) { + throw new StreamFailure("stream_too_large", "Upstream SSE exceeded 32 MiB"); + } + for (int i = 0; i < count; i++) { + byte current = buffer[i]; + if (current == '\n') { + byte[] bytes = currentLine.toByteArray(); + currentLine.reset(); + int length = bytes.length; + if (length > 0 && bytes[length - 1] == '\r') { + length--; + } + if (length == 0) { + emit(frameData, eventName, firstData, receiver, false); + } else if (startsWith(bytes, length, "data:")) { + int offset = 5; + while (offset < length && bytes[offset] == ' ') { + offset++; + } + if (!firstData[0]) { + frameData.write('\n'); + } + frameData.write(bytes, offset, length - offset); + firstData[0] = false; + } else if (startsWith(bytes, length, "event:")) { + int offset = 6; + String name = new String(bytes, offset, length - offset, StandardCharsets.UTF_8); + eventName[0] = name.trim(); + } + } else { + currentLine.write(current); + } + } + } + if (currentLine.size() > 0 + && !new String(currentLine.toByteArray(), StandardCharsets.UTF_8).trim().isEmpty()) { + throw new StreamFailure("truncated_sse", "Upstream SSE ended mid-line"); + } + emit(frameData, eventName, firstData, receiver, true); + } + + private static boolean startsWith(byte[] data, int length, String ascii) { + if (length < ascii.length()) return false; + for (int i = 0; i < ascii.length(); i++) { + if (data[i] != (byte) ascii.charAt(i)) return false; + } + return true; + } + + private static void emit( + ByteArrayOutputStream data, String[] eventName, boolean[] firstData, + Consumer receiver, boolean endOfStream) throws IOException { + if (firstData[0]) { + eventName[0] = ""; + return; + } + byte[] raw = data.toByteArray(); + boolean done = raw.length == 6 && new String(raw, StandardCharsets.UTF_8).equals("[DONE]"); + JsonNode json = null; + if (!done) { + try { + json = JSON.readTree(raw); + if (json == null) { + throw new IOException("empty JSON event"); + } + } catch (IOException error) { + throw new StreamFailure( + endOfStream ? "truncated_sse" : "bad_sse", + "Invalid upstream SSE event JSON", error); + } + } + String name = eventName[0]; + if ("error".equals(name) || (json != null && (json.hasNonNull("error") + || "error".equals(json.path("type").asText()) + || "response.failed".equals(json.path("type").asText())))) { + throw new StreamFailure("upstream_stream_error", "Upstream reported an error event"); + } + receiver.accept(new Event(name, json, done)); + data.reset(); + eventName[0] = ""; + firstData[0] = true; + } +} diff --git a/opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenUpstreamDiagnostics.java b/opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenUpstreamDiagnostics.java new file mode 100644 index 0000000..87fd10f --- /dev/null +++ b/opencode-zen-spring-boot-autoconfigure/src/main/java/com/sharedlibs/opencodezen/OpenCodeZenUpstreamDiagnostics.java @@ -0,0 +1,56 @@ +package com.sharedlibs.opencodezen; + +import java.nio.charset.StandardCharsets; +import java.util.Locale; +import java.util.Objects; + +/** + * Bounded, credential-free categorization of an upstream response. + * Classifications describe the server response; they do not change access + * permissions and never trigger proxy switching or retry attempts. + */ +public final class OpenCodeZenUpstreamDiagnostics { + public enum Outcome { + SUCCESS, ABUSE_REJECTED, FREE_TIER_REJECTED, + FORBIDDEN, RATE_LIMITED, OTHER_HTTP_ERROR + } + + private OpenCodeZenUpstreamDiagnostics() {} + + /** + * Mirrors Python V2.4's _gateway_rejection. No response body is retained, + * logged or exposed. This method never inspects user prompts or API keys. + */ + public static Outcome classify(int statusCode, byte[] responseBody) { + if (statusCode >= 200 && statusCode <= 299) { + return Outcome.SUCCESS; + } + if (statusCode == 429) { + return Outcome.RATE_LIMITED; + } + if (statusCode == 403) { + byte[] bytes = responseBody == null ? new byte[0] : responseBody; + int length = Math.min(bytes.length, 65536); + String message = new String(bytes, 0, length, StandardCharsets.UTF_8) + .toLowerCase(Locale.ROOT); + if (message.contains("detect abuse traffic") + || message.contains("abuse traffic")) { + return Outcome.ABUSE_REJECTED; + } + if (message.contains("freetiererror") || message.contains("free tier")) { + return Outcome.FREE_TIER_REJECTED; + } + return Outcome.FORBIDDEN; + } + return Outcome.OTHER_HTTP_ERROR; + } + + /** + * In particular, 403/429 are never automatically retried. + * A caller may handle unrelated connectivity problems separately. + */ + public static boolean shouldRetryAutomatically(Outcome outcome) { + Objects.requireNonNull(outcome, "outcome"); + return false; + } +} diff --git a/opencode-zen-spring-boot-autoconfigure/src/test/java/com/sharedlibs/opencodezen/OpenCodeZenResponseParityTest.java b/opencode-zen-spring-boot-autoconfigure/src/test/java/com/sharedlibs/opencodezen/OpenCodeZenResponseParityTest.java new file mode 100644 index 0000000..65b4f3f --- /dev/null +++ b/opencode-zen-spring-boot-autoconfigure/src/test/java/com/sharedlibs/opencodezen/OpenCodeZenResponseParityTest.java @@ -0,0 +1,104 @@ +package com.sharedlibs.opencodezen; + +import org.junit.jupiter.api.Test; +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Arrays; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class OpenCodeZenResponseParityTest { + @Test + void detectsExplicitTrafficRejectionAndNeverRetries() { + var result = OpenCodeZenUpstreamDiagnostics.classify(403, + "{\"error\":{\"type\":\"FreeTierError\",\"message\":\"Detect Abuse Traffic\"}}" + .getBytes(StandardCharsets.UTF_8)); + assertThat(result).isEqualTo(OpenCodeZenUpstreamDiagnostics.Outcome.ABUSE_REJECTED); + assertThat(OpenCodeZenUpstreamDiagnostics.shouldRetryAutomatically(result)).isFalse(); + } + + @Test + void distinguishesFreeTierAndOtherHttpErrorsWithoutStoringBody() { + assertThat(OpenCodeZenUpstreamDiagnostics.classify(403, + "FreeTierError".getBytes(StandardCharsets.UTF_8))) + .isEqualTo(OpenCodeZenUpstreamDiagnostics.Outcome.FREE_TIER_REJECTED); + assertThat(OpenCodeZenUpstreamDiagnostics.classify(403, "blocked".getBytes(StandardCharsets.UTF_8))) + .isEqualTo(OpenCodeZenUpstreamDiagnostics.Outcome.FORBIDDEN); + assertThat(OpenCodeZenUpstreamDiagnostics.classify(429, null)) + .isEqualTo(OpenCodeZenUpstreamDiagnostics.Outcome.RATE_LIMITED); + assertThat(OpenCodeZenUpstreamDiagnostics.classify(200, null)) + .isEqualTo(OpenCodeZenUpstreamDiagnostics.Outcome.SUCCESS); + } + + @Test + void parsesMultilineCrLfEventsAndTerminalDone() throws Exception { + String sse = ":comment\r\nevent: response.delta\r\ndata: {\"type\":\r\ndata: \"response.output_text.delta\",\"delta\":\"hi\"}\r\n\r\ndata: [DONE]\n\n"; + List events = new ArrayList<>(); + OpenCodeZenSseEventReader.read( + new ByteArrayInputStream(sse.getBytes(StandardCharsets.UTF_8)), events::add); + assertThat(events).hasSize(2); + assertThat(events.get(0).name()).isEqualTo("response.delta"); + assertThat(events.get(0).payload().path("delta").asText()).isEqualTo("hi"); + assertThat(events.get(0).done()).isFalse(); + assertThat(events.get(1).done()).isTrue(); + } + + @Test + void parsesUtf8AcrossChunkBoundaries() throws Exception { + byte[] data = "data: {\"text\":\"中文\"}\r\n\r\n".getBytes(StandardCharsets.UTF_8); + var chunks = new ByteArrayInputStream(data) { + @Override + public synchronized int read(byte[] b, int off, int len) { + return super.read(b, off, Math.min(len, 1)); + } + }; + List events = new ArrayList<>(); + OpenCodeZenSseEventReader.read(chunks, events::add); + assertThat(events).hasSize(1); + assertThat(events.get(0).payload().path("text").asText()).isEqualTo("中文"); + } + + @Test + void recognizesJsonErrorEventWithoutLeakingPayload() { + var source = new ByteArrayInputStream( + "event: error\ndata: {\"error\":{\"message\":\"stop\"}}\n\n".getBytes(StandardCharsets.UTF_8)); + assertThatThrownBy(() -> OpenCodeZenSseEventReader.read(source, event -> {})) + .isInstanceOf(OpenCodeZenSseEventReader.StreamFailure.class) + .satisfies(err -> assertThat( + ((OpenCodeZenSseEventReader.StreamFailure) err).code()) + .isEqualTo("upstream_stream_error")); + } + + @Test + void rejectsIncompleteFramesAndBadJson() { + var truncated = new ByteArrayInputStream( + "data: {\"text\":1}".getBytes(StandardCharsets.UTF_8)); + assertThatThrownBy(() -> OpenCodeZenSseEventReader.read(truncated, event -> {})) + .isInstanceOf(OpenCodeZenSseEventReader.StreamFailure.class) + .satisfies(err -> assertThat( + ((OpenCodeZenSseEventReader.StreamFailure) err).code()) + .isEqualTo("truncated_sse")); + + var badJson = new ByteArrayInputStream( + "data: not-json\n\n".getBytes(StandardCharsets.UTF_8)); + assertThatThrownBy(() -> OpenCodeZenSseEventReader.read(badJson, event -> {})) + .isInstanceOf(OpenCodeZenSseEventReader.StreamFailure.class) + .satisfies(err -> assertThat( + ((OpenCodeZenSseEventReader.StreamFailure) err).code()) + .isEqualTo("bad_sse")); + } + + @Test + void capsOversizedUntrustedStreams() { + byte[] huge = new byte[OpenCodeZenSseEventReader.MAX_SSE_BYTES + 1]; + Arrays.fill(huge, (byte) 'x'); + assertThatThrownBy(() -> OpenCodeZenSseEventReader.read( + new ByteArrayInputStream(huge), event -> {})) + .isInstanceOf(OpenCodeZenSseEventReader.StreamFailure.class) + .satisfies(err -> assertThat( + ((OpenCodeZenSseEventReader.StreamFailure) err).code()) + .isEqualTo("stream_too_large")); + } +} From 518814a1cbd3e283c29e4277fcffaacff409718e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=AF=97=E9=80=94=E4=B9=90=E6=97=85?= <3296249574@qq.com> Date: Mon, 19 Oct 2026 15:25:39 +0800 Subject: [PATCH 2/2] docs: demonstrate streaming with InputStream handler and status checks --- README.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index c54e31d..240ab6f 100644 --- a/README.md +++ b/README.md @@ -207,11 +207,13 @@ if (result == OpenCodeZenUpstreamDiagnostics.Outcome.ABUSE_REJECTED) { // 尊重服务端限制：停止调用，并通过服务商支持渠道处理。 } -// 当上游返回 text/event-stream 时，建议使用 BodyHandlers.ofInputStream() 直接流式处理： -OpenCodeZenSseEventReader.read(response.body(), event -> { +// 以下代码使用专门的流式响应（不是上方的 HttpResponse）： +HttpResponse streamResponse = http.send( + request, HttpResponse.BodyHandlers.ofInputStream()); +OpenCodeZenSseEventReader.read(streamResponse.body(), event -> { if (event.done()) return; // 对每个 JsonNode 做模型协议和工具授权校验，再发给下游。 }); ``` -注意：第二段示例的 `response` 应由 `HttpResponse` 取得，而非上面的 `byte[]` 响应；具体业务中请选择一种 BodyHandler。不得将 `200 /models` 当作推理授权成功证明。 +注意：上述两种 BodyHandler 为演示分类和流式解析的独立示例。实际运行时先检查 HTTP 状态码，403/429 不要进入 SSE 解析。`200 /models` 也不代表推理授权成功。