fix(tts): correct ttfb attribution (#1955)

Co-authored-by: rosetta-livekit-bot[bot] <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com>
Co-authored-by: David Zhao <dz@livekit.io>

Author: rosetta-livekit-bot[bot]

.changeset/fix-duplicate-user-turn-commit.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Fix duplicated user chat items in observability: a superseding EOU bounce created while an earlier bounce was mid-commit could fire after the transcript was already committed and cleared, committing a second empty user turn with stale metrics. The transcript guard is now re-checked at fire time. Also, session-report chat items now only upload string content (matching Python), so non-string content can no longer render as garbage in the dashboard.

.changeset/fix-tts-ttfb-attribution.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Fix TTS TTFB attribution: `tts_node` TTFB is now anchored on the time the first sentence is sent to the TTS provider instead of the time the first LLM token arrives, so upstream text generation and tokenization latency is no longer counted as TTS latency.

.changeset/neuphonic-sentence-tokenizer.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents-plugin-neuphonic': patch
+---
+
+Batch streamed text into sentences before sending it to the Neuphonic websocket (matching the Python plugin), with a configurable `tokenizer` option. This also anchors TTS TTFB on the first sentence sent to the provider instead of the first LLM token.

.changeset/trim-eou-span-chat-ctx.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Trim the chat context recorded on the `eou_detection` span to the last 6 items and exclude function calls, instructions, empty messages, handoffs, and config updates (matching Python), so the span no longer re-emits the whole conversation on every end-of-turn inference.

agents/src/inference/tts.ts

@@ -617,6 +617,7 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
         if (this.opts.model) generationConfig.model = this.opts.model;
         if (this.opts.language) generationConfig.language = this.opts.language;
 
+        this.markStarted();
         await sendClientEvent(
           {
             type: 'input_transcript',

agents/src/telemetry/traces.ts

@@ -410,9 +410,15 @@ function chatItemToProto(item: ChatItem): ProtoChatItem {
     const msg: ProtoMessage = {
       id: item.id,
       role: ROLE_MAP[item.role] ?? (item.role.toUpperCase() as ProtoRole),
-      content: item.content.map((c: ChatContent) => ({
-        text: isInstructions(c) ? c.value : c,
-      })),
+      // Match Python's `_build_proto_chat_item`: only string content is uploaded.
+      // Non-string content (image/audio) must not leak into the wire format —
+      // the ChatContent proto's `text` field is a string, and non-string values
+      // render as garbage in the dashboard.
+      content: item.content
+        .filter((c: ChatContent) => typeof c === 'string' || isInstructions(c))
+        .map((c) => ({
+          text: isInstructions(c) ? c.value : (c as string),
+        })),
       createdAt: toRFC3339(item.createdAt),
     };
 

agents/src/tts/fallback_adapter.test.ts

@@ -7,6 +7,7 @@ import { beforeAll, describe, expect, it } from 'vitest';
 import { APIError } from '../_exceptions.js';
 import { initializeLogger } from '../log.js';
 import type { APIConnectOptions } from '../types.js';
+import { USERDATA_TTS_STARTED_TIME } from '../types.js';
 import { FallbackAdapter } from './fallback_adapter.js';
 import { ChunkedStream, SynthesizeStream, TTS } from './tts.js';
 
@@ -23,8 +24,31 @@ class MockSynthesizeStream extends SynthesizeStream {
     super(mockTts, connOptions);
   }
 
+  // Simulate sending text to the provider, like a real plugin does right
+  // before ws.send(): optionally delay (sentence buffering / connection
+  // setup), then mark the started time and record it for assertions.
+  private async sendToProvider(): Promise<void> {
+    if (this.mockTts.sendDelayMs > 0) {
+      await new Promise((resolve) => setTimeout(resolve, this.mockTts.sendDelayMs));
+    }
+    this.markStarted();
+    this.mockTts.lastMarkedTime = this.startedTime?.time;
+  }
+
   protected async run(): Promise<void> {
     if (this.shouldFail) {
+      if (this.mockTts.failAfterInput) {
+        // Simulate a provider that receives text but dies before emitting
+        // any audio: the started time it recorded must still anchor the
+        // fallback adapter's TTFB.
+        for await (const data of this.input) {
+          if (this.abortController.signal.aborted) break;
+          if (data === SynthesizeStream.FLUSH_SENTINEL) continue;
+          await this.sendToProvider();
+          break;
+        }
+        throw new APIError('mock TTS failed after receiving input');
+      }
       // Throw immediately, before any pushText has been called.
       // This is the scenario that previously deadlocked the FallbackAdapter:
       // the inner stream's mainTask finishes before forwardBufferToTTS gets
@@ -37,6 +61,7 @@ class MockSynthesizeStream extends SynthesizeStream {
     for await (const data of this.input) {
       if (this.abortController.signal.aborted) break;
       if (data === SynthesizeStream.FLUSH_SENTINEL) continue;
+      await this.sendToProvider();
       this.queue.put({
         requestId: 'mock-req',
         segmentId: 'mock-seg',
@@ -73,6 +98,12 @@ class MockChunkedStream extends ChunkedStream {
 class MockTTS extends TTS {
   label: string;
   shouldFail = false;
+  /** When failing, first consume a token (and mark started) before throwing. */
+  failAfterInput = false;
+  /** Simulated latency between receiving text and sending it to the provider. */
+  sendDelayMs = 0;
+  /** The started time the stream recorded when it "sent" text to the provider. */
+  lastMarkedTime?: number;
 
   constructor(label: string, sampleRate: number = SAMPLE_RATE) {
     super(sampleRate, 1, { streaming: true });
@@ -227,4 +258,114 @@ describe('TTS FallbackAdapter', () => {
 
     await adapter.close();
   });
+
+  it('anchors ttfb on the time text was sent to the provider, not when it was pushed', async () => {
+    const primary = new MockTTS('primary');
+    // Simulate sentence buffering / connection latency between the text being
+    // pushed to the TTS node and it actually being sent to the provider.
+    primary.sendDelayMs = 120;
+    const adapter = new FallbackAdapter({
+      ttsInstances: [primary],
+      maxRetryPerTTS: 0,
+      recoveryDelayMs: 60_000,
+    });
+
+    const stream = adapter.stream();
+    const pushTime = performance.now() / 1000;
+    stream.updateInputStream(
+      new ReadableStream<string>({
+        start(controller) {
+          controller.enqueue('hello world');
+          controller.close();
+        },
+      }),
+    );
+
+    const startedTimes = new Set<unknown>();
+    for await (const event of stream) {
+      if (event === SynthesizeStream.END_OF_STREAM) break;
+      startedTimes.add(event.frame.userdata[USERDATA_TTS_STARTED_TIME]);
+    }
+
+    expect(startedTimes.size).toBe(1);
+    const startedTime = [...startedTimes][0];
+    // the stamp must be the exact time the underlying stream sent the text to
+    // the provider, so the send delay is excluded from downstream TTFB
+    expect(startedTime).toBe(primary.lastMarkedTime);
+    expect(startedTime as number).toBeGreaterThanOrEqual(pushTime + 0.1);
+
+    stream.close();
+    await adapter.close();
+  });
+
+  it('keeps the ttfb anchor from a provider that failed after receiving text', async () => {
+    const primary = new MockTTS('primary');
+    primary.shouldFail = true;
+    primary.failAfterInput = true;
+    const secondary = new MockTTS('secondary');
+    secondary.sendDelayMs = 50;
+    const adapter = new FallbackAdapter({
+      ttsInstances: [primary, secondary],
+      maxRetryPerTTS: 0,
+      recoveryDelayMs: 60_000,
+    });
+
+    const stream = adapter.stream();
+    stream.updateInputStream(
+      new ReadableStream<string>({
+        start(controller) {
+          controller.enqueue('hello world');
+          controller.close();
+        },
+      }),
+    );
+
+    const startedTimes = new Set<unknown>();
+    for await (const event of stream) {
+      if (event === SynthesizeStream.END_OF_STREAM) break;
+      startedTimes.add(event.frame.userdata[USERDATA_TTS_STARTED_TIME]);
+    }
+
+    // the fallback adapter is measured as a single TTS node: the anchor stays
+    // on the first provider that received the text — even though it failed
+    // before emitting audio — so failover time counts towards TTFB
+    expect(primary.lastMarkedTime).toBeDefined();
+    expect(secondary.lastMarkedTime).toBeDefined();
+    expect(startedTimes.size).toBe(1);
+    const startedTime = [...startedTimes][0];
+    expect(startedTime).toBe(primary.lastMarkedTime);
+    expect(startedTime as number).toBeLessThan(secondary.lastMarkedTime!);
+
+    stream.close();
+    await adapter.close();
+  });
+
+  it('stamps chunked synthesis with the submission time, kept across failover', async () => {
+    const primary = new MockTTS('primary');
+    primary.shouldFail = true;
+    const secondary = new MockTTS('secondary');
+    const adapter = new FallbackAdapter({
+      ttsInstances: [primary, secondary],
+      maxRetryPerTTS: 0,
+      recoveryDelayMs: 60_000,
+    });
+
+    const submitTime = performance.now() / 1000;
+    const chunked = adapter.synthesize('hello world');
+
+    const startedTimes = new Set<unknown>();
+    for await (const event of chunked) {
+      startedTimes.add(event.frame.userdata[USERDATA_TTS_STARTED_TIME]);
+    }
+
+    // the full text is submitted at creation time; failing over to the
+    // secondary must not move the anchor
+    expect(startedTimes.size).toBe(1);
+    const startedTime = [...startedTimes][0] as number;
+    expect(typeof startedTime).toBe('number');
+    expect(startedTime).toBeGreaterThanOrEqual(submitTime);
+    expect(startedTime).toBeLessThan(submitTime + 0.1);
+
+    await adapter.close();
+  });
 });

agents/src/tts/fallback_adapter.ts

@@ -452,6 +452,12 @@ class FallbackSynthesizeStream extends SynthesizeStream {
       }
       const resampler = this.adapter.createResamplerForTTS(i);
 
+      // ttfb measures the fallback adapter as a whole: anchor on the first
+      // time a sentence was handed to any underlying TTS — even one that
+      // failed before emitting audio — and never overwrite it when falling
+      // back to another TTS (markStarted only takes effect once).
+      let captureStartedTime: () => void = () => {};
+
       try {
         this._logger.debug({ tts: originalTts.label }, 'attempting TTS stream');
 
@@ -463,6 +469,13 @@ class FallbackSynthesizeStream extends SynthesizeStream {
         const stream = tts.stream({ connOptions });
         let bufferIndex = 0;
         let streamOutputCompleted = false;
+
+        captureStartedTime = () => {
+          const startedTime = stream.startedTime;
+          if (startedTime !== undefined) {
+            this.markStarted(startedTime);
+          }
+        };
         const forwardBufferToTTS = async () => {
           while (true) {
             while (bufferIndex < this.tokenBuffer.length) {
@@ -505,6 +518,7 @@ class FallbackSynthesizeStream extends SynthesizeStream {
               }
 
               sawRawAudio = true;
+              captureStartedTime();
 
               if (resampler) {
                 for (const frame of resampler.push(audio.frame)) {
@@ -593,6 +607,9 @@ class FallbackSynthesizeStream extends SynthesizeStream {
           throw error;
         }
       } finally {
+        // the stream may have received text and failed before emitting audio;
+        // its started time must still anchor the fallback's ttfb
+        captureStartedTime();
         resampler?.close();
       }
     }

agents/src/tts/stream_adapter.ts

@@ -108,6 +108,7 @@ export class StreamAdapterWrapper extends SynthesizeStream {
       prevTask: Task<void> | undefined,
       controller: AbortController,
     ) => {
+      this.markStarted();
       const audioStream = this.#tts.synthesize(token, this.connOptions, this.abortSignal);
 
       // wait for previous audio transcription to complete before starting