Stream web chat responses token-by-token over WebSocket

AGENTS.md

@@ -131,7 +131,7 @@ Incoming message → ChannelMessageReceivedEvent (channel name, message text)
 - **`ChannelRegistry`**: Registers channels, tracks last-active channel so background task replies are routed correctly.
 - **`DiscordChannel`**: JDA `ListenerAdapter`; accepts DMs from the configured user and guild messages only when the bot is mentioned.
 - **`TelegramChannel`**: `SpringLongPollingBot`; filters by `allowedUsername`; stores `chatId` for routing background replies.
-- **`ChatChannel`**: WebSocket-first delivery (`setWsSession()`/`clearWsSession()`); falls back to buffering replies in `ConcurrentLinkedQueue` exposed via `drainPendingMessages()` REST endpoint when no WebSocket session is active.
+- **`ChatChannel`**: WebSocket-first delivery (`setWsSession()`/`clearWsSession()`); falls back to buffering replies in `ConcurrentLinkedQueue` exposed via `drainPendingMessages()` REST endpoint when no WebSocket session is active. Web chat responses stream live: `Agent.respondTo(conversationId, question, ResponseListener)` uses `ChatClient.stream()` and reports each token via the listener callback (`onToken`/`onComplete`/`onError`); `ChatChannel` supplies a listener that pushes JSON frames (`chunk`, `done`, `error` — see `StreamFrameType`) to the browser over the WebSocket. `Channel` itself knows nothing about streaming; all other channels use the blocking `respondTo`/`sendMessage()` path.
 
 ---
 

app/src/main/java/ai/javaclaw/chat/ChatChannel.java

@@ -1,6 +1,7 @@
 package ai.javaclaw.chat;
 
 import ai.javaclaw.agent.Agent;
+import ai.javaclaw.agent.ResponseListener;
 import ai.javaclaw.channels.Channel;
 import ai.javaclaw.channels.ChannelMessageReceivedEvent;
 import ai.javaclaw.channels.ChannelRegistry;
