refactor: align streaming objects with OpenClaw

Author: loong10k

src/main/java/io/github/easy4j/opencode/OpenCodeClient.java

@@ -127,7 +127,7 @@ public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, OpenCodeCliConfig cli
             this.chatClient = null;
             this.sseClient = null;
         }
-        this.streamExecutor = this.chatClient == null ? createStreamExecutor(httpConfig) : null;
+        this.streamExecutor = createStreamExecutor(httpConfig);
 
         // CLI 子系统初始化
         if (cliEnabled) {
@@ -166,7 +166,7 @@ public OpenCodeClient(OpenCodeClientConfig config,
         this.chatClient = httpClient instanceof OpenCodeChatClient ? (OpenCodeChatClient) httpClient : null;
         this.sseClient = sseClient;
         this.cli = cli;
-        this.streamExecutor = this.chatClient == null ? createStreamExecutor(config.getHttp()) : null;
+        this.streamExecutor = createStreamExecutor(config.getHttp());
     }
 
     private static ExecutorService createStreamExecutor(OpenCodeHttpClientConfig config) {
@@ -382,9 +382,6 @@ public ChatStreamingResponse chatCompletionStream(ChatRequest request, String se
     public ChatStreamingResponse chatCompletionStream(ChatRequest request, String sessionKey,
                                                        OpenCodeRequestContext context,
                                                        Consumer<String> deltaConsumer) {
-        if (chatClient != null) {
-            return chatClient.chatCompletionStream(request, sessionKey, context, deltaConsumer);
-        }
         String sessionId = httpClient.ensureSession(sessionKey, context);
         PromptRequest promptRequest = ChatMessageMapper.toPromptRequest(request);
 

src/main/java/io/github/easy4j/opencode/api/OpenCodeChatClient.java

@@ -6,7 +6,7 @@
 import io.github.easy4j.opencode.api.mapper.ChatMessageMapper;
 import io.github.easy4j.opencode.api.model.ChatRequest;
 import io.github.easy4j.opencode.api.model.ChatResponse;
-import io.github.easy4j.opencode.api.model.ChatStreamingResponse;
+import io.github.easy4j.opencode.api.sse.StreamingChatResponse;
 import io.github.easy4j.opencode.api.model.Event;
 import io.github.easy4j.opencode.api.model.PromptRequest;
 import io.github.easy4j.opencode.api.model.PromptResult;
@@ -78,22 +78,22 @@ public ChatResponse chatCompletionWithSession(ChatRequest request, String sessio
         return ChatMessageMapper.toChatResponse(result);
     }
 
-    public ChatStreamingResponse chatCompletionStream(ChatRequest request, String sessionKey) {
+    public StreamingChatResponse chatCompletionStream(ChatRequest request, String sessionKey) {
         return chatCompletionStream(request, sessionKey, null, null);
     }
 
-    public ChatStreamingResponse chatCompletionStream(ChatRequest request, String sessionKey,
+    public StreamingChatResponse chatCompletionStream(ChatRequest request, String sessionKey,
                                                        OpenCodeRequestContext context) {
         return chatCompletionStream(request, sessionKey, context, null);
     }
 
     /** 在事件订阅启动前绑定增量回调，避免丢失首批分片。 */
-    public ChatStreamingResponse chatCompletionStream(ChatRequest request, String sessionKey,
+    public StreamingChatResponse chatCompletionStream(ChatRequest request, String sessionKey,
                                                        OpenCodeRequestContext context,
                                                        Consumer<String> deltaConsumer) {
         String sessionId = ensureSession(sessionKey, context);
         PromptRequest promptRequest = ChatMessageMapper.toPromptRequest(request);
-        ChatStreamingResponse stream = new ChatStreamingResponse().onDelta(deltaConsumer);
+        StreamingChatResponse stream = new StreamingChatResponse().onDelta(deltaConsumer);
         OpenCodeSseClient.QueueSubscription subscription = eventClient.subscribeQueueSubscription(context);
         BlockingQueue<Event> queue = subscription.getQueue();
 
@@ -119,7 +119,7 @@ public ChatStreamingResponse chatCompletionStream(ChatRequest request, String se
 
     private void consumeEvents(String sessionId, BlockingQueue<Event> queue,
                                OpenCodeSseClient.QueueSubscription subscription,
-                               ChatStreamingResponse stream) {
+                               StreamingChatResponse stream) {
         try {
             long timeoutMillis = Math.max(1L, config.getReadTimeoutMillis());
             long deadline = System.currentTimeMillis() + timeoutMillis;

src/main/java/io/github/easy4j/opencode/api/model/ChatStreamingResponse.java

@@ -1,7 +1,7 @@
 package io.github.easy4j.opencode.api.model;
 
-import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Streaming chat response for OpenCode (aligned with Hermes {@code ChatStreamingResponse}).
@@ -14,49 +14,14 @@
  * @see ChatRequest
  * @see ChatResponse
  */
-public class ChatStreamingResponse extends CompletableFuture<String> {
-
-    private final StringBuilder content = new StringBuilder();
-    private Consumer<String> deltaConsumer;
+@Deprecated
+public class ChatStreamingResponse extends io.github.easy4j.opencode.api.sse.StreamingChatResponse {
 
     /**
      * 注册增量文本回调，每收到一段 delta 触发一次。
      */
     public ChatStreamingResponse onDelta(Consumer<String> consumer) {
-        this.deltaConsumer = consumer;
+        super.onDelta(consumer);
         return this;
     }
-
-    /**
-     * 接收一段增量文本。
-     */
-    public void acceptDelta(String delta) {
-        if (delta != null && !delta.isEmpty()) {
-            content.append(delta);
-            if (deltaConsumer != null) {
-                deltaConsumer.accept(delta);
-            }
-        }
-    }
-
-    /**
-     * 流结束，完成 future。
-     */
-    public void finish() {
-        complete(content.toString());
-    }
-
-    /**
-     * 流异常，异常完成 future。
-     */
-    public void fail(Throwable error) {
-        completeExceptionally(error);
-    }
-
-    /**
-     * 获取已累积的完整文本。
-     */
-    public String getAccumulatedContent() {
-        return content.toString();
-    }
 }

src/main/java/io/github/easy4j/opencode/api/sse/StreamingChatResponse.java

@@ -0,0 +1,39 @@
+package io.github.easy4j.opencode.api.sse;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+/**
+ * OpenCode 流式聊天响应，逐段回调并在结束时返回累积全文。
+ */
+public class StreamingChatResponse extends CompletableFuture<String> {
+
+    private final StringBuilder content = new StringBuilder();
+    private Consumer<String> deltaConsumer;
+
+    public StreamingChatResponse onDelta(Consumer<String> consumer) {
+        this.deltaConsumer = consumer;
+        return this;
+    }
+
+    public void acceptDelta(String delta) {
+        if (delta != null && !delta.isEmpty()) {
+            content.append(delta);
+            if (deltaConsumer != null) {
+                deltaConsumer.accept(delta);
+            }
+        }
+    }
+
+    public void finish() {
+        complete(content.toString());
+    }
+
+    public void fail(Throwable error) {
+        completeExceptionally(error);
+    }
+
+    public String getAccumulatedContent() {
+        return content.toString();
+    }
+}

src/test/java/io/github/easy4j/opencode/OpenCodeChatClientArchitectureTest.java

@@ -2,6 +2,7 @@
 
 import io.github.easy4j.opencode.api.OpenCodeChatClient;
 import io.github.easy4j.opencode.api.OpenCodeHttpClient;
+import io.github.easy4j.opencode.api.sse.StreamingChatResponse;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -19,4 +20,12 @@ void shouldExposeUnifiedConfigAndChatScenarioClient() {
             assertEquals(OpenCodeHttpClient.class, client.getClass().getSuperclass());
         }
     }
+
+    @Test
+    void shouldExposeStreamingResponseUnderSsePackage() {
+        StreamingChatResponse response = new StreamingChatResponse();
+        response.acceptDelta("hello");
+        response.finish();
+        assertEquals("hello", response.join());
+    }
 }