diff --git a/backend/src/agent.py b/backend/src/agent.py index d7ad926..db08672 100644 --- a/backend/src/agent.py +++ b/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( - chat_id=message.chat_id, + new_message = await self.app.send_chat_message( 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", { - "runId": self.id, + self.emit_event('created', { "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 - }) - diff --git a/backend/src/server.py b/backend/src/server.py index c41e27a..f74d10b 100644 --- a/backend/src/server.py +++ b/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"}) diff --git a/frontend/src/backend.js b/frontend/src/backend.js index 9891b61..9167e32 100644 --- a/frontend/src/backend.js +++ b/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') { - // Broadcast run events to all listeners - runListeners.forEach(callback => { - try { - callback(packet); - } catch (e) { - console.error(`Error in run event listener:`, e); + } else if (packet.action) { + let listeners = actionListeners[packet.action]; + if (listeners && listeners.length) { + for (let f of listeners) { + f(packet); } - }); - } else { - if (orphanHandlers.length) { - const f = orphanHandlers.shift(); - f(packet); - } else if (packet.error) { - console.error(`Backend Error: ${packet.error}`); } + } else 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"); diff --git a/frontend/src/components/Message.vue b/frontend/src/components/Message.vue index 43c2bc0..591e992 100644 --- a/frontend/src/components/Message.vue +++ b/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) { {{message.expanded ? "Hide" : "Show"}} + + diff --git a/frontend/src/runs.js b/frontend/src/runs.js new file mode 100644 index 0000000..ce04913 --- /dev/null +++ b/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); + } + }; +} \ No newline at end of file diff --git a/frontend/src/views/Chat.vue b/frontend/src/views/Chat.vue index c2c6e13..7111420 100644 --- a/frontend/src/views/Chat.vue +++ b/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, to simulate a Set 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])" /> diff --git a/frontend/src/views/RunsView.vue b/frontend/src/views/RunsView.vue index b0f3670..1016534 100644 --- a/frontend/src/views/RunsView.vue +++ b/frontend/src/views/RunsView.vue @@ -1,19 +1,14 @@ @@ -96,18 +43,16 @@

Agent Runs

-
Loading runs...
- -
+
No agent runs found.
-
+
{{ run.id }} Agent: {{ run.agentId }} - {{ getStatusBadge(run) }} + {{ run.done ? 'Done' : 'Running' }}