fix(chat): show streamed answers as they arrive and finish them on push

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",

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,32 +234,43 @@ 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 ?? '')
+			},
 		},
 	},
 
 	mounted() {
 		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'))

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
 				}
 			})
 		},

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)
+}

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)
+	})
+})