@@ -16,10 +17,13 @@
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
+import tools.jackson.databind.ObjectMapper;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -38,13 +42,15 @@ public class ChatChannel implements Channel {
     private final Agent agent;
     private final ChannelRegistry channelRegistry;
     private final ChatMemoryRepository chatMemoryRepository;
+    private final ObjectMapper objectMapper;
     private final ConcurrentLinkedQueue<String> pendingMessages = new ConcurrentLinkedQueue<>();
     private final AtomicReference<WebSocketSession> wsSession = new AtomicReference<>();
 
-    public ChatChannel(Agent agent, ChannelRegistry channelRegistry, ChatMemoryRepository chatMemoryRepository) {
+    public ChatChannel(Agent agent, ChannelRegistry channelRegistry, ChatMemoryRepository chatMemoryRepository, ObjectMapper objectMapper) {
         this.agent = agent;
         this.channelRegistry = channelRegistry;
         this.chatMemoryRepository = chatMemoryRepository;
+        this.objectMapper = objectMapper;
         channelRegistry.registerChannel(this);
         log.info("Started Web Chat channel");
     }
@@ -93,6 +99,18 @@ public void sendMessage(String message) {
         }
     }
 
+    /**
+     * Delivers messages buffered while no WebSocket session was active.
+     * Each buffered message is attempted once; a failed push re-buffers it.
+     */
+    public void flushPendingMessages() {
+        for (int i = pendingMessages.size(); i > 0; i--) {
+            String message = pendingMessages.poll();
+            if (message == null) break;
+            sendMessage(message);
+        }
+    }
+
     /**
      * Returns all known conversation IDs, always with "web" first.
      */
@@ -124,10 +142,46 @@ public List<String> loadHistoryAsHtml(String conversationId) {
 
     /**
      * Handles a chat message from the web UI for the given conversationId.
+     * The response is streamed to the WebSocket session as JSON frames
+     * ({@code chunk}/{@code done}/{@code error}); the full response text is returned.
      */
     public String chat(String conversationId, String message) {
         channelRegistry.publishMessageReceivedEvent(new ChannelMessageReceivedEvent(getName(), message));
-        return agent.respondTo(conversationId, message);
+
+        return agent.respondTo(conversationId, message, ResponseListener.of(
+                token -> sendChunkFrame(conversationId, token),
+                () -> sendDoneFrame(conversationId),
+                error -> sendErrorFrame(conversationId, error)));
+    }
+
+    private void sendChunkFrame(String conversationId, String token) {
+        sendFrame(frame(StreamFrameType.CHUNK, conversationId, token));
+    }
+
+    private void sendDoneFrame(String conversationId) {
+        sendFrame(frame(StreamFrameType.DONE, conversationId, null));
+    }
+
+    private void sendErrorFrame(String conversationId, String error) {
+        sendFrame(frame(StreamFrameType.ERROR, conversationId, error == null ? "Unknown error" : error));
+    }
+
+    private static Map<String, Object> frame(StreamFrameType type, String conversationId, Object payload) {
+        Map<String, Object> frame = new LinkedHashMap<>();
+        frame.put("type", type.type());
+        if (payload != null) frame.put("data", payload);
+        frame.put("conversationId", conversationId);
+        return frame;
+    }
+
+    private void sendFrame(Map<String, Object> frame) {
+        WebSocketSession session = wsSession.get();
+        if (session == null || !session.isOpen()) return;
+        try {
+            session.sendMessage(new TextMessage(objectMapper.writeValueAsString(frame)));
+        } catch (IOException e) {
+            log.warn("WS push failed, dropping stream frame: {}", e.getMessage());
+        }
     }
 
     private static String buildBackgroundMessageHtml(String text) {

app/src/main/java/ai/javaclaw/chat/StreamFrameType.java

@@ -0,0 +1,21 @@
+package ai.javaclaw.chat;
+
+/**
+ * Type discriminator of the JSON frames streamed to the web chat WebSocket client.
+ * Every frame carries its payload (if any) in a uniform {@code data} field.
+ */
+public enum StreamFrameType {
+    CHUNK("chunk"),
+    DONE("done"),
+    ERROR("error");
+
+    private final String type;
+
+    StreamFrameType(String type) {
+        this.type = type;
+    }
+
+    public String type() {
+        return type;
+    }
+}

app/src/main/java/ai/javaclaw/chat/ws/ChatWebSocketHandler.java

@@ -45,6 +45,7 @@ public void afterConnectionEstablished(WebSocketSession session) throws Exceptio
                 Htmx.oobInnerHtml("channel-selector", conversationSelector),
                 Htmx.oobInnerHtml("chat-messages", bubbles),
                 Htmx.oobInnerHtml("chat-input-area", inputArea));
+        chatChannel.flushPendingMessages();
     }
 
     @Override
@@ -91,11 +92,10 @@ private void handleUserMessage(Map<String, Object> payload) throws Exception {
                 Htmx.oobReplace("typing-indicator", ChatHtml.typingDots()));
 
         try {
-            // Call agent (blocking — background tasks may push messages via ChatChannel during this)
-            String response = chatChannel.chat(conversationId, userMessage);
-            chatChannel.sendHtml(
-                    Htmx.oobAppend("chat-messages", ChatHtml.agentBubble(response)),
-                    Htmx.oobReplace("typing-indicator", ""));
+            // Call agent (blocking — the response is streamed to the client as JSON frames
+            // by ChatChannel while this call runs; background tasks may push messages too)
+            chatChannel.chat(conversationId, userMessage);
+            chatChannel.sendHtml(Htmx.oobReplace("typing-indicator", ""));
         } catch (RuntimeException ex) {
             log.warn("Chat request failed for conversation {}", conversationId, ex);
             chatChannel.sendHtml(

app/src/main/resources/templates/chat.html.peb

@@ -149,6 +149,11 @@
 .chat-readonly-notice strong {
     color: rgba(180, 195, 235, .7);
 }
+
+.ar-msg--error .ar-msg__bubble {
+    border-color: color-mix(in srgb, var(--bulma-danger) 45%, transparent);
+    color: var(--bulma-danger);
+}
 </style>
 {% endblock %}
 
@@ -259,9 +264,71 @@
         if (ta) { ta.value = ''; ta.style.height = ''; }
     });
     document.body.addEventListener('htmx:wsAfterMessage', function () {
+        scrollToBottom();
+    });
+
+    // --- Streaming frames ------------------------------------------------
+    // The server streams the agent response as JSON frames (type: chunk /
+    // done / error). These are intercepted here and never reach htmx; HTML
+    // frames (no JSON type) keep the htmx OOB-swap behavior unchanged.
+    var streamBubbles = {};   // conversationId -> in-progress bubble element
+
+    document.body.addEventListener('htmx:wsBeforeMessage', function (e) {
+        var frame = parseStreamFrame(e.detail.message);
+        if (!frame) return;
+        e.preventDefault();
+        handleStreamFrame(frame);
+        scrollToBottom();
+    });
+
+    function parseStreamFrame(text) {
+        if (!text || text.charAt(0) !== '{') return null;
+        try {
+            var frame = JSON.parse(text);
+            return frame && typeof frame === 'object' && frame.type ? frame : null;
+        } catch (err) {
+            return null;
+        }
+    }
+
+    function handleStreamFrame(frame) {
+        var id = frame.conversationId || 'web';
+        if (frame.type === 'chunk') {
+            streamBubble(id).textContent += frame.data;
+        } else if (frame.type === 'done') {
+            clearTypingIndicator();
+            delete streamBubbles[id];
+        } else if (frame.type === 'error') {
+            var bubble = streamBubble(id);
+            bubble.textContent = frame.data;
+            bubble.closest('.ar-msg').classList.add('ar-msg--error');
+            delete streamBubbles[id];
+        }
+    }
+
+    function streamBubble(conversationId) {
+        var bubble = streamBubbles[conversationId];
+        if (!bubble || !bubble.isConnected) {
+            var article = document.createElement('article');
+            article.className = 'ar-msg ar-msg--agent';
+            article.innerHTML = '<div class="ar-msg__avatar">JC</div><div class="ar-msg__bubble"></div>';
+            document.getElementById('chat-messages').appendChild(article);
+            bubble = article.querySelector('.ar-msg__bubble');
+            streamBubbles[conversationId] = bubble;
+            clearTypingIndicator();
+        }
+        return bubble;
+    }
+
+    function clearTypingIndicator() {
+        var typing = document.getElementById('typing-indicator');
+        if (typing) typing.innerHTML = '';
+    }
+
+    function scrollToBottom() {
         var body = document.querySelector('.chat-body');
         if (body) body.scrollTop = body.scrollHeight;
-    });
+    }
 }());
 </script>
 {% endblock %}