feat: connect chat and runs, cancellable chat messages, unified frontend runs module

backend/src/agent.py

@@ -75,14 +75,20 @@ class Agent:
                 "thread_id": message.thread_id,
             },
         )
+
         await self.run(run)
-        await self.app.send_chat_message(
+        new_message = await self.app.send_chat_message(
-            chat_id=message.chat_id,
             sender=self.id,
-            content=run.stream,
+            chat_id=message.chat_id,
             thread_id=message.thread_id,
+            content=run.stream,
         )
-        return run.stream;
+        run.emit_event('chat-message', {
+            'messageId': new_message['id'],
+            'chatId': message.chat_id,
+            'threadId': message.thread_id,
+        })
+        return run.stream
 
     def _role_for_message(self, message):
         sender = message.sender
@@ -140,18 +146,17 @@ class AgentRun:
         
         # Cancellation tracking
         self.task = None  # For asyncio tasks
-        self.thread = None  # For threads
         self.cancelled = False
 
         app.tmp.runs[self.id] = self
         
         # Emit run start event
-        app.emit_event("run-started", {
+        self.emit_event('created', {
-            "runId": self.id,
             "agentId": self.agent_id,
             "userId": self.user_id,
             "startTime": self.start_time,
-            "input": self.input
+            "input": self.input,
+            "streamId": self.stream.id,
         })
         
         return self
@@ -259,12 +264,3 @@ class AgentRun:
         self.output = self.stream.content
         self.stream.close()
         self.emit_event("done", {"output": self.output, "agent_id": self.agent_id})
-        
-        # Broadcast run completion event
-        self.app.emit_event("run-completed", {
-            "runId": self.id,
-            "agentId": self.agent_id,
-            "endTime": self.end_time,
-            "output": self.output
-        })
-        

backend/src/server.py

@@ -141,7 +141,7 @@ async def start_server(app):
             
             # Listen for events and broadcast them
             async def on_event(name, data):
-                await send({"action": "event", "name": name, "data": data})
+                await send({"action": name, "data": data})
             app.on_event(on_event)
 
             while True:
@@ -153,12 +153,6 @@ async def start_server(app):
                 body = req.get('body', {})
                 nonce = req.get('nonce', None)
 
-                def get_agent_id ():
-                    raise Exception("Cannot get agent id")
-
-                def get_chat_id ():
-                    raise Exception("Cannot get chat id")
-
                 if action == "ping":
                     await send({'message': "pong"})
 

frontend/src/backend.js

@@ -29,8 +29,9 @@ let credentialsDeferred = createDeferred();
 // Handlers
 const pendingActions = {}; // Maps nonce to { resolve, reject }
 const chatListeners = {}; // Maps chatId to an array of callback functions
+const actionListeners = {}; // Listeners for other arbitrary actions
 const orphanHandlers = []; // Queue of callbacks for orphan server messages
-const runListeners = []; // Queue of callbacks for run events
+const connectListeners = []; // Listeners for horde connection
 
 function createDeferred() {
   let resolve, reject;
@@ -164,8 +165,6 @@ function handleIncomingMessage(rawData) {
     return;
   }
 
-  console.log("Incoming Packet:", packet);
-
   if (packet.nonce != null) {
     const nonce = packet.nonce;
     const actionPromise = pendingActions[nonce];
@@ -179,6 +178,7 @@ function handleIncomingMessage(rawData) {
       console.warn(`Orphan response for action nonce ${nonce}. No pending action found.`);
     }
   } else if (packet.action === 'chat-message') {
+    // This can be in a 'chat-message' listener in the future
     const chatId = packet.chatId;
     const listeners = chatListeners[chatId];
     if (listeners) {
@@ -192,22 +192,20 @@ function handleIncomingMessage(rawData) {
     } else {
       console.log(`No active listeners for chatId ${chatId}`);
     }
-  } else if (packet.action == 'run-event') {
+  } else if (packet.action) {
-    // Broadcast run events to all listeners
+    let listeners = actionListeners[packet.action];
-    runListeners.forEach(callback => {
+    if (listeners && listeners.length) {
-      try {
+      for (let f of listeners) {
-        callback(packet);
+        f(packet);
-      } catch (e) {
-        console.error(`Error in run event listener:`, e);
       }
-    });
+    }
-  } else {
+  } else if (orphanHandlers.length) {
-    if (orphanHandlers.length) {
     const f = orphanHandlers.shift();
     f(packet);
   } else if (packet.error) {
     console.error(`Backend Error: ${packet.error}`);
-    }
+  } else {
+    console.warn('ignored server message', packet);
   }
 }
 
@@ -257,18 +255,23 @@ export function onChatMessages(chatId, callback) {
   };
 }
 
-export function onRunEvents(callback) {
+export function onAction(action, callback) {
   if (typeof callback !== 'function') {
-    throw new TypeError("onRunEvents: callback must be a function.");
+    throw new TypeError("onAction: callback must be a function.");
   }
 
-  runListeners.push(callback);
+  let listeners = actionListeners[action];
+  if (!listeners) {
+    listeners = [];
+    actionListeners[action] = listeners;
+  }
+  listeners.push(callback);
 
   // Return an unsubscribe function
   return () => {
-    const index = runListeners.indexOf(callback);
+    const index = listeners.indexOf(callback);
     if (index > -1) {
-      runListeners.splice(index, 1);
+      listeners.splice(index, 1);
     }
   };
 }
@@ -326,6 +329,11 @@ export function invalidateCache({ chats=false, agents=false } = {}) {
 
 // --- Connection ---
 
+export function onConnect (callback) {
+  // TODO: return cancel function
+  connectListeners.push(callback);
+}
+
 function reconnect() {
   if (reconnectTimeout) {
     clearTimeout(reconnectTimeout);
@@ -361,6 +369,8 @@ function connect() {
         // Successful connection, store token long term
         authDeferred.resolve(true);
         localStorage.setItem('loginToken', loginToken);
+
+        for (let cb of connectListeners) cb();
       } else {
         authDeferred.resolve(false);
         console.error("Socket authentication failed:", response ? response.error : "Unknown error");

frontend/src/components/Message.vue

@@ -18,6 +18,10 @@ const props = defineProps({
   message: {
     type: Object,
     required: true
+  },
+  cancellable: {
+    type: Boolean,
+    required: false,
   }
 });
 
@@ -59,6 +63,8 @@ function chatEntryClass(msg) {
         <a href="#" @click.prevent="$emit('toggleThread', message.threadId, !message.expanded)">{{message.expanded ? "Hide" : "Show"}}</a>
       </div>
     </template>
+
+    <button v-if="cancellable" @click="$emit('cancel')">Cancel</button>
   </div>
 </template>
 

frontend/src/runs.js

@@ -0,0 +1,142 @@
+import * as backend from './backend';
+
+// State
+let runs = [];
+let unsubscribeRunEvents = null;
+
+// Event listeners for UI components
+const listeners = {
+  runs: [],
+  chats: {},
+};
+
+class RunListener {
+  constructor() {  
+    this.runIds = new Set();   
+    this.callbacks = [];
+    this.cancelled = false; 
+  }
+
+  setIdList(runIds) {   
+    this.runIds = new Set(runIds);  
+  }
+
+  addRunId(runId) {   
+    this.runIds.add(runId);
+  }
+
+  addCallback(callback) {
+    this.callbacks.push(callback);
+  }
+
+  cancel() {
+    this.cancelled = true;
+    listeners.runs = listeners.runs.filter(l => l !== this);
+  }    
+  
+  hasRunId(runId) {
+    return this.runIds.has(runId);
+  }
+
+  emitChange(data) {
+    if (this.cancelled) return;
+    for (let cb of this.callbacks) cb(data);
+  }
+}
+
+// Update a run based on an incoming event
+function updateRun(event) {
+  const { name, runId, data } = event.data;
+
+  // Find the run by ID
+  const runIndex = runs.findIndex(r => r.id === runId);
+  let run = runIndex >= 0 ? runs[runIndex] : null;
+
+  if (name == 'created') {
+    if (!run) {
+      run = Object.assign({}, data, {id: runId, events: []});
+      runs.push(run);
+    } else {
+      console.warn("Created but existing?", run, data);
+    }
+  } else if (name == 'done') {
+    run.done = true;
+    run.endTime = data.endTime;
+    run.output = data.output;
+  } else if (name == 'chat-message') {
+    let chatListeners = listeners.chats[data.chatId] || [];
+    for (let f of chatListeners) {
+      f({runId: runId, ...data});
+    }
+  }
+
+  if (run) {
+    run.events.push(event);
+    for (let listener of listeners.runs) {
+      if (listener.hasRunId(runId)) {
+        listener.emitChange(run);
+      }
+    }
+  } else {
+    console.error('Bad run event (non-existing run)', event);
+  }
+}
+
+// Initialize the runs module
+async function init() {
+  // Listen for connection establishment
+  const res = await backend.sendAction("get-agent-runs");
+  runs = res.runs || [];
+
+  // Listen for run events
+  unsubscribeRunEvents = backend.onAction('run-event', updateRun);
+}
+
+init();
+
+// Get all runs
+export function getRuns() {
+  return runs;
+}
+
+// Get a specific run by ID
+export function getRun(runId) {
+  return runs.find(r => r.id === runId);
+}
+
+
+// Cancel a run
+export async function cancelRun(runId) {
+  try {
+    const res = await backend.sendAction("cancel-agent-run", { runId });
+    console.log(`Cancelled run ${runId}`, res);
+    return res;
+  } catch (err) {
+    console.error(`Failed to cancel run ${runId}:`, err);
+    throw err;
+  }
+}
+
+// Subscribe to runs changes
+export function addRunListener(callback) {
+  const listener = new RunListener();
+  listener.addCallback(callback);
+  listeners.runs.push(listener);   
+  return listener; 
+} 
+
+// Subscribe to run chat message events
+export function onChatMessage(chatId, callback) {
+  let chatListeners = listeners.chats[chatId];
+  if (!chatListeners) {
+    chatListeners = [];
+    listeners.chats[chatId] = chatListeners;
+  }
+  chatListeners.push(callback);
+  return () => {
+    const index = chatListeners.indexOf(callback);
+    if (index > -1) {
+      chatListeners.splice(index, 1);
+    }
+  };
+}

frontend/src/views/Chat.vue

@@ -9,6 +9,7 @@
 
   import * as backend from '../backend';
   import * as ui from '../ui';
+  import * as runs from '../runs';
 
   const props = defineProps({
     id: {
@@ -33,6 +34,7 @@
   const expandedThreads = ref({}); // Object<int: bool>, to simulate a Set<int>
   const editingMessageId = ref(null);
   const noteMode = ref(false);
+  const messageRunMap = ref({});
 
   const editInfo = computed(() => {
     if (editingMessageId.value != null) {
@@ -50,6 +52,8 @@
 
   let _isMounted = false;
   let unsubscribeChat = null;
+  let unsubscribeRuns = null;
+  let runListener = null;
 
   // TODO: user sent message is added without id.
   // User should receive his own message notification, and find its message to assign it an id.
@@ -110,6 +114,8 @@
       });
     });
 
+    unsubscribeRuns = runs.onChatMessage(chatId, onRunMessage);
+
     let response = await backend.sendAction("get-chat-messages", {chatId});
     // Always good to duplicate when assigning to a ref
     messages.value = response.messages.map(it => ({
@@ -147,6 +153,29 @@
   async function unloadChatState () {
     if (unsubscribeChat) unsubscribeChat();
     unsubscribeChat = null;
+    if (unsubscribeRuns) unsubscribeRuns();
+    unsubscribeRuns = null;
+  }
+
+  function onRunMessage (event) {
+    if (!runListener) {
+      runListener = runs.addRunListener(onRunUpdate);
+    }
+    runListener.addRunId(event.runId);
+    messageRunMap.value[event.messageId] = event.runId;
+    ui.alert("Chat Run Message: " + JSON.stringify(event));
+  }
+  
+  function onRunUpdate (newRun) {
+    if (newRun.done) {
+      runListener.runIds.delete(newRun.id);
+      let map = messageRunMap.value;
+      for (let messageId in map) {
+        if (map[messageId] == newRun.id) {
+          delete map[messageId];
+        }
+      }
+    }
   }
 
   // This won't fire when called, the mounted callback will take care
@@ -428,6 +457,9 @@
         :message="msg" 
         @open-menu="chatEntryContextMenu"
         @toggle-thread="expandThread"
+
+        :cancellable="messageRunMap[msg.id]"
+        @cancel="runs.cancelRun(messageRunMap[msg.id])"
       />
     </div>
 

frontend/src/views/RunsView.vue

@@ -1,19 +1,14 @@
 <script setup>
   import { ref, onMounted, onUnmounted } from 'vue';
   import TitleBar from '../components/TitleBar.vue';
-  import * as backend from '../backend';
+  import * as runs from '../runs';
 
-  const runs = ref([]);
+  const runList = ref([]);
   const expandedRunId = ref(null);
-  const loading = ref(true);
+  const runListener = runs.addRunListener(onRunEvent);
-  let unsubscribeRunEvents = null;
 
   const toggleRun = (id) => {
-    if (expandedRunId.value === id) {
+    expandedRunId.value = expandedRunId.value == id ? null : id;
-      expandedRunId.value = null;
-    } else {
-      expandedRunId.value = id;
-    }
   };
 
   const formatTime = (timestamp) => {
@@ -21,72 +16,24 @@
     return new Date(timestamp * 1000).toLocaleString();
   };
 
-  const getStatusBadge = (run) => {
-    return run.done ? 'Done' : 'Running';
-  };
-
-  const updateRun = (event) => {
-    const { name, data } = event;
-    
-    // Find the run by ID
-    const runIndex = runs.value.findIndex(r => r.id === data.runId);
-    
-    if (runIndex === -1) {
-      // New run was created
-      if (name === 'run-started') {
-        runs.value.push(data);
-      }
-    } else {
-      // Update existing run
-      const run = runs.value[runIndex];
-      
-      if (name === 'run-completed') {
-        run.done = true;
-        run.endTime = data.endTime;
-        run.output = data.output;
-        
-        // Mark as expanded to show the completion
-        if (expandedRunId.value !== data.runId) {
-          expandedRunId.value = data.runId;
-        }
-      } else if (name === 'run-event') {
-        // Add new event to the run
-        if (!run.events) {
-          run.events = [];
-        }
-        run.events.push(data.data);
-        
-        // If this run is expanded, update it
-        if (expandedRunId.value === data.runId) {
-          // No DOM update needed as we're re-rendering the whole list
-        }
-      }
-    }
-  };
-
   async function cancelRun (runId) {
-    const res = await backend.sendAction("cancel-agent-run", {runId});
+    await runs.cancelRun(runId);
-    console.log(`Cancelled run ${runId}`, res);
   };
 
-  onMounted(async () => {
+  function onRunEvent (newRun) {
-    try {
+    let run = runList.value.findOne(r => r.id == newRun.id);
-      const res = await backend.sendAction("get-agent-runs");
+    Object.assign(run, newRun);
-      runs.value = res.runs || [];
-    } catch (err) {
-      console.error("Failed to load agent runs:", err);
-    } finally {
-      loading.value = false;
   }
 
-    // Listen for live run events
+  onMounted(() => {
-    unsubscribeRunEvents = backend.onRunEvents(updateRun);
+    runList.value = runs.getRuns().map(
+      run => Object.assign({}, run)
+    );
+    runListener.setIdList(runList.value.map(r => r.id));
   });
 
   onUnmounted(() => {
-    if (unsubscribeRunEvents) {
+    runListener.setIdList([]);
-      unsubscribeRunEvents();
-    }
   });
 </script>
 
@@ -96,18 +43,16 @@
       <h1>Agent Runs</h1>
     </TitleBar>
 
-    <div v-if="loading" class="loading">Loading runs...</div>
+    <div v-if="runList.length === 0" class="empty">
-
-    <div v-else-if="runs.length === 0" class="empty">
       No agent runs found.
     </div>
 
-    <div v-else v-for="run in runs" :key="run.id" class="run-entry">
+    <div v-else v-for="run in runList" :key="run.id" class="run-entry">
       <div class="run-header" @click="toggleRun(run.id)">
         <span class="run-id">{{ run.id }}</span>
         <span class="run-agent">Agent: {{ run.agentId }}</span>
         <span :class="['run-status', run.done ? 'status-done' : 'status-running']">
-          {{ getStatusBadge(run) }}
+          {{ run.done ? 'Done' : 'Running' }}
         </span>
         <button v-if="!run.done" @click="cancelRun(run.id)">
           Cancel