diff --git a/backend/src/app_mixins/chat.py b/backend/src/app_mixins/chat.py
index 003818a..fc1c107 100644
--- a/backend/src/app_mixins/chat.py
+++ b/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,
diff --git a/frontend/src/backend.js b/frontend/src/backend.js
index 9167e32..fb6bb30 100644
--- a/frontend/src/backend.js
+++ b/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;
diff --git a/frontend/src/components/Message.vue b/frontend/src/components/Message.vue
index 591e992..7975207 100644
--- a/frontend/src/components/Message.vue
+++ b/frontend/src/components/Message.vue
@@ -1,5 +1,7 @@
@@ -43,18 +70,18 @@ function chatEntryClass(msg) {
{{message.sender}}
-
- {{message.content}}
+
+ {{content}}
{{message.sender}}
- {{message.content}}
+ {{content}}
Log ({{message.sender}})
- {{message.content}}
+ {{content}}
diff --git a/frontend/src/stream.js b/frontend/src/stream.js
new file mode 100644
index 0000000..a099375
--- /dev/null
+++ b/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();
+ }
+ }
+}
diff --git a/frontend/src/views/Chat.vue b/frontend/src/views/Chat.vue
index 7111420..f3f19c8 100644
--- a/frontend/src/views/Chat.vue
+++ b/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);