Send Expert chat messages one at a time

frontend/src/components/expert/components/messages/components/AnswerWrapper.vue

@@ -103,6 +103,7 @@
             :tool-class="answer.toolClass"
             :params="answer.params"
             :status="resolvedToolApprovalStatus"
+            :disabled="interactionDisabled"
             @approve="onToolApprove"
             @allow-always="onToolAllowAlways"
             @deny="onToolDeny"
@@ -193,7 +194,7 @@ export default {
             return msgs.length > 0 && msgs[msgs.length - 1]?._uuid === this.messageUuid
         },
         interactionDisabled () {
-            // Disable the question/plan cards while a response is in flight, and once the turn
+            // Disable the question/plan/approval cards while a response is in flight, and once the turn
             // has passed — i.e. any message has arrived after this one — so a stale card from
             // an earlier turn can no longer be answered.
             return this.isWaitingForResponse || !this.isLatestMessage

frontend/src/stores/product-expert.js

@@ -53,6 +53,10 @@ export const useProductExpertStore = defineStore('product-expert', {
         // reconnect reporting running again doesn't repeat the announcement.
         _relayedInstanceIds: new Set(),
         _relayedFailedInstanceIds: new Set(),
+        // Sends waiting for the turn in flight to settle, so the agent never gets two
+        // messages for one session at once. Not persisted.
+        _sendQueue: [],
+        _sendDraining: false,
         // Open human-in-the-loop approval batch (#421). When a turn defers a tool batch
         // for approval the agent ends the turn and returns the card(s); we hold the
         // decisions here until every card is answered, then send them back in one resume
@@ -256,6 +260,10 @@ export const useProductExpertStore = defineStore('product-expert', {
                 agentStore.sessionId = uuidv4()
             }
 
+            return this._enqueueSend(() => this._openConversation())
+        },
+        async _openConversation () {
+            const agentStore = this._agentStore
             agentStore.abortController = markRaw(new AbortController())
             try {
                 const result = await this.sendQuery({ query: '' })
@@ -275,7 +283,35 @@ export const useProductExpertStore = defineStore('product-expert', {
                 agentStore.abortController = null
             }
         },
-        async handleQuery ({ query }) {
+        handleQuery ({ query }) {
+            return this._enqueueSend(() => this._handleQuery({ query }))
+        },
+        // Runs one send at a time. A job is finished once its turn has settled: the reply
+        // arrived, or the turn was stopped or errored. Whoever settles a turn calls
+        // _drainSendQueue to release the next job.
+        _enqueueSend (job) {
+            return new Promise((resolve, reject) => {
+                this._sendQueue.push({ job, resolve, reject })
+                this._drainSendQueue()
+            })
+        },
+        async _drainSendQueue () {
+            if (this._sendDraining) return
+            this._sendDraining = true
+            try {
+                while (this._sendQueue.length && !this.isWaitingForResponse) {
+                    const { job, resolve, reject } = this._sendQueue.shift()
+                    try {
+                        resolve(await job())
+                    } catch (error) {
+                        reject(error)
+                    }
+                }
+            } finally {
+                this._sendDraining = false
+            }
+        },
+        async _handleQuery ({ query }) {
             const agentStore = this._agentStore
 
             // Auto-initialize session ID if not set
@@ -653,7 +689,10 @@ export const useProductExpertStore = defineStore('product-expert', {
         // resumes from the persisted tool calls, runs the approved ones, and continues.
         // Transport-agnostic: over MQTT the reply arrives via the push handler; over HTTP it
         // is the awaited response, rendered here (mirrors handleQuery).
-        async resumeToolApprovals (decisions) {
+        resumeToolApprovals (decisions) {
+            return this._enqueueSend(() => this._resumeToolApprovals(decisions))
+        },
+        async _resumeToolApprovals (decisions) {
             const agentStore = this._agentStore
             agentStore.abortController = markRaw(new AbortController())
             try {
@@ -696,6 +735,11 @@ export const useProductExpertStore = defineStore('product-expert', {
             // The transcript this notice belonged to is about to be cleared.
             this._clearDisconnectNotice()
 
+            // Queued sends belong to the conversation being dropped.
+            this._sendQueue.splice(0).forEach(({ resolve }) => resolve(undefined))
+            // The old session's reply can no longer arrive, so don't stay waiting on it.
+            this._inFlightRequests.clear()
+
             agentStore.sessionId = uuidv4()
             agentStore.messages = []
             agentStore.activeTaskList = null
@@ -994,6 +1038,7 @@ export const useProductExpertStore = defineStore('product-expert', {
                     this._recentlyCompleted.delete(this._recentlyCompleted.keys().next().value)
                 }
                 await this.handleMessageResponse(JSON.parse(message.toString()))
+                this._drainSendQueue()
                 break
             case parsedTopic.isInflightRequest: // in-flight request from the agent (e.g., action invocation, status update)
                 await this.handleInFlightRequest({
@@ -1420,6 +1465,12 @@ export const useProductExpertStore = defineStore('product-expert', {
             if (seen.has(instance.id)) return
             seen.add(instance.id)
 
+            const system = buildSystem(instance)
+            if (showCard) this.addEventMessage(system)
+
+            return this._enqueueSend(() => this._publishSystemEvent(instance, system, onPublished))
+        },
+        async _publishSystemEvent (instance, system, onPublished) {
             try {
                 const mqttService = getAppOrchestrator().$services.mqtt
                 const mqttTopicHelper = useMqttExpertTopicHelper()
@@ -1443,10 +1494,6 @@ export const useProductExpertStore = defineStore('product-expert', {
                     topicAction: 'request'
                 })
 
-                const system = buildSystem(instance)
-
-                if (showCard) this.addEventMessage(system)
-
                 await mqttService.publishMessage(mqttConnectionKey, {
                     topic,
                     qos: 2,
@@ -1536,6 +1583,8 @@ export const useProductExpertStore = defineStore('product-expert', {
             this._inFlightRequests.clear()
             this._clearInFlightUpdates()
             this._agentStore.activeTaskList = null
+            // Stopping ends only the turn in flight; whatever is queued goes out next.
+            this._drainSendQueue()
         }
     },
     persist: {

test/unit/frontend/components/expert/AnswerWrapper.spec.js

@@ -0,0 +1,65 @@
+import { createTestingPinia } from '@pinia/testing'
+import { flushPromises, mount } from '@vue/test-utils'
+import { describe, expect, test, vi } from 'vitest'
+
+vi.mock('../../../../../frontend/src/services/product.js', () => ({
+    default: {
+        capture: vi.fn()
+    }
+}))
+
+// The flows list pulls in the CJS flow renderer, which cannot load in the test environment
+vi.mock('@flowfuse/flow-renderer', () => ({
+    default: class FlowRenderer {}
+}))
+
+const AnswerWrapper = (await import('../../../../../frontend/src/components/expert/components/messages/components/AnswerWrapper.vue')).default
+const ToolApprovalCard = (await import('../../../../../frontend/src/components/expert/components/messages/components/resources/ToolApprovalCard.vue')).default
+const { useProductExpertStore } = await import('../../../../../frontend/src/stores/product-expert.js')
+
+const approvalAnswer = {
+    kind: 'tool-approval',
+    id: 'call-1',
+    toolUseId: 'call-1',
+    toolKey: 'flow_building_add_nodes',
+    name: 'Adding Nodes',
+    toolClass: 'write',
+    params: {},
+    status: 'pending',
+    _uuid: 'answer-1'
+}
+
+async function mountApprovalCard ({ messageUuids, isWaitingForResponse = false }) {
+    const pinia = createTestingPinia({ createSpy: vi.fn, stubActions: true })
+    const expert = useProductExpertStore(pinia)
+    expert.messages = messageUuids.map(_uuid => ({ _uuid }))
+    expert.isWaitingForResponse = isWaitingForResponse
+
+    const wrapper = mount(AnswerWrapper, {
+        props: { answer: approvalAnswer, messageUuid: 'message-1', instant: true },
+        global: {
+            plugins: [pinia],
+            stubs: { 'message-bubble': { template: '<div><slot /></div>' } }
+        }
+    })
+    // the card joins the streaming order on mount and renders on the next tick
+    await flushPromises()
+    return wrapper.findComponent(ToolApprovalCard)
+}
+
+describe('AnswerWrapper approval card', () => {
+    test('can be answered while it is the latest message', async () => {
+        const card = await mountApprovalCard({ messageUuids: ['message-1'] })
+        expect(card.props('disabled')).toBe(false)
+    })
+
+    test('goes inactive once a newer message arrives', async () => {
+        const card = await mountApprovalCard({ messageUuids: ['message-1', 'instance-ready-event'] })
+        expect(card.props('disabled')).toBe(true)
+    })
+
+    test('is inactive while a reply is in flight', async () => {
+        const card = await mountApprovalCard({ messageUuids: ['message-1'], isWaitingForResponse: true })
+        expect(card.props('disabled')).toBe(true)
+    })
+})

test/unit/frontend/stores/product-expert.spec.js

@@ -846,7 +846,10 @@ describe('product-expert store', () => {
 
             const store = useProductExpertStore()
             await store.relayInstanceReady({ id: 'inst-1', name: 'first' })
-            await store.relayInstanceReady({ id: 'inst-2', name: 'second' })
+            const second = store.relayInstanceReady({ id: 'inst-2', name: 'second' })
+            useProductExpertSupportAgentStore().inFlightRequests.clear()
+            store._drainSendQueue()
+            await second
 
             expect(mqttService.publishMessage).toHaveBeenCalledTimes(2)
         })
@@ -986,6 +989,94 @@ describe('product-expert store', () => {
         })
     })
 
+    describe('send queue', () => {
+        afterEach(() => {
+            contextState.team = null
+            uxState.building = false
+            delete accountSettingsState.featuresCheck.isExternalMqttBrokerFeatureEnabled
+            mqttService.hasClient.mockReturnValue(false)
+        })
+
+        function arrangeMqtt () {
+            uxState.building = true
+            accountSettingsState.featuresCheck.isExternalMqttBrokerFeatureEnabled = true
+            contextState.team = { id: 'team-1' }
+            mqttService.hasClient.mockReturnValue(true)
+            useAccountAuthStore().user = { id: 'user-1' }
+        }
+
+        function settle (store) {
+            useProductExpertSupportAgentStore().inFlightRequests.clear()
+            store._drainSendQueue()
+        }
+
+        it('holds an instance event while a turn is in flight', async () => {
+            arrangeMqtt()
+            const store = useProductExpertStore()
+            useProductExpertSupportAgentStore().inFlightRequests.set('turn-1', { query: 'x', transactionId: 'turn-1' })
+
+            const relay = store.relayInstanceReady({ id: 'inst-1', name: 'my-instance' })
+            await Promise.resolve()
+
+            expect(mqttService.publishMessage).not.toHaveBeenCalled()
+
+            settle(store)
+            await relay
+
+            expect(mqttService.publishMessage).toHaveBeenCalledTimes(1)
+        })
+
+        it('sends queued items one at a time in arrival order', async () => {
+            arrangeMqtt()
+            const store = useProductExpertStore()
+            const order = []
+            const agent = useProductExpertSupportAgentStore()
+            agent.inFlightRequests.set('turn-1', { query: 'x', transactionId: 'turn-1' })
+
+            const first = store._enqueueSend(async () => {
+                order.push('first')
+                agent.inFlightRequests.set('turn-2', { query: 'a', transactionId: 'turn-2' })
+            })
+            const second = store._enqueueSend(async () => { order.push('second') })
+
+            settle(store)
+            await first
+            await Promise.resolve()
+            expect(order).toEqual(['first'])
+
+            settle(store)
+            await second
+            expect(order).toEqual(['first', 'second'])
+        })
+
+        it('sends the next queued item after the turn in flight is stopped', async () => {
+            arrangeMqtt()
+            const store = useProductExpertStore()
+            useProductExpertSupportAgentStore().inFlightRequests.set('turn-1', { query: 'x', transactionId: 'turn-1' })
+
+            const relay = store.relayInstanceReady({ id: 'inst-1', name: 'my-instance' })
+            store.stopInflightChat()
+            await relay
+
+            expect(mqttService.publishMessage).toHaveBeenCalledTimes(2)
+            expect(mqttService.publishMessage.mock.calls[0][1].payload.abort).toBe(true)
+            expect(mqttService.publishMessage.mock.calls[1][1].payload.system.kind).toBe('instance-ready')
+        })
+
+        it('drops queued items on Start Over', async () => {
+            arrangeMqtt()
+            const store = useProductExpertStore()
+            useProductExpertSupportAgentStore().inFlightRequests.set('turn-1', { query: 'x', transactionId: 'turn-1' })
+
+            const relay = store.relayInstanceReady({ id: 'inst-1', name: 'my-instance' })
+            await store.startOver()
+            await relay
+
+            expect(mqttService.publishMessage).not.toHaveBeenCalled()
+            expect(store._sendQueue).toHaveLength(0)
+        })
+    })
+
     describe('approvePlan', () => {
         afterEach(() => {
             contextState.team = null