共享库 1.3.2：V2.4 SSE 解析与 403 拒绝诊断（不修改客户端身份）

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<InputStream> 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,31 @@ HttpResponse<InputStream> 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<byte[]> response = http.send(request, HttpResponse.BodyHandlers.ofByteArray());
+var result = OpenCodeZenUpstreamDiagnostics.classify(response.statusCode(), response.body());
+if (result == OpenCodeZenUpstreamDiagnostics.Outcome.ABUSE_REJECTED) {
+    // 尊重服务端限制：停止调用，并通过服务商支持渠道处理。
+}
+
+// 以下代码使用专门的流式响应（不是上方的 HttpResponse<byte[]>）：
+HttpResponse<InputStream> streamResponse = http.send(
+    request, HttpResponse.BodyHandlers.ofInputStream());
+OpenCodeZenSseEventReader.read(streamResponse.body(), event -> {
+    if (event.done()) return;
+    // 对每个 JsonNode 做模型协议和工具授权校验，再发给下游。
+});
+```
+
+注意：上述两种 BodyHandler 为演示分类和流式解析的独立示例。实际运行时先检查 HTTP 状态码，403/429 不要进入 SSE 解析。`200 /models` 也不代表推理授权成功。

build.gradle.kts

@@ -3,7 +3,7 @@ plugins {
 }
 
 group = "com.sharedlibs"
-version = "1.3.1"
+version = "1.3.2"
 
 subprojects {
     group = rootProject.group

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<Event> 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<Event> 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;
+    }
+}

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;
+    }
+}

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<OpenCodeZenSseEventReader.Event> 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<OpenCodeZenSseEventReader.Event> 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"));
+    }
+}