From 548a6c79cacfdce9f8ea43e5a4cd3ee1e408396b Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 10 Oct 2026 19:25:11 +0200 Subject: [PATCH 1/2] fix(chat): show streamed text as soon as it arrives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The streaming chat message typed the received text one character every 5 ms and re-rendered the markdown after each character. That caps the display at 200 characters per second, slower than common models stream (about 140 tokens/s, roughly 550 characters/s, for a flash model), so the visible text fell behind the received text and jumped ahead with each new chunk. The first chunk was not displayed at all: the watcher was not immediate, and the streaming message is mounted with the first chunk already set, so it showed "Reasoning…" until the second chunk. Reveal each new chunk over at most 200 ms in ~16 ms steps. That is shorter than the 250 ms interval at which integration_openai reports intermediate output, so the display catches up before the next chunk. Show the content that was received before the component was mounted. Signed-off-by: Daniel --- package.json | 3 +- src/components/ChattyLLM/ChatMessage.vue | 46 +++++++++----- src/components/ChattyLLM/streamReveal.js | 47 ++++++++++++++ tests/frontend/streamReveal.test.js | 79 ++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 18 deletions(-) create mode 100644 src/components/ChattyLLM/streamReveal.js create mode 100644 tests/frontend/streamReveal.test.js diff --git a/package.json b/package.json index a9a57b4e4..9cc7794fe 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,8 @@ "lint": "eslint --ext=.js,.vue src", "lint:fix": "eslint --ext=.js,.vue src --fix", "stylelint": "stylelint src/**/*.vue src/**/*.scss src/**/*.css", - "stylelint:fix": "stylelint src/**/*.vue src/**/*.scss src/**/*.css --fix" + "stylelint:fix": "stylelint src/**/*.vue src/**/*.scss src/**/*.css --fix", + "test:js": "node --test 'tests/frontend/**/*.test.js'" }, "repository": { "type": "git", diff --git a/src/components/ChattyLLM/ChatMessage.vue b/src/components/ChattyLLM/ChatMessage.vue index 3a1fd95d5..76e801c99 100644 --- a/src/components/ChattyLLM/ChatMessage.vue +++ b/src/components/ChattyLLM/ChatMessage.vue @@ -136,6 +136,7 @@ import { generateOcsUrl } from '@nextcloud/router' import axios from '@nextcloud/axios' import { SHAPE_TYPE_NAMES } from '../../constants.js' import FileDisplay from '../fields/FileDisplay.vue' +import { getRevealLength, STREAM_REVEAL_STEP_MS } from './streamReveal.js' const PLAIN_URL_PATTERN = /(?:\s|^|\()((?:https?:\/\/)(?:[-A-Z0-9+_.]+(?::[0-9]+)?(?:\/[-A-Z0-9+&@#%?=~_|!:,.;()]*)*))(?:\s|$|\))/ig const MARKDOWN_LINK_PATTERN = /\[[-A-Z0-9+&@#%?=~_|!:,.;()]+\]\(((?:https?:\/\/)(?:[-A-Z0-9+_.]+(?::[0-9]+)?(?:\/[-A-Z0-9+&@#%?=~_|!:,.;]*)*))\)/ig @@ -233,24 +234,16 @@ export default { }, watch: { - // Pseudo streaming - 'message.content': async function(messageContent, oldMessageContent) { - if (!this.streaming) { - return - } - if (oldMessageContent) { - this.streamedMessageContent = oldMessageContent - messageContent = messageContent.replace(oldMessageContent, '') - } - let cachedStreamedMessageContent - for (const char of messageContent.split('')) { - this.streamedMessageContent += char - cachedStreamedMessageContent = this.streamedMessageContent - await new Promise(resolve => setTimeout(resolve, 5)) - if (cachedStreamedMessageContent !== this.streamedMessageContent) { - break + // Pseudo streaming: reveal the received text progressively. + // Immediate, so the text received before this component was mounted (the first chunk) is shown too. + 'message.content': { + immediate: true, + handler(messageContent) { + if (!this.streaming) { + return } - } + this.revealStreamedContent(messageContent ?? '') + }, }, }, @@ -258,7 +251,26 @@ export default { this.fetch() }, + beforeUnmount() { + clearTimeout(this.revealTimer) + }, + methods: { + revealStreamedContent(content) { + clearTimeout(this.revealTimer) + const shown = this.streamedMessageContent + // if the new content does not continue the displayed text, show it all at once + const fromLength = content.startsWith(shown) ? shown.length : content.length + const receivedAt = Date.now() + const step = () => { + const length = getRevealLength(content, fromLength, Date.now() - receivedAt) + this.streamedMessageContent = content.slice(0, length) + if (length < content.length) { + this.revealTimer = setTimeout(step, STREAM_REVEAL_STEP_MS) + } + } + step() + }, copyMessage(message) { navigator.clipboard.writeText(message) showSuccess(t('assistant', 'Message copied to clipboard')) diff --git a/src/components/ChattyLLM/streamReveal.js b/src/components/ChattyLLM/streamReveal.js new file mode 100644 index 000000000..a2b61a114 --- /dev/null +++ b/src/components/ChattyLLM/streamReveal.js @@ -0,0 +1,47 @@ +/** + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +/** + * Maximum time it takes to display newly received streamed text. + * + * It is shorter than the 250 ms interval at which integration_openai reports + * intermediate output, so the displayed text has caught up with the received + * text before the next chunk arrives. + */ +export const STREAM_REVEAL_DURATION_MS = 200 + +/** + * Delay between two reveal steps (about one frame). + */ +export const STREAM_REVEAL_STEP_MS = 16 + +/** + * Get the length of streamed text to display `elapsedMs` after it was received. + * + * The part of `text` after `fromLength` is revealed linearly over `durationMs`, + * so the speed follows how fast the text arrives instead of being a fixed + * number of characters per second. After `durationMs`, all of `text` is displayed. + * + * @param {string} text The whole text received so far + * @param {number} fromLength Length that was already displayed when the text was received + * @param {number} elapsedMs Time since the text was received + * @param {number} durationMs Time after which all of the text is displayed + * @return {number} The length to display. It never splits a surrogate pair. + */ +export function getRevealLength(text, fromLength, elapsedMs, durationMs = STREAM_REVEAL_DURATION_MS) { + const toLength = text.length + const start = Math.min(Math.max(fromLength, 0), toLength) + if (start === toLength || durationMs <= 0 || elapsedMs >= durationMs) { + return toLength + } + const progress = Math.max(elapsedMs, 0) / durationMs + let length = start + Math.ceil((toLength - start) * progress) + // do not cut an emoji or any other character outside the BMP in half + const lastCode = text.charCodeAt(length - 1) + if (lastCode >= 0xD800 && lastCode <= 0xDBFF) { + length++ + } + return Math.min(length, toLength) +} diff --git a/tests/frontend/streamReveal.test.js b/tests/frontend/streamReveal.test.js new file mode 100644 index 000000000..7e7650700 --- /dev/null +++ b/tests/frontend/streamReveal.test.js @@ -0,0 +1,79 @@ +/** + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' + +import { + getRevealLength, + STREAM_REVEAL_DURATION_MS, + STREAM_REVEAL_STEP_MS, +} from '../../src/components/ChattyLLM/streamReveal.js' + +describe('getRevealLength', () => { + const text = 'x'.repeat(100) + + it('starts from what is already displayed', () => { + assert.equal(getRevealLength(text, 40, 0, 200), 40) + }) + + it('reveals the new text linearly over the duration', () => { + assert.equal(getRevealLength(text, 0, 50, 200), 25) + assert.equal(getRevealLength(text, 0, 100, 200), 50) + assert.equal(getRevealLength(text, 40, 100, 200), 70) + }) + + it('displays everything once the duration has elapsed', () => { + assert.equal(getRevealLength(text, 0, 200, 200), 100) + assert.equal(getRevealLength(text, 0, 5000, 200), 100) + }) + + it('displays everything when the duration is not positive', () => { + assert.equal(getRevealLength(text, 0, 0, 0), 100) + assert.equal(getRevealLength(text, 0, 0, -1), 100) + }) + + it('clamps out of range start lengths', () => { + assert.equal(getRevealLength(text, -10, 100, 200), 50) + assert.equal(getRevealLength(text, 150, 0, 200), 100) + assert.equal(getRevealLength('', 0, 0, 200), 0) + }) + + it('treats a negative elapsed time as zero', () => { + assert.equal(getRevealLength(text, 10, -50, 200), 10) + }) + + it('never splits a surrogate pair', () => { + // 'ab' + one emoji (2 UTF-16 code units) + 'cd': length 6 + const emojiText = 'ab\u{1F600}cd' + assert.equal(emojiText.length, 6) + // ceil(6 * 0.5) = 3 would end between the two halves of the emoji + const length = getRevealLength(emojiText, 0, 100, 200) + assert.equal(length, 4) + assert.equal(emojiText.slice(0, length), 'ab\u{1F600}') + }) + + it('only grows and stays within the text while time passes', () => { + const longText = 'Streamed answer with an emoji \u{1F680} and more text. '.repeat(40) + let previous = 0 + for (let elapsed = 0; elapsed <= STREAM_REVEAL_DURATION_MS + STREAM_REVEAL_STEP_MS; elapsed += STREAM_REVEAL_STEP_MS) { + const length = getRevealLength(longText, 0, elapsed) + assert.ok(length >= previous, `length went down at ${elapsed} ms`) + assert.ok(length <= longText.length) + const lastCode = longText.charCodeAt(length - 1) + assert.ok(!(lastCode >= 0xD800 && lastCode <= 0xDBFF), `surrogate pair split at ${elapsed} ms`) + previous = length + } + assert.equal(previous, longText.length) + }) + + it('catches up with a large chunk within the default duration', () => { + // a 250 ms chunk of a model streaming ~140 tokens/s is roughly 140 characters, + // a fast model or a 2 s polling fallback can deliver more than 1000 at once + const chunk = 'y'.repeat(1200) + assert.ok(STREAM_REVEAL_DURATION_MS < 250) + assert.equal(getRevealLength(chunk, 0, STREAM_REVEAL_DURATION_MS), chunk.length) + }) +}) From 26d3dc03da676f5ee7fb60982ae8cde1995097d9 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 10 Oct 2026 19:25:11 +0200 Subject: [PATCH 2/2] perf(chat): get the answer as soon as the task is done The chat checked the generation task only every 2 seconds, so the final message and the re-enabled input came 0 to 2 s (1 s on average) after the task had finished, even when notify_push had already announced it. assistant.js already turns the taskprocessing:task_update push into assistant:task:updated and assistant:task:status:updated events. Listen to them while polling and check right away when the task succeeded, failed or was cancelled. Polling every 2 s stays as the fallback without notify_push. Only one check request runs at a time. Signed-off-by: Daniel --- .../ChattyLLM/ChattyLLMInputForm.vue | 77 ++++++++++++++----- 1 file changed, 59 insertions(+), 18 deletions(-) diff --git a/src/components/ChattyLLM/ChattyLLMInputForm.vue b/src/components/ChattyLLM/ChattyLLMInputForm.vue index 401345e99..d2ad2bbaf 100644 --- a/src/components/ChattyLLM/ChattyLLMInputForm.vue +++ b/src/components/ChattyLLM/ChattyLLMInputForm.vue @@ -287,6 +287,7 @@ import axios, { isCancel } from '@nextcloud/axios' import { showError } from '@nextcloud/dialogs' import { generateUrl, generateOcsUrl } from '@nextcloud/router' import { loadState } from '@nextcloud/initial-state' +import { subscribe, unsubscribe } from '@nextcloud/event-bus' import { listen } from '@nextcloud/notify_push' import moment from 'moment' import { SHAPE_TYPE_NAMES, TASK_STATUS_INT } from '../../constants.js' @@ -1210,17 +1211,39 @@ export default { console.debug('[assistant] HAS PUSH', hasPush) return new Promise((resolve, reject) => { - const timerId = setInterval(() => { + let timerId = null + let checkInProgress = false + let checkAgain = false + // [eventName, handler] pairs subscribed on the event bus while polling + const subscriptions = [] + + const stopPolling = () => { + clearInterval(timerId) + subscriptions.forEach(([eventName, handler]) => unsubscribe(eventName, handler)) + subscriptions.length = 0 + if (this.pollMessageGenerationTimerId === timerId) { + this.pollMessageGenerationTimerId = null + this.pollMessageGenerationCancel = null + } + } + + const check = () => { + if (this.pollMessageGenerationTimerId !== timerId) { + // this polling has been replaced or stopped + stopPolling() + return + } + if (checkInProgress) { + checkAgain = true + return + } if (this.active === null || sessionId !== this.active.id) { console.debug('Stop polling messages for session ' + sessionId + ' because it is not selected anymore') - clearInterval(timerId) - if (this.pollMessageGenerationTimerId === timerId) { - this.pollMessageGenerationTimerId = null - this.pollMessageGenerationCancel = null - } + stopPolling() reject(new TaskPollCancelledError('generation polling cancelled')) return } + checkInProgress = true axios.get( getChatURL('/check_generation'), { params: { taskId, sessionId } }, @@ -1229,9 +1252,7 @@ export default { return } const responseData = response.data - clearInterval(timerId) - this.pollMessageGenerationTimerId = null - this.pollMessageGenerationCancel = null + stopPolling() if (sessionId === this.active.id) { this.active.sessionAgencyPendingActions = responseData.sessionAgencyPendingActions this.active.agencyAnswered = false @@ -1258,9 +1279,7 @@ export default { // do not reject if response code is Http::STATUS_EXPECTATION_FAILED (417) if (error.response?.status !== 417) { console.error('checkTaskPolling error', error) - clearInterval(timerId) - this.pollMessageGenerationTimerId = null - this.pollMessageGenerationCancel = null + stopPolling() reject(error) } else { console.debug('checkTaskPolling, task is still scheduled or running') @@ -1274,16 +1293,38 @@ export default { this.updateStreamingMessage(error.response.data.task_output || {}, sessionId) } } + }).finally(() => { + checkInProgress = false + if (checkAgain) { + checkAgain = false + check() + } }) - }, 2000) + } + + // assistant.js turns the 'taskprocessing:task_update' push notification into these events. + // When our task has finished, check it right away instead of waiting for the next poll. + const onTaskUpdated = (task) => { + if (task?.id === taskId) { + check() + } + } + const onTaskStatusUpdated = ({ taskId: updatedTaskId, status }) => { + if (updatedTaskId === taskId && [TASK_STATUS_INT.failed, TASK_STATUS_INT.cancelled].includes(status)) { + check() + } + } + + timerId = setInterval(check, 2000) this.pollMessageGenerationTimerId = timerId + subscriptions.push( + ['assistant:task:updated', onTaskUpdated], + ['assistant:task:status:updated', onTaskStatusUpdated], + ) + subscriptions.forEach(([eventName, handler]) => subscribe(eventName, handler)) this.pollMessageGenerationCancel = () => { - clearInterval(timerId) - if (this.pollMessageGenerationTimerId === timerId) { - this.pollMessageGenerationTimerId = null - } + stopPolling() reject(new TaskPollCancelledError('generation polling cancelled')) - this.pollMessageGenerationCancel = null } }) },