chat-frontend: redesign chat entry computation using thread objects

frontend/src/chat.js

@@ -22,6 +22,38 @@ function isMessageStale (message) {
   return time == null || (Date.now() - time) > THREAD_STALE_MS;
 }
 
+function earliestMessage (...messages) {
+  let minId = null;
+  let minTime = null;
+  let useTime = true;
+
+  for (let msg of messages) {
+    if (!msg) continue;
+
+    if (!minId || msg.id < minId.id) {
+      minId = msg;
+    }
+    let msgTime = messageTime(msg);
+    if (msgTime == null) {
+      useTime = false;
+    } else if (!minTime || msgTime < messageTime(minTime)) {
+      minTime = msg;
+    }
+  }
+
+  return useTime ? minTime : minId;
+}
+
+export class Thread {
+  constructor(obj) {
+    Object.assign(this, {messages: []}, obj || {});
+  }
+
+  getLastMessage () {
+    return this.messages[this.messages.length - 1];
+  }
+}
+
 
 // === Chat Session === //
 
@@ -29,27 +61,29 @@ function isMessageStale (message) {
 // The view creates a session, hooks onRescroll() to its message container,
 // and drives it with load()/unload(). Shared state is exposed as refs.
 export class ChatSession {
+  // TODO: This class should not be reused in different chats,
+  // a chat session is created for a single chat id and never changes,
+  // it only ever gets disposed of
   constructor () {
     // Chat state
     this.chat = ref({});
     this.messages = ref([]);
     this.unsentMessages = ref([]);
-    this.currentThreadId = ref(null);
-    this.isNewThread = ref(true); // "New Thread" UI state
     this.messageRunMap = ref({}); // messageId -> runId of the run it started
 
-    // Editing state
+    // Input mode and target
+    this.currentThreadId = ref(null);
+    this.creatingThread = ref(true); // "New Thread" UI state
     this.editingMessageId = ref(null);
     this.noteMode = ref(false);
 
-    this.chatEntries = computed(() => {
-      return this.computeChatEntries(
-        this.messages.value.concat(this.unsentMessages.value)
-      );
-    });
+    // this must be recomputed everytime the displayed entries update
+    this.chatEntries = ref([]);
+    // real chat data
+    this.threads = [];
 
     this.editInfo = computed(() => {
-      const parts = [this.isNewThread.value ? 'New Thread' : `Thread ${this.currentThreadId.value}`];
+      const parts = [this.creatingThread.value ? 'New Thread' : `Thread ${this.currentThreadId.value}`];
       if (this.editingMessageId.value != null) {
         if (this.noteMode.value) {
           parts.push(`Editing Note ${this.editingMessageId.value}`);
@@ -98,13 +132,16 @@ export class ChatSession {
     this.unsentMessages.value = [];
     this.messageRunMap.value = {};
     this.currentThreadId.value = null;
-    this.isNewThread.value = true;
+    this.creatingThread.value = true;
     this.editingMessageId.value = null;
     this.noteMode.value = false;
 
+    this.chatEntries.value = [];
+    this.threads = [];
+
     // Create an agent name dictionary
     const agents = await backend.getAgents();
-    if (this.chatId != chatId) return; // A newer load or unload has won
+
     this.agentNames = {};
     for (let agent of agents) {
       this.agentNames[agent.id] = agent.name;
@@ -119,29 +156,22 @@ export class ChatSession {
     this._unsubscribeChat = backend.onChatMessages(chatId, it => this.receiveMessage(it));
     this._unsubscribeRuns = runs.onChatMessage(chatId, event => this.trackRun(event));
 
-    // Don't load all messages, only thread headers
-    //const response = await backend.sendAction("get-chat-messages", {chatId});
-    const response = await backend.sendAction("get-chat-thread-heads", {chatId});
-    if (this.chatId != chatId) return;
+    // Load threads
+    const {messages: heads} = await backend.sendAction("get-chat-thread-heads", {chatId});
 
     // Always good to duplicate when assigning to a ref
-    this.messages.value = response.messages.map(it => ({
-      id: this._messageNonce++,
-      type: it.kind,
-      subkind: it.subkind,
-      sender: it.sender,
-      content: it.content,
-      threadId: it.threadId,
-    }));
+    for (const it of heads) {
+      this.getThread(it.threadId).messages.push(it);
+    }
 
     // Continue the last thread only if it is recent (under 10 minutes old);
     // otherwise start in "New Thread" state
-    const lastMessage = response.messages[response.messages.length - 1];
+    const lastMessage = heads[heads.length - 1];
     if (lastMessage && lastMessage.threadId != null && !isMessageStale(lastMessage)) {
       this.activateThread(lastMessage.threadId);
     }
 
-    // Must happen on the same async-thread that sets `messages`
+    this.computeChatEntries();
     this.rescroll({forceBottom: true});
   }
 
@@ -162,14 +192,18 @@ export class ChatSession {
   receiveMessage (it) {
     if (it.update) {
       // Find message in list. If not found, add it
-      for (let entry of this.messages.value) {
-        if (entry.id == it.id) {
-          entry.type = it.kind;
-          entry.subkind = it.subkind;
-          entry.format = it.format;
-          entry.content = it.content;
-          entry.threadId = it.threadId;
-          return;
+      for (let thread of this.threads) {
+        for (let entry of thread.messages) {
+          if (entry.id == it.id) {
+            entry.type = it.kind;
+            entry.subkind = it.subkind;
+            entry.format = it.format;
+            entry.content = it.content;
+            entry.threadId = it.threadId;
+
+            this.computeChatEntries();
+            return;
+          }
         }
       }
     }
@@ -189,6 +223,7 @@ export class ChatSession {
       }
     }
 
+    // addMessageBulk I think is more correct
     this.addMessage({
       id: it.id,
       type: it.kind,
@@ -198,6 +233,9 @@ export class ChatSession {
       content: it.content,
       threadId: it.threadId,
     });
+
+    this.computeChatEntries();
+    this.rescroll();
   }
 
 
@@ -238,7 +276,8 @@ export class ChatSession {
       message.id = this._messageNonce++;
       this.unsentMessages.value.push(message);
     } else {
-      this.messages.value.push(message);
+      let messages = this.getThread(message.threadId).messages;
+      messages.push(message);
     }
   }
 
@@ -250,12 +289,10 @@ export class ChatSession {
     if (kind == null && this.noteMode.value) kind = 'note';
 
     const messageId = this.editingMessageId.value;
+    const threadId = await this.resolveTargetThread();
 
-    if (messageId == null) {
-      // "New Thread" is a UI state: the thread is only created upstream
-      // here, once there is a message to put into it
-      const threadId = await this.resolveSendThread();
-      if (threadId == null) return false;
+    if (messageId == null || this.creatingThread.value) {
+      // New message in target thread (maybe fork)
 
       backend.sendAction("send-chat-message", {
         chatId: this.chatId, content, threadId, kind, subkind,
@@ -272,8 +309,9 @@ export class ChatSession {
         content,
         threadId,
       }, {unsent: true});
+
     } else {
-      const messageRef = this.messages.value.find(it => it.id == messageId);
+      // Editing existing message
 
       backend.sendAction("update-chat-message", {
         messageId, content, kind,
@@ -283,19 +321,30 @@ export class ChatSession {
         }
       });
 
-      messageRef.type = kind || 'message';
-      messageRef.content = content;
+      // No update event received, local manual update
+      let thread = await this.getThread(threadId);
+      let existing = thread.messages.find(msg => msg.id == messageId);
+      if (existing) {
+        existing.content = content;
+        existing.type = kind || 'message';
+      } else {
+        console.error("Can't edit locally?", {messageId, threadId, t_m: thread.messages});
+      }
+
       this.editingMessageId.value = null;
     }
 
     if (kind == 'note') this.noteMode.value = false;
+
+    this.computeChatEntries();
+    this.rescroll();
     return true;
   }
 
   // Returns the threadId to send into, creating the thread upstream
   // if we're in "New Thread" state
-  async resolveSendThread () {
-    if (!this.isNewThread.value) return this.currentThreadId.value;
+  async resolveTargetThread () {
+    if (!this.creatingThread.value) return this.currentThreadId.value;
 
     const res = await backend.sendAction("create-chat-thread", {chatId: this.chatId});
     if (res.threadId == null) {
@@ -303,7 +352,7 @@ export class ChatSession {
       return null;
     }
     this.currentThreadId.value = res.threadId;
-    this.isNewThread.value = false;
+    this.creatingThread.value = false;
     return res.threadId;
   }
 
@@ -312,14 +361,86 @@ export class ChatSession {
     this.editingMessageId.value = messageId ?? null;
   }
 
+  // Adds a bulk of messages to the message list, in their correct
+  // chronological position and replacing duplicates
+  addMessageBulk (messages) {
+    while (messages.length) {
+      console.log("bulk iteration", messages.length);
+
+      let thread = this.getThread(messages[0].threadId);
+
+      let skipped = [];
+      for (let msg of messages) {
+        if (msg.threadId != thread.id) {
+          skipped.push(msg);
+          continue;
+        }
+
+        /// rename for the algorithm
+        let id = msg.id;
+        let list = thread.messages;
+
+        // Replace an already-known message in place
+        let existingIndex = list.findIndex(it => it.id == id);
+        if (existingIndex >= 0) {
+          list[existingIndex] = msg;
+          continue;
+        }
+
+        // An unsent user message echoed back with an id: adopt it
+        if (msg.sender == 'user') {
+          let unsentIndex = this.unsentMessages.value.findIndex(
+            it => it.content == msg.content
+          );
+          if (unsentIndex >= 0) {
+            this.unsentMessages.value.splice(unsentIndex, 1);
+          }
+        }
+
+        // Insert chronologically
+        let msgTime = messageTime(msg);
+        let insertIndex = list.length;
+        while (insertIndex > 0) {
+          // from latest to earliest, skip all "later" messages
+          let elem = list[insertIndex-1];
+          if (messageTime(elem) > msgTime)
+            insertIndex--;
+          else break;
+        }
+        list.splice(insertIndex, 0, msg);
+      }
+
+      if (skipped.length && skipped.length != messages.length)
+        messages = skipped;
+      else break;
+    }
+
+    this.computeChatEntries();
+  }
+
 
   // === Threads === //
 
+  getThread (threadId) {
+    // `null` (orphan messages) serializes to the "null" key
+    let thread = this.threads.find(it => it.id ==threadId);
+    if (!thread) {
+      thread = new Thread({
+        id: threadId,
+        loaded: false,
+        time: null, // TODO
+        messages: [],
+      });
+      this.threads.push(thread);
+    }
+    return thread;
+  }
+
   activateThread (threadId) {
     if (threadId == null) return; // "New Thread" placeholder isn't clickable
     // Continuing a thread: it becomes the one open thread
     this.currentThreadId.value = threadId;
-    this.isNewThread.value = false;
+    this.creatingThread.value = false;
 
     // Load thread messages asynchronously
     backend.sendAction("get-thread-messages", {
@@ -335,59 +456,20 @@ export class ChatSession {
         threadId: it.threadId,
       })));
     });
+
+    this.computeChatEntries();
     this.rescroll();
   }
 
   startNewThread () {
     // Pure UI state — nothing is created upstream until a message is sent
-    this.isNewThread.value = true;
+    this.creatingThread.value = true;
     this.currentThreadId.value = null;
+
+    this.computeChatEntries();
     this.rescroll();
   }
 
-  // Adds a bulk of messages to the message list, in their correct
-  // chronological position and replacing duplicates
-  addMessageBulk (messages) {
-    const list = this.messages.value;
-
-    for (let message of messages) {
-      const id = message.id;
-
-      // Replace an already-known message in place
-      const existingIndex = list.findIndex(it => it.id == id);
-      if (existingIndex >= 0) {
-        list[existingIndex] = message;
-        continue;
-      }
-
-      // An unsent user message echoed back with an id: adopt it
-      if (message.sender == 'user') {
-        const unsentIndex = this.unsentMessages.value.findIndex(
-          it => it.content == message.content
-        );
-        if (unsentIndex >= 0) {
-          this.unsentMessages.value.splice(unsentIndex, 1);
-        }
-      }
-
-      // Insert after the last message known to belong to the same thread,
-      // keeping each thread's messages contiguous and ordered
-      let insertIndex = -1;
-      for (let i = list.length - 1; i >= 0; i--) {
-        if (list[i].threadId != null && list[i].threadId == message.threadId) {
-          insertIndex = i + 1;
-          break;
-        }
-      }
-
-      if (insertIndex < 0) insertIndex = list.length;
-      list.splice(insertIndex, 0, message);
-    }
-
-    // Refs need reassignment to notify watchers; duplicate the list
-    this.messages.value = list.slice();
-  }
-
 
   // === Entry Computation === //
 
@@ -395,8 +477,7 @@ export class ChatSession {
   // other thread collapsed into a single preview header
   computeChatEntries (messages) {
     let entries = [];
-    let threadGroups = {}; // Map<int, message[]>
-
+    
     const senderName = (sender) => {
       let isInt = !isNaN(parseInt(sender));
       if (isInt) {
@@ -406,66 +487,88 @@ export class ChatSession {
       }
     };
 
-    const addThreadHeader = (threadId, group, options={}) => {
+    const getThreadHeader = (thread, last, options={}) => {
       let preview = null;
-      const last = group[group.length - 1];
       if (last && !options.active) {
         let content = String(last.content || '').split('\n')[0].trim();
         if (content.length > 100) content = content.slice(0, 100) + '…';
         preview = {sender: senderName(last.sender), content};
       }
-      entries.push({
+      return {
         type: 'thread-header',
-        threadId,
+        threadId: thread.id,
         active: !!options.active,
-        messageCount: group.length,
+        messageCount: thread.messages.length,
         preview,
-      });
+      };
     };
 
-    // First pass: Group messages by thread
-    for (let message of messages) {
-      if (message.threadId != null) {
-        let group = threadGroups[message.threadId] ||
-          (threadGroups[message.threadId] = []);
-        group.push(message);
-      }
-    }
+    const activeThreadMessages = this.currentThreadId.value == null ?
+      [] : this.getThread(this.currentThreadId.value).messages;
 
-    // Only the active thread is ever open; every other thread collapses
-    // into a single preview header
-    const activeThreadId = this.isNewThread.value ? null : this.currentThreadId.value;
-    let activeHeaderAdded = false;
+    let activeIndex = 0;
+    let threadIndex = 0;
 
-    // Second pass: Build the display list with correct ordering
-    for (let _msg of messages) {
-      const message = Object.assign({}, _msg, {sender: senderName(_msg.sender)});
+    while (true) {
+      let active = activeThreadMessages[activeIndex];
+      let thread = this.threads[threadIndex];
 
-      if (message.threadId == null) {
-        entries.push(message);
-        continue;
+      if (!active && !thread) break;
+
+      // Skip active thread
+      if (thread && thread.id != null && thread.id == this.currentThreadId.value) {
+        thread = null;
+        threadIndex++;
+
+        if (!active) continue;
       }
 
-      let threadId = message.threadId;
-      let group = threadGroups[threadId];
+      let threadLast = thread ? thread.getLastMessage() : null;
+      let entry = earliestMessage(active, threadLast);
 
-      if (threadId === activeThreadId) {
-        if (!activeHeaderAdded) {
-          addThreadHeader(threadId, group, {active: true});
-          activeHeaderAdded = true;
+      // None is earliest?
+      if (!entry) break;
+
+      if (entry == active) {
+        // Add thread header before first thread message
+        if (activeIndex == 0) {
+          entries.push({
+            type: 'thread-header',
+            threadId: entry.threadId,
+            active: true,
+            messageCount: activeThreadMessages.length,
+            preview: null,
+          });
         }
-        entries.push(message);
-      } else if (message.id == group[group.length - 1].id) {
-        // Collapsed: one preview card in place of the thread's last message
-        addThreadHeader(threadId, group);
+        activeIndex++;
+      } else if (entry == threadLast) {
+        // Replace with header
+        entry = getThreadHeader(thread, threadLast);
+        threadIndex++;
       }
+
+      // never halt
+      else break;
+
+      // Update sender name
+      entry = Object.assign(
+        {type: entry.kind || "message"}, // default
+        entry, // value
+        {sender: senderName(entry.sender)} // override
+      );
+      entries.push(entry);
     }
 
-    // "New Thread" UI state: placeholder where the new thread will start
-    if (this.isNewThread.value) {
-      addThreadHeader(null, []);
+    if (this.creatingThread.value) {
+      entries.push({
+        type: 'thread-header',
+        threadId: null,
+        active: true,
+        messageCount: 0,
+        preview: null,
+      });
     }
 
-    return entries;
+    this.chatEntries.value = entries;
   }
 }

frontend/src/views/Chat.vue

@@ -80,10 +80,10 @@
     await session.sendMessage(selected, {kind: 'resource', subkind: 'remote'});
   }
 
-  function editMessage (message) {
+  function editMessage (message, fork) {
     let elem = inputFieldElem.value;
     elem.innerText = message ? message.content : "";
-    session.setEditing(message ? message.id : null);
+    session.setEditing(message ? message.id : null, fork);
   }
 
   async function deleteChat (chat) {