diff --git a/package.json b/package.json index a9a57b4e..9cc7794f 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 3a1fd95d..76e801c9 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/ChattyLLMInputForm.vue b/src/components/ChattyLLM/ChattyLLMInputForm.vue index 401345e9..d2ad2bba 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 } }) }, diff --git a/src/components/ChattyLLM/streamReveal.js b/src/components/ChattyLLM/streamReveal.js new file mode 100644 index 00000000..a2b61a11 --- /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 00000000..7e765070 --- /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) + }) +})