feat: chat stream messages

backend/src/app_mixins/chat.py

@@ -107,7 +107,9 @@ class ChatMixin:
         now = int(datetime.now().timestamp())
 
         if isinstance(content, Stream):
-            initial_content = "..."
+            initial_format = format
+            format = format and "steam-" + format or "stream"
+            initial_content = str(content.id)
         else:
             initial_content = content
 
@@ -169,9 +171,11 @@ class ChatMixin:
             notify_users()
 
             def chunk_fn (chunk):
-                message.content = content.content + "..."
+                pass
             async def end_fn ():
+                message.format = initial_format
                 message.content = content.content
+
                 if db.is_instance(message):
                     await message.save()
                 await notify_agents()
@@ -228,6 +232,7 @@ class ChatMixin:
 
     def notify_message_to_user(self, user_id, message, update=False):
         if user_id in self.user_connections:
+            # TODO: Server's responsibility, and Message.to_json responsibility as well
             user_message = {
                 'action': 'chat-message',
                 'id': message.id,
@@ -235,6 +240,8 @@ class ChatMixin:
                 'time': message.time,
                 'sender': message.sender,
                 'kind': message.kind,
+                'subkind': message.subkind,
+                'format': message.format,
                 'content': message.content,
                 'threadId': message.thread_id,
                 'update': update,

frontend/src/backend.js

@@ -1,5 +1,6 @@
 
 import * as ui from './ui';
+import Stream from './stream';
 
 export class ChatInfo {
   constructor({ id, name }) {
@@ -42,6 +43,16 @@ function createDeferred() {
   return {promise, resolve, reject};
 }
 
+function createDeferredStream(callback) {
+  let resolve, reject;
+  const promise = new Promise((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  let deferred = {promise, resolve, reject};
+  return {promise, resolve (value) { callback(value, deferred) }, reject}
+}
+
 function initState () {
   backendUrl = window.location.origin;
   const tk = localStorage.getItem('loginToken');
@@ -126,12 +137,12 @@ export async function newToken () {
 
 // --- Communication ---
 
-export function sendAction(action, body = {}) {
+export function sendAction(action, body = {}, callback = null) {
   if (socket?.readyState !== WebSocket.OPEN) {
     return Promise.reject(new Error("WebSocket is not open. Cannot send action."));
   }
   const nonce = nextNonce++;
-  const deferred = createDeferred();
+  const deferred = callback ? createDeferredStream(callback) : createDeferred();
 
   pendingActions[nonce] = deferred;
   deferred.promise.finally(() => {
@@ -320,6 +331,19 @@ export async function getAgent(agentId) {
   return allAgents.find(it => it.id == agentId);
 }
 
+export function readStream(streamId) {
+  let stream = new Stream(streamId);
+  sendAction("read-stream", {streamId}, (packet, deferred) => {
+    if (packet.eof) {
+      stream.close();
+      deferred.resolve();
+    } else if (packet.data != null) {
+      stream.feed(packet.data);
+    }
+  });
+  return stream;
+}
+
 export function invalidateCache({ chats=false, agents=false } = {}) {
   if (chats) cache.chats = null;
   if (agents) cache.agents = null;

frontend/src/components/Message.vue

@@ -1,5 +1,7 @@
 <script setup>
+import { ref, watch } from 'vue';
 import { marked } from "marked";
+import * as backend from '../backend';
 marked.use({
   renderer: {
     code (code, infostring, escaped) {
@@ -36,6 +38,31 @@ function chatEntryClass(msg) {
     return ['info'];
   }
 }
+
+const format = ref(null);
+const content = ref("");
+
+let currentStreamId = null;
+
+watch(() => [props.message.format, props.message.content], ([newFormat, newContent]) => {
+  // TODO: if format.startsWith("stream:")
+  if (newFormat == "stream") {
+    format.value = null;
+    currentStreamId = parseInt(newContent);
+
+    let stream = backend.readStream(currentStreamId);
+    stream.on('data', (chunk) => {
+      if (stream.id == currentStreamId) {
+        content.value = stream.partialContent;
+      }
+    });
+  } else {
+    format.value = newFormat;
+    content.value = newContent;
+    currentStreamId = null;
+  }
+}, {immediate: true})
+
 </script>
 
 <template>
@@ -43,18 +70,18 @@ function chatEntryClass(msg) {
     <template v-if="message.type == 'message' || message.type == null">
       <div v-if="message.sender != 'user'" class="sender">{{message.sender}}</div>
 
-      <div v-if="message.format == 'markdown'" v-html="marked.parse(message.content)" class="content"></div>
+      <div v-if="format == 'markdown'" v-html="marked.parse(message.content)" class="content"></div>
-      <div v-else class="content">{{message.content}}</div>
+      <div v-else class="content">{{content}}</div>
     </template>
 
     <template v-if="message.type == 'note'">
       <div class="sender">{{message.sender}}</div>
-      <div class="content">{{message.content}}</div>
+      <div class="content">{{content}}</div>
     </template>
 
     <template v-if="message.type == 'log'">
       <div class="sender">Log ({{message.sender}})</div>
-      <div class="content">{{message.content}}</div>
+      <div class="content">{{content}}</div>
     </template>
 
     <template v-if="message.type == 'thread-header'">

frontend/src/stream.js

@@ -0,0 +1,62 @@
+export default class Stream {
+    constructor (id) {
+        this.id = id;
+        this.partialContent = "";
+        this.eof = false;
+
+        this.listeners = {
+            data: [],
+            end: [],
+        };
+
+        let onResponse = (msg) => {
+            if (this.eof) return;
+
+            if (msg.eof) {
+                this.eof = true;
+                this.handle('end');
+            } else {
+                this.partialContent += msg.data;
+                this.handle('data', msg.data);
+            }
+        };
+
+        this.contentPromise = new Promise((resolve, reject) => {
+            this.handle("end", () => {
+                resolve(this.partialContent);
+            });
+        });
+    }
+
+    feed (chunk) {
+        if (this.eof) return;
+        this.partialContent += chunk;
+        this.handle('data', chunk);
+    }
+
+    close () {
+        this.eof = true;
+        this.handle('end', this.partialContent);
+    }
+
+    getContent () {
+        return this.contentPromise;
+    }
+
+    handle (event, ...args) {
+        for (let fn of this.listeners[event]) {
+            fn(...args)
+        }
+    }
+
+    on (event, callback) {
+        this.listeners[event].push(callback);
+
+        if (event == "data" && this.partialContent) {
+            callback(this.partialContent);
+        }
+        if (event == "end" && this.eof) {
+            callback();
+        }
+    }
+}

frontend/src/views/Chat.vue

@@ -84,6 +84,7 @@
         for (let entry of messages.value) {
           if (entry.id == it.id) {
             entry.type = it.kind;
+            entry.format = it.format;
             entry.content = it.content;
             entry.threadId = it.threadId;
             return;
@@ -108,6 +109,7 @@
       addMessage({
         id: it.id,
         type: it.kind,
+        format: it.format,
         sender: it.sender,
         content: it.content,
         threadId: it.threadId,
@@ -163,7 +165,6 @@
     }
     runListener.addRunId(event.runId);
     messageRunMap.value[event.messageId] = event.runId;
-    ui.alert("Chat Run Message: " + JSON.stringify(event));
   }
   
   function onRunUpdate (newRun) {
@@ -335,7 +336,7 @@
 
   async function startNewThread () {
     const res = await backend.sendAction("create-chat-thread", {chatId});
-    console.log("Starting a nedeleteChatw thread", res);
+    console.log("Starting a new thread", res);
     if (res.threadId != null) {
       currentThreadId.value = res.threadId;
       expandThread(res.threadId, true);