From 9c1302d99d7a1fc89b4a21b5caeec2cd9d9f450e Mon Sep 17 00:00:00 2001 From: andypalmi Date: Tue, 13 Oct 2026 12:38:25 +0200 Subject: [PATCH 01/22] Route streamed activity on the Expert status topic into the reply bubble --- .../Components/expert/streamedActivity.js | 12 +++ frontend/src/stores/product-expert.js | 81 ++++++++++++++++++- .../frontend/stores/product-expert.spec.js | 68 ++++++++++++++++ 3 files changed, 158 insertions(+), 3 deletions(-) create mode 100644 frontend/src/composables/Components/expert/streamedActivity.js diff --git a/frontend/src/composables/Components/expert/streamedActivity.js b/frontend/src/composables/Components/expert/streamedActivity.js new file mode 100644 index 0000000000..83c7b28b3e --- /dev/null +++ b/frontend/src/composables/Components/expert/streamedActivity.js @@ -0,0 +1,12 @@ +// Answer kinds that render as a tool-call card. 'tool-record' is what a streamed +// record arrives as; the others are what the same record carries in a final answer. +export const TOOL_RECORD_KINDS = ['tool-record', 'mcp_tool', 'mcp_resource', 'mcp_resource_template', 'mcp_prompt'] + +// Kinds that arrive on expert:status-message and render in seq order ahead of the final reply. +export const STREAMED_KINDS = ['tool-record', 'reasoning', 'message'] + +export const ACTIVITY_KINDS = [...TOOL_RECORD_KINDS, 'reasoning', 'message'] + +export function isToolRecord (item) { + return TOOL_RECORD_KINDS.includes(item?.kind) +} diff --git a/frontend/src/stores/product-expert.js b/frontend/src/stores/product-expert.js index 14d0d0f142..ee1a5b3cb3 100644 --- a/frontend/src/stores/product-expert.js +++ b/frontend/src/stores/product-expert.js @@ -4,6 +4,7 @@ import { markRaw } from 'vue' import expertApi from '../api/expert.js' import userApi from '../api/user.js' +import { STREAMED_KINDS, isToolRecord } from '../composables/Components/expert/streamedActivity.js' import useTimerHelper from '../composables/TimerHelper.js' import { useAccountAuthStore } from './account-auth.js' @@ -555,6 +556,16 @@ export const useProductExpertStore = defineStore('product-expert', { } } + // Streamed activity rides expert:status-message with a `kind`. It renders on the + // turn's reply, so it is neither acked nor shown on the loading line, and a kind + // this client does not know is ignored whole. + if (parsedTopic.inflightType === 'expert:status-message' && payload?.kind != null) { + if (STREAMED_KINDS.includes(payload.kind) && this._inFlightRequests.has(chatTransactionId)) { + this._appendStreamedActivity(chatTransactionId, payload) + } + return + } + // expert:tasks has its own panel; its status rides expert:status-message. // Every other inflight type feeds the loading line, but only while the turn is // still live: a status trailing the final reply must not repaint the next turn. @@ -661,7 +672,7 @@ export const useProductExpertStore = defineStore('product-expert', { // do nothing } }, - async handleMessageResponse (response) { + async handleMessageResponse (response, transactionId) { // ignore aborted messages through mqtt if (Object.prototype.hasOwnProperty.call(response, 'aborted') && response.aborted === true) return @@ -675,7 +686,14 @@ export const useProductExpertStore = defineStore('product-expert', { } if (response.answer && Array.isArray(response.answer)) { - this.addAiMessage(response) + const streamedBubble = transactionId + ? this._agentStore.messages.find(m => m._type === 'ai' && m._transactionId === transactionId) + : null + if (streamedBubble) { + this._mergeFinalAnswer(streamedBubble, response) + } else { + this.addAiMessage(response) + } this._clearInFlightUpdates() // A deferred-approval turn (#421) ends with one card per tool that needs // approval. Track the batch so we can send every decision back at once. @@ -1017,6 +1035,63 @@ export const useProductExpertStore = defineStore('product-expert', { _uuid: uuidv4() }) }, + // Place one streamed item in the turn's reply bubble, creating the bubble on the + // first item. An id already in the bubble is replaced in place; otherwise the item + // goes in seq order, so events arriving out of order still render in order. + _appendStreamedActivity (transactionId, item) { + if (!transactionId || !item) return + + const messages = this._agentStore.messages + let bubble = messages.find(m => m._type === 'ai' && m._transactionId === transactionId) + if (!bubble) { + bubble = { + answer: [], + _type: 'ai', + _timestamp: Date.now(), + _streamed: false, + _uuid: uuidv4(), + _transactionId: transactionId + } + messages.push(bubble) + } + + const existingIndex = item.id ? bubble.answer.findIndex(a => a.id === item.id) : -1 + if (existingIndex !== -1) { + const existing = bubble.answer[existingIndex] + bubble.answer.splice(existingIndex, 1, { ...item, _uuid: existing._uuid, _streamed: true }) + return + } + this._insertBySeq(bubble, { ...item, _uuid: uuidv4(), _streamed: true }) + }, + _insertBySeq (bubble, entry) { + if (typeof entry.seq !== 'number') { + bubble.answer.push(entry) + return + } + const insertBefore = bubble.answer.findIndex(a => typeof a.seq === 'number' && a.seq > entry.seq) + if (insertBefore === -1) { + bubble.answer.push(entry) + } else { + bubble.answer.splice(insertBefore, 0, entry) + } + }, + // The final reply lands in the bubble its streamed items built. Tool records it + // carries upgrade the streamed card with the same id; the rest is appended. + _mergeFinalAnswer (bubble, response) { + for (const item of response.answer) { + const existingIndex = item.id && isToolRecord(item) + ? bubble.answer.findIndex(a => a.id === item.id && isToolRecord(a)) + : -1 + if (existingIndex !== -1) { + const existing = bubble.answer[existingIndex] + bubble.answer.splice(existingIndex, 1, { ...existing, ...item, seq: existing.seq ?? item.seq, _uuid: existing._uuid }) + continue + } + this._insertBySeq(bubble, { ...item, _uuid: uuidv4(), _streamed: false }) + } + const { answer: _answer, ...rest } = response + Object.assign(bubble, rest) + }, addUserMessage (message) { this._agentStore.messages.push({ _type: 'human', @@ -1111,7 +1186,7 @@ export const useProductExpertStore = defineStore('product-expert', { this._recentlyCompleted.delete(this._recentlyCompleted.keys().next().value) } try { - await this.handleMessageResponse(JSON.parse(message.toString())) + await this.handleMessageResponse(JSON.parse(message.toString()), transactionId) } finally { this._drainSendQueue() } diff --git a/test/unit/frontend/stores/product-expert.spec.js b/test/unit/frontend/stores/product-expert.spec.js index 098b516adb..078dfe0f09 100644 --- a/test/unit/frontend/stores/product-expert.spec.js +++ b/test/unit/frontend/stores/product-expert.spec.js @@ -684,6 +684,74 @@ describe('product-expert store', () => { expect(JSON.parse(mqttService.publishMessage.mock.calls[0][1].payload)).toEqual({ ok: true }) }) + describe('streamed activity', () => { + const statusTopic = 'ff/v1/expert/u1/chat-1/team/t1/support/inflight/expert:status-message/request' + const send = (payload, srf = 'srf-1', turn = 'turn-1') => store._onMqttMessage(statusTopic, JSON.stringify(payload), packet(srf, turn)) + const bubble = () => agent.messages.find(m => m._transactionId === 'turn-1') + + beforeEach(() => { + agent.inFlightRequests.set('turn-1', { query: 'x', transactionId: 'turn-1' }) + }) + + it('inserts a known kind by seq without acking or touching the loading line', async () => { + await send({ kind: 'message', id: 'm2', seq: 2, text: 'second' }, 'srf-1') + await send({ kind: 'tool-record', id: 't1', seq: 1, reason: 'why', ok: true }, 'srf-2') + + expect(bubble().answer.map(a => a.id)).toEqual(['t1', 'm2']) + expect(mqttService.publishMessage).not.toHaveBeenCalled() + expect(store.inFlightUpdates).toHaveLength(0) + }) + + it('replaces an item with the same id in place', async () => { + await send({ kind: 'reasoning', id: 'r1', seq: 1, text: 'Looking', done: false }, 'srf-1') + await send({ kind: 'message', id: 'm1', seq: 2, text: 'hi' }, 'srf-2') + await send({ kind: 'reasoning', id: 'r1', seq: 1, text: 'Looking at the flow', done: true }, 'srf-3') + + expect(bubble().answer).toHaveLength(2) + expect(bubble().answer[0]).toMatchObject({ id: 'r1', text: 'Looking at the flow', done: true }) + }) + + it('keeps handling a payload without a kind as a status line', async () => { + await send({ status: 'Checking the flow' }) + + expect(bubble()).toBeUndefined() + expect(store.inFlightUpdates).toEqual(['Checking the flow']) + expect(JSON.parse(mqttService.publishMessage.mock.calls[0][1].payload)).toEqual({ ack: true }) + }) + + it('ignores an unknown kind entirely', async () => { + await send({ kind: 'hologram', id: 'h1', seq: 1, status: 'new thing' }) + + expect(agent.messages).toHaveLength(0) + expect(store.inFlightUpdates).toHaveLength(0) + expect(mqttService.publishMessage).not.toHaveBeenCalled() + }) + + it('drops an event that arrives after the turn has finished', async () => { + await store._onMqttMessage(replyTopic, JSON.stringify({}), packet('turn-1', 'turn-1')) + + await send({ kind: 'message', id: 'm1', seq: 1, text: 'late' }) + + expect(agent.messages).toHaveLength(0) + expect(mqttService.publishMessage).not.toHaveBeenCalled() + }) + + it('merges final-answer tool records into streamed cards by id', async () => { + await send({ kind: 'tool-record', id: 't1', seq: 1, reason: 'why', ok: true }, 'srf-1') + const answer = [ + { kind: 'mcp_tool', id: 't1', seq: 1, reason: 'why', toolName: 'list_flows' }, + { kind: 'mcp_tool', toolName: 'legacy' }, + { kind: 'chat', content: 'Done' } + ] + + await store._onMqttMessage(replyTopic, JSON.stringify({ answer }), packet('turn-1', 'turn-1')) + + expect(agent.messages).toHaveLength(1) + expect(bubble().answer.map(a => a.id || a.toolName || a.kind)).toEqual(['t1', 'legacy', 'chat']) + expect(bubble().answer[0]).toMatchObject({ toolName: 'list_flows', ok: true }) + }) + }) + it('answers an editor action that fails with its error and keeps the chat going', async () => { agent.inFlightRequests.set('turn-1', { query: 'x', transactionId: 'turn-1' }) invokeAction.mockRejectedValueOnce(new Error('Node-RED command timed out')) From 5e21fe320e58ad91bfab1fa0d5c0f751759e2bf6 Mon Sep 17 00:00:00 2001 From: andypalmi Date: Tue, 13 Oct 2026 12:39:47 +0200 Subject: [PATCH 02/22] Render Expert tool calls as collapsible cards --- .../expert/components/messages/AiMessage.vue | 62 ++-- .../messages/components/TimelineItem.vue | 264 ++++++++++++++++++ .../components/resources/ToolApprovalCard.vue | 2 +- frontend/src/stores/context.js | 1 + 4 files changed, 289 insertions(+), 40 deletions(-) create mode 100644 frontend/src/components/expert/components/messages/components/TimelineItem.vue diff --git a/frontend/src/components/expert/components/messages/AiMessage.vue b/frontend/src/components/expert/components/messages/AiMessage.vue index ba97cf0ee2..a9b2dd86d8 100644 --- a/frontend/src/components/expert/components/messages/AiMessage.vue +++ b/frontend/src/components/expert/components/messages/AiMessage.vue @@ -1,6 +1,6 @@