From 8cf31657a6dbe4f1bd037f2d33a413f19a266b16 Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Fri, 17 Jul 2026 05:27:48 +0000 Subject: [PATCH 1/2] fix: avoid mid-word replace splits --- .changeset/replace-prefix-holdback.md | 5 ++ .../transcription/text_transforms.test.ts | 38 +++++++++++ .../voice/transcription/text_transforms.ts | 64 ++++++++++++++----- 3 files changed, 91 insertions(+), 16 deletions(-) create mode 100644 .changeset/replace-prefix-holdback.md diff --git a/.changeset/replace-prefix-holdback.md b/.changeset/replace-prefix-holdback.md new file mode 100644 index 000000000..79a8a0be6 --- /dev/null +++ b/.changeset/replace-prefix-holdback.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Avoid splitting streamed replacement output mid-word when no replacement key prefix is pending. diff --git a/agents/src/voice/transcription/text_transforms.test.ts b/agents/src/voice/transcription/text_transforms.test.ts index 22f4b45f0..15b3436db 100644 --- a/agents/src/voice/transcription/text_transforms.test.ts +++ b/agents/src/voice/transcription/text_transforms.test.ts @@ -31,6 +31,21 @@ async function collect(stream: ReadableStream): Promise { return result; } +async function collectChunks(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const result: string[] = []; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + result.push(value); + } + } finally { + reader.releaseLock(); + } + return result; +} + describe('textTransforms.replace', () => { for (const chunkSize of [1, 2, 5, 11, 50]) { it(`replaces across chunk size ${chunkSize}`, async () => { @@ -70,6 +85,29 @@ describe('textTransforms.replace', () => { ).toBe(String.raw`a \1 \n \t here`); }); + it('flushes non-prefix text immediately', async () => { + const transform = replace({ LiveKit: 'Lyve Kit' }); + const chunks = await collectChunks(transform(streamText('you connect.', 100))); + expect(chunks).toEqual(['you connect.']); + }); + + it('holds only potential prefix', async () => { + const transform = replace({ LiveKit: 'Lyve Kit' }); + const chunks = await collectChunks(transform(streamText('visit Live', 100))); + expect(chunks[0]).toBe('visit '); + expect(chunks.join('')).toBe('visit Live'); + }); + + it('prefers longest overlapping key', async () => { + const transform = replace({ a: 'X', ab: 'Y' }); + expect(await collect(transform(streamText('ab', 100)))).toBe('Y'); + }); + + it('does not cascade', async () => { + const transform = replace({ a: 'b', b: 'c' }); + expect(await collect(transform(streamText('a', 100)))).toBe('b'); + }); + it('applies callable and built-in transforms', async () => { expect( await collect( diff --git a/agents/src/voice/transcription/text_transforms.ts b/agents/src/voice/transcription/text_transforms.ts index 38edb8b7f..f36693761 100644 --- a/agents/src/voice/transcription/text_transforms.ts +++ b/agents/src/voice/transcription/text_transforms.ts @@ -193,8 +193,49 @@ export function replace( options: { caseSensitive?: boolean } = {}, ): (text: ReadableStream) => ReadableStream { const entries = Object.entries(replacements); - const tailLen = entries.length > 0 ? Math.max(...entries.map(([old]) => old.length)) - 1 : 0; - const flags = options.caseSensitive ? 'gu' : 'giu'; + const flags = options.caseSensitive ? 'u' : 'iu'; + const lookup = new Map( + entries.map(([old, replacement]) => [ + options.caseSensitive ? old : old.toLowerCase(), + replacement, + ]), + ); + const pattern = + entries.length > 0 + ? new RegExp( + entries + .map(([old]) => old) + .sort((a, b) => b.length - a.length) + .map(escapeRegex) + .join('|'), + `g${flags}`, + ) + : null; + const maxPrefix = entries.length > 0 ? Math.max(...entries.map(([old]) => old.length - 1)) : 0; + const prefixes = new Set(); + for (const [old] of entries) { + for (let length = 1; length < old.length; length += 1) { + prefixes.add(old.slice(0, length)); + } + } + const holdbackPattern = + prefixes.size > 0 + ? new RegExp(`(?:${Array.from(prefixes).map(escapeRegex).join('|')})$`, flags) + : null; + + const apply = (value: string): string => { + if (!pattern) return value; + return value.replace( + pattern, + (match) => lookup.get(options.caseSensitive ? match : match.toLowerCase())!, + ); + }; + + const holdback = (value: string): number => { + if (!holdbackPattern) return 0; + const match = holdbackPattern.exec(value.slice(-maxPrefix)); + return match ? match[0].length : 0; + }; return (text: ReadableStream) => streamFromAsyncIterable( @@ -202,24 +243,15 @@ export function replace( let buffer = ''; for await (const chunk of readStream(text)) { - buffer += chunk; - if (buffer.length <= tailLen) { - continue; + buffer = apply(buffer + chunk); + const flushTo = buffer.length - holdback(buffer); + if (flushTo > 0) { + yield buffer.slice(0, flushTo); + buffer = buffer.slice(flushTo); } - - for (const [old, replacement] of entries) { - buffer = buffer.replace(new RegExp(escapeRegex(old), flags), () => replacement); - } - - const flushTo = buffer.length - tailLen; - yield buffer.slice(0, flushTo); - buffer = buffer.slice(flushTo); } if (buffer) { - for (const [old, replacement] of entries) { - buffer = buffer.replace(new RegExp(escapeRegex(old), flags), () => replacement); - } yield buffer; } })(), From 9a7e1e7844c58a0f255cd941490f12c087c00688 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 10:21:24 -0700 Subject: [PATCH 2/2] fix: preserve single-pass streaming replacements Resolve Unicode case-equivalent matches by their originating entry and keep transformed output out of source prefix holdback so later chunks cannot cascade. Co-authored-by: Cursor --- .../transcription/text_transforms.test.ts | 99 +++++++++++++++++++ .../voice/transcription/text_transforms.ts | 56 ++++++----- 2 files changed, 128 insertions(+), 27 deletions(-) diff --git a/agents/src/voice/transcription/text_transforms.test.ts b/agents/src/voice/transcription/text_transforms.test.ts index 15b3436db..efacad149 100644 --- a/agents/src/voice/transcription/text_transforms.test.ts +++ b/agents/src/voice/transcription/text_transforms.test.ts @@ -16,6 +16,17 @@ function streamText(text: string, chunkSize: number): ReadableStream { }); } +function streamChunks(chunks: string[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(chunk); + } + controller.close(); + }, + }); +} + async function collect(stream: ReadableStream): Promise { const reader = stream.getReader(); let result = ''; @@ -91,6 +102,17 @@ describe('textTransforms.replace', () => { expect(chunks).toEqual(['you connect.']); }); + it('preserves no-match source chunk topology', async () => { + const transform = replace({ LiveKit: 'Lyve Kit' }); + expect(await collectChunks(transform(streamChunks(['you connect.'])))).toEqual([ + 'you connect.', + ]); + expect(await collectChunks(transform(streamChunks(['you con', 'nect.'])))).toEqual([ + 'you con', + 'nect.', + ]); + }); + it('holds only potential prefix', async () => { const transform = replace({ LiveKit: 'Lyve Kit' }); const chunks = await collectChunks(transform(streamText('visit Live', 100))); @@ -98,16 +120,93 @@ describe('textTransforms.replace', () => { expect(chunks.join('')).toBe('visit Live'); }); + it('completes or rejects a held prefix with the next source chunk', async () => { + const transform = replace({ LiveKit: 'Lyve Kit' }); + expect(await collectChunks(transform(streamChunks(['visit Live', 'Kit now'])))).toEqual([ + 'visit ', + 'Lyve Kit now', + ]); + expect(await collectChunks(transform(streamChunks(['visit Live', 'ly now'])))).toEqual([ + 'visit ', + 'Lively now', + ]); + }); + + it('holds whitespace prefixes and flushes unresolved source only at EOF', async () => { + const whitespaceTransform = replace({ '\nLive': ' START' }); + expect(await collectChunks(whitespaceTransform(streamChunks(['ready\n', 'Live!'])))).toEqual([ + 'ready', + ' START!', + ]); + + const unresolvedTransform = replace({ LiveKit: 'Lyve Kit' }); + expect(await collectChunks(unresolvedTransform(streamChunks(['visit Live'])))).toEqual([ + 'visit ', + 'Live', + ]); + }); + it('prefers longest overlapping key', async () => { const transform = replace({ a: 'X', ab: 'Y' }); expect(await collect(transform(streamText('ab', 100)))).toBe('Y'); }); + it('retains the source split-overlap fallback policy', async () => { + const transform = replace({ a: 'X', ab: 'Y' }); + expect(await collectChunks(transform(streamChunks(['a', 'b'])))).toEqual(['X', 'b']); + }); + it('does not cascade', async () => { const transform = replace({ a: 'b', b: 'c' }); expect(await collect(transform(streamText('a', 100)))).toBe('b'); }); + it('does not cascade replacement output across source chunks', async () => { + const transform = replace({ a: 'b', bc: 'X' }); + expect(await collectChunks(transform(streamChunks(['a', 'c'])))).toEqual(['b', 'c']); + }); + + it('replaces Unicode regex case equivalents', async () => { + const transform = replace({ S: 'ess', Σ: 'sigma' }); + expect(await collectChunks(transform(streamChunks(['Aſ', 'BςC'])))).toEqual([ + 'Aess', + 'BsigmaC', + ]); + }); + + it('uses source order for case-equivalent keys of equal length', async () => { + const transform = replace({ S: 'first', s: 'second' }); + expect(await collect(transform(streamChunks(['s'])))).toBe('first'); + }); + + it('handles UTF-16 splits without normalizing combining marks', async () => { + const transform = replace({ '😀x': 'smile', é: 'e' }); + expect( + await collectChunks(transform(streamChunks(['go \ud83d', '\ude00x; ', 'e\u0301']))), + ).toEqual(['go ', 'smile; ', 'e\u0301']); + }); + + it('propagates source errors without flushing an unresolved prefix', async () => { + const sourceError = new Error('source failed'); + let sourceController: ReadableStreamDefaultController | undefined; + const source = new ReadableStream({ + start(controller) { + sourceController = controller; + }, + }); + const reader = replace({ LiveKit: 'Lyve Kit' })(source).getReader(); + if (!sourceController) throw new Error('source controller was not initialized'); + + sourceController.enqueue('already emitted '); + expect(await reader.read()).toEqual({ done: false, value: 'already emitted ' }); + sourceController.enqueue('Live'); + await new Promise((resolve) => setImmediate(resolve)); + sourceController.error(sourceError); + + await expect(reader.read()).rejects.toBe(sourceError); + reader.releaseLock(); + }); + it('applies callable and built-in transforms', async () => { expect( await collect( diff --git a/agents/src/voice/transcription/text_transforms.ts b/agents/src/voice/transcription/text_transforms.ts index f36693761..aa657e18d 100644 --- a/agents/src/voice/transcription/text_transforms.ts +++ b/agents/src/voice/transcription/text_transforms.ts @@ -193,23 +193,14 @@ export function replace( options: { caseSensitive?: boolean } = {}, ): (text: ReadableStream) => ReadableStream { const entries = Object.entries(replacements); + const sortedEntries = [...entries].sort(([a], [b]) => b.length - a.length); const flags = options.caseSensitive ? 'u' : 'iu'; - const lookup = new Map( - entries.map(([old, replacement]) => [ - options.caseSensitive ? old : old.toLowerCase(), - replacement, - ]), + const entryPatterns = sortedEntries.map( + ([old, replacement]) => [new RegExp(`^(?:${escapeRegex(old)})$`, flags), replacement] as const, ); const pattern = entries.length > 0 - ? new RegExp( - entries - .map(([old]) => old) - .sort((a, b) => b.length - a.length) - .map(escapeRegex) - .join('|'), - `g${flags}`, - ) + ? new RegExp(sortedEntries.map(([old]) => escapeRegex(old)).join('|'), `g${flags}`) : null; const maxPrefix = entries.length > 0 ? Math.max(...entries.map(([old]) => old.length - 1)) : 0; const prefixes = new Set(); @@ -223,12 +214,19 @@ export function replace( ? new RegExp(`(?:${Array.from(prefixes).map(escapeRegex).join('|')})$`, flags) : null; - const apply = (value: string): string => { - if (!pattern) return value; - return value.replace( - pattern, - (match) => lookup.get(options.caseSensitive ? match : match.toLowerCase())!, - ); + const apply = (value: string): { output: string; lastMatchEnd: number } => { + if (!pattern) return { output: value, lastMatchEnd: 0 }; + let lastMatchEnd = 0; + const output = value.replace(pattern, (match: string, offset: number) => { + const entry = entryPatterns.find(([entryPattern]) => entryPattern.test(match)); + if (!entry) { + throw new Error(`Unable to resolve replacement for matched text: ${match}`); + } + const replacement = entry[1]; + lastMatchEnd = offset + match.length; + return replacement; + }); + return { output, lastMatchEnd }; }; const holdback = (value: string): number => { @@ -240,19 +238,23 @@ export function replace( return (text: ReadableStream) => streamFromAsyncIterable( (async function* () { - let buffer = ''; + let sourceBuffer = ''; for await (const chunk of readStream(text)) { - buffer = apply(buffer + chunk); - const flushTo = buffer.length - holdback(buffer); - if (flushTo > 0) { - yield buffer.slice(0, flushTo); - buffer = buffer.slice(flushTo); + const source = sourceBuffer + chunk; + const applied = apply(source); + const heldLength = holdback(source); + const heldStart = source.length - heldLength; + const retainSource = heldLength > 0 && heldStart >= applied.lastMatchEnd; + sourceBuffer = retainSource ? source.slice(heldStart) : ''; + const emitted = retainSource ? applied.output.slice(0, -heldLength) : applied.output; + if (emitted) { + yield emitted; } } - if (buffer) { - yield buffer; + if (sourceBuffer) { + yield sourceBuffer; } })(), );