From ab30f9667c70e695cc02dd57748797e58e2421a9 Mon Sep 17 00:00:00 2001 From: andypalmi Date: Fri, 9 Oct 2026 14:48:48 +0200 Subject: [PATCH 1/2] Send Expert chat messages one at a time Queue every send to the Expert (chat turns, opening a conversation, approval resumes and the instance ready/failed notices) so only one message is in flight per session. A turn is finished when its reply arrives or it is stopped; Stop only ends the turn in flight and the next queued message goes out. Start Over drops the queue. --- frontend/src/stores/product-expert.js | 61 ++++++++++-- .../frontend/stores/product-expert.spec.js | 93 ++++++++++++++++++- 2 files changed, 147 insertions(+), 7 deletions(-) diff --git a/frontend/src/stores/product-expert.js b/frontend/src/stores/product-expert.js index 844b1f86ef..8d9df3518a 100644 --- a/frontend/src/stores/product-expert.js +++ b/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: { diff --git a/test/unit/frontend/stores/product-expert.spec.js b/test/unit/frontend/stores/product-expert.spec.js index 5f13aae513..b5cd417e59 100644 --- a/test/unit/frontend/stores/product-expert.spec.js +++ b/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 From e228dc56972f3481b43fbebcdf6c2b67b1e99bb6 Mon Sep 17 00:00:00 2001 From: andypalmi Date: Fri, 9 Oct 2026 15:53:50 +0200 Subject: [PATCH 2/2] Disable Expert approval cards once a newer message arrives Question and plan cards already stop being answerable once anything is posted after them. Approval cards stayed clickable, so a decision could be sent for a call the Expert had already cancelled. --- .../messages/components/AnswerWrapper.vue | 3 +- .../components/expert/AnswerWrapper.spec.js | 65 +++++++++++++++++++ 2 files changed, 67 insertions(+), 1 deletion(-) create mode 100644 test/unit/frontend/components/expert/AnswerWrapper.spec.js diff --git a/frontend/src/components/expert/components/messages/components/AnswerWrapper.vue b/frontend/src/components/expert/components/messages/components/AnswerWrapper.vue index 57ffde5f9e..8dd13f71ce 100644 --- a/frontend/src/components/expert/components/messages/components/AnswerWrapper.vue +++ b/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 diff --git a/test/unit/frontend/components/expert/AnswerWrapper.spec.js b/test/unit/frontend/components/expert/AnswerWrapper.spec.js new file mode 100644 index 0000000000..5b642aacdb --- /dev/null +++ b/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: '
' } } + } + }) + // 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) + }) +})