fix(bedrock): forward each Nova Sonic assistant sentence once over the realtime API

litellm/llms/bedrock/realtime/transformation.py

@@ -118,6 +118,7 @@ def __init__(self):
         # so the USER/ASSISTANT split from contentStart is tracked here)
         self._user_transcript_active = False
         self._user_transcript_generation_stage: str | None = None
+        self._assistant_final_text_active = False
         self._user_item_id: str | None = None
         self._user_transcript_buffer = ""
         self._cumulative_usage = BedrockUsageEvent()
@@ -748,6 +749,20 @@ def transform_content_start_event(
                 None,
             )
 
+        generation_stage: Final = self._parse_generation_stage(content_start.get("additionalModelFields"))
+        if content_start.get("type") == "TEXT" and generation_stage == "FINAL":
+            verbose_logger.debug(
+                "Dropping FINAL ASSISTANT text block: the SPECULATIVE block already carried its sentences"
+            )
+            self._assistant_final_text_active = True
+            return (
+                [],
+                current_response_id,
+                current_output_item_id,
+                current_conversation_id,
+                None,
+            )
+
         verbose_logger.debug("Handling ASSISTANT contentStart")
         is_new_response: Final = current_response_id is None
 
@@ -943,6 +958,8 @@ def transform_text_output_event(
         verbose_logger.debug("Handling textOutput")
         text_content: Final = event["textOutput"].get("content", "")
 
+        if self._assistant_final_text_active:
+            return [], current_delta_chunks
         if not current_output_item_id or not current_response_id:
             return [], current_delta_chunks
 
@@ -1022,6 +1039,9 @@ def transform_content_end_event(
         content_end: Final = event["contentEnd"]
         verbose_logger.debug("Handling contentEnd: %s", content_end)
 
+        if self._assistant_final_text_active:
+            self._assistant_final_text_active = False
+            return [], current_delta_chunks
         if not current_output_item_id or not current_response_id:
             return [], current_delta_chunks
 

tests/e2e/llm_translation/realtime/realtime_client.py

@@ -11,17 +11,19 @@
 from __future__ import annotations
 
 import time
-from collections.abc import Generator, Mapping
+from collections.abc import Generator, Iterator, Mapping, Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass
-from typing import TypeVar
+from itertools import chain, repeat
+from typing import Final, TypeVar
 from urllib.parse import urlencode
 
 from pydantic import BaseModel, ConfigDict
 from websockets.sync.client import connect
 from websockets.sync.connection import Connection
 
 from e2e_config import unique_marker, ws_base_url
+from e2e_metadata import step
 from proxy_client import ProxyClient
 from models import LiteLLMParamsBody
 
@@ -164,6 +166,11 @@ class ResponseCreate(BaseModel):
     type: str = "response.create"
 
 
+class InputAudioBufferAppend(BaseModel):
+    type: str = "input_audio_buffer.append"
+    audio: str
+
+
 def user_message(text: str) -> ConversationItemCreate:
     return ConversationItemCreate(
         item=MessageItem(content=[InputTextContent(text=text)])
@@ -214,6 +221,12 @@ class ResponsePayload(BaseModel):
     output: list[OutputItem] | None = None
 
 
+class TextDone(BaseModel):
+    type: str
+    response_id: str
+    text: str
+
+
 class ResponseDone(BaseModel):
     type: str
     response: ResponsePayload
@@ -320,6 +333,50 @@ def collect_until(
         )
 
 
+    @step("Stream the spoken question as mic chunks, then silence until the turn goes idle")
+    def stream_and_collect(
+        self,
+        chunks: Sequence[InputAudioBufferAppend],
+        *,
+        tail: InputAudioBufferAppend,
+        interval: float,
+        idle: float,
+        timeout: float,
+    ) -> tuple[ReceivedEvent, ...]:
+        """Send `chunks`, then `tail` every `interval` seconds like a live mic, until
+        no event arrives for `idle` seconds or `timeout` elapses."""
+        return tuple(self._live_mic_events(chunks, tail=tail, interval=interval, idle=idle, timeout=timeout))
+
+    def _live_mic_events(
+        self,
+        chunks: Sequence[InputAudioBufferAppend],
+        *,
+        tail: InputAudioBufferAppend,
+        interval: float,
+        idle: float,
+        timeout: float,
+    ) -> Iterator[ReceivedEvent]:
+        start: Final = time.monotonic()
+        last_event = start  # rebind-ok: the idle timer restarts at every received event
+        for frame in chain(chunks, repeat(tail)):
+            if (now := time.monotonic()) - start >= timeout or now - last_event >= idle:
+                return
+            self.send(frame)
+            for event in self._events_until(now + interval):
+                last_event = time.monotonic()
+                yield event
+
+    def _events_until(self, deadline: float) -> Iterator[ReceivedEvent]:
+        while (remaining := deadline - time.monotonic()) > 0 and (text := self._recv_text(remaining)) is not None:
+            yield ReceivedEvent(type=ServerEnvelope.model_validate_json(text).type, payload=text)
+
+    def _recv_text(self, timeout: float) -> str | None:
+        try:
+            return as_text(self.connection.recv(timeout=timeout))
+        except TimeoutError:
+            return None
+
+
 @dataclass(frozen=True, slots=True)
 class RealtimeClient:
     proxy: ProxyClient

tests/e2e/llm_translation/realtime/test_realtime_bedrock_duplicate_text_e2e.py

@@ -0,0 +1,134 @@
+"""Live e2e: Bedrock Nova 2 Sonic realtime shows each assistant sentence once.
+
+User flow: a voice client on /v1/realtime streams mic audio
+through input_audio_buffer.append, keeps streaming after the user stops talking,
+and renders every response.text.done as a chat bubble. Nova 2 Sonic sends each
+assistant text block twice, a SPECULATIVE preview next to the audio and then the
+FINAL transcript once the audio turn has ended, tagging each block with
+generationStage in its contentStart. The test streams a spoken question the same
+way, reads the whole turn including events after the first response.done, and
+asserts no assistant sentence reaches the client in more than one
+response.text.done.
+
+NOVA_SONIC pins a vendor-owned model id; see test_realtime_bedrock_e2e.py for
+how to re-check it before concluding litellm broke.
+"""
+
+from __future__ import annotations
+
+import array
+import base64
+import wave
+from collections import Counter
+from pathlib import Path
+from typing import Final
+
+import pytest
+
+from e2e_config import unique_marker
+from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
+from lifecycle import ResourceManager
+from models import LiteLLMParamsBody
+from realtime_client import (
+    InputAudioBufferAppend,
+    RealtimeClient,
+    SessionConfig,
+    SessionUpdate,
+    TextDone,
+    events_of_type,
+)
+
+pytestmark = pytest.mark.e2e
+
+NOVA_SONIC = "bedrock/amazon.nova-2-sonic-v1:0"
+QUESTION_WAV = Path(__file__).parent / "fixtures" / "weather_question_24k.wav"
+NOVA_INPUT_RATE = 16000
+CHUNK_SECONDS = 0.1
+
+
+def pcm16_at_nova_input_rate(path: Path) -> bytes:
+    """Mono PCM16 samples from `path`, linearly resampled to Nova Sonic's 16 kHz input rate."""
+    with wave.open(str(path), "rb") as wav:
+        assert (wav.getnchannels(), wav.getsampwidth()) == (1, 2), "fixture must be mono PCM16"
+        ratio: Final = wav.getframerate() / NOVA_INPUT_RATE
+        source: Final = array.array("h", wav.readframes(wav.getnframes()))
+    positions: Final = (i * ratio for i in range(int(len(source) / ratio) - 1))
+    return array.array("h", (_interpolated_sample(source, position) for position in positions)).tobytes()
+
+
+def _interpolated_sample(source: array.array[int], position: float) -> int:
+    base: Final = int(position)
+    fraction: Final = position - base
+    return round(source[base] * (1 - fraction) + source[base + 1] * fraction)
+
+
+def mic_chunks(pcm: bytes) -> list[InputAudioBufferAppend]:
+    size = int(NOVA_INPUT_RATE * CHUNK_SECONDS) * 2
+    return [
+        InputAudioBufferAppend(audio=base64.b64encode(pcm[i : i + size]).decode()) for i in range(0, len(pcm), size)
+    ]
+
+
+class TestNovaSonicAssistantText:
+    @pytest.mark.covers(
+        "llm.realtime.bedrock_converse.basic.stream.works",
+        exercised_on=["realtime"],
+    )
+    @meta(
+        Subject(
+            domain=Domain.LLM_TRANSLATION,
+            route=Route.REALTIME,
+            providers=(Provider.BEDROCK,),
+            models=(NOVA_SONIC,),
+            capabilities=(Capability.AUDIO_INPUT, Capability.AUDIO_OUTPUT),
+            mode=Mode.WEBSOCKET,
+        )
+    )
+    def test_nova_sonic_voice_turn_shows_each_sentence_once(
+        self, client: RealtimeClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        model: Final = f"e2e-nova-sonic-dupe-{unique_marker()}"
+        model_id: Final = client.proxy.create_model(
+            model,
+            LiteLLMParamsBody(
+                model=NOVA_SONIC,
+                aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
+                aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
+                aws_region_name="os.environ/AWS_REGION",
+            ),
+            mode="realtime",
+        )
+        resources.defer(lambda: client.proxy.delete_model(model_id))
+        silence: Final = InputAudioBufferAppend(
+            audio=base64.b64encode(bytes(int(NOVA_INPUT_RATE * CHUNK_SECONDS) * 2)).decode()
+        )
+
+        with client.connect(key=scoped_key, model=model) as session:
+            session.collect_until("session.created", timeout=30)
+            session.send(
+                SessionUpdate(
+                    session=SessionConfig(
+                        modalities=["text", "audio"],
+                        instructions="You are a friendly assistant. Answer in two or three short sentences.",
+                    )
+                )
+            )
+            session.collect_until("session.updated", timeout=30)
+            events: Final = session.stream_and_collect(
+                mic_chunks(pcm16_at_nova_input_rate(QUESTION_WAV)),
+                tail=silence,
+                interval=CHUNK_SECONDS,
+                idle=15,
+                timeout=90,
+            )
+
+        types: Final = [e.type for e in events]
+        assert "response.done" in types, f"Nova Sonic never finished a response; types={types}"
+        spoken: Final = [
+            text
+            for e in events_of_type(events, "response.text.done")
+            if (text := TextDone.model_validate_json(e.payload).text.strip())
+        ]
+        assert spoken, f"Nova Sonic produced no assistant text; types={types}"
+        repeated: Final = {t: n for t, n in Counter(spoken).items() if n > 1}
+        assert not repeated, f"assistant text delivered more than once: {repeated}; all text.done={spoken}"

tests/unit/llms/bedrock/realtime/test_bedrock_realtime_transformation.py

@@ -1,4 +1,5 @@
 import json
+from typing import Final
 from unittest.mock import MagicMock
 
 import pytest
@@ -828,8 +829,9 @@ def test_session_updated_defaults_modalities_when_unspecified(self):
 
 
 class TestBedrockRealtimeUserEventsAndUsage:
-    """Regression tests for #38346: USER ASR transcripts, speech boundary events,
-    usage propagation, and duplicate response.created"""
+    """Regression tests for #38346 (USER ASR transcripts, speech boundary events,
+    usage propagation, duplicate response.created) and for FINAL assistant text
+    blocks replaying every sentence after the audio turn ended"""
 
     @staticmethod
     def _run(config, messages):
@@ -1131,6 +1133,84 @@ def test_speculative_transcript_block_end_emits_no_completed(self):
         )
         assert [e["type"] for e in events] == ["conversation.item.input_audio_transcription.delta"]
 
+    @staticmethod
+    def _assistant_text_block(stage: str, text: str, stop_reason: str) -> list[dict[str, object]]:
+        return [
+            {
+                "event": {
+                    "contentStart": {
+                        "role": "ASSISTANT",
+                        "type": "TEXT",
+                        "additionalModelFields": json.dumps({"generationStage": stage}),
+                    }
+                }
+            },
+            {"event": {"textOutput": {"content": text}}},
+            {"event": {"contentEnd": {"stopReason": stop_reason}}},
+        ]
+
+    @staticmethod
+    def _assistant_audio_block(stop_reason: str) -> list[dict[str, object]]:
+        return [
+            {"event": {"contentStart": {"role": "ASSISTANT", "type": "AUDIO"}}},
+            {"event": {"audioOutput": {"content": base64.b64encode(b"\x00\x01" * 8).decode()}}},
+            {"event": {"contentEnd": {"stopReason": stop_reason}}},
+        ]
+
+    def test_final_assistant_text_blocks_do_not_replay_the_spoken_sentences(self):
+        sentences: Final = ["I can't check live weather.", " Try a weather app.", " It will have the forecast."]
+        events: Final = self._run(
+            BedrockRealtimeConfig(),
+            self._assistant_text_block("SPECULATIVE", sentences[0], "PARTIAL_TURN")
+            + self._assistant_audio_block("PARTIAL_TURN")
+            + self._assistant_text_block("SPECULATIVE", sentences[1], "PARTIAL_TURN")
+            + self._assistant_audio_block("PARTIAL_TURN")
+            + self._assistant_text_block("SPECULATIVE", sentences[2], "PARTIAL_TURN")
+            + self._assistant_audio_block("END_TURN")
+            + self._assistant_text_block("FINAL", sentences[0], "END_TURN")
+            + self._assistant_text_block("FINAL", sentences[1], "PARTIAL_TURN")
+            + self._assistant_text_block("FINAL", sentences[2], "PARTIAL_TURN"),
+        )
+        types: Final = [e["type"] for e in events]
+        assert [e["text"] for e in events if e["type"] == "response.text.done"] == sentences
+        assert types.count("response.created") == 1
+        assert types.count("response.done") == 1
+        assert types[-1] == "response.done"
+        response_ids: Final = {e["response_id"] for e in events if e["type"] == "response.text.delta"}
+        assert response_ids == {events[0]["response"]["id"]}
+
+    def test_final_assistant_text_after_end_turn_opens_no_response(self):
+        events: Final = self._run(
+            BedrockRealtimeConfig(),
+            self._assistant_text_block("FINAL", "Already spoken.", "END_TURN"),
+        )
+        assert events == []
+
+    def test_speculative_text_after_a_final_block_is_forwarded(self):
+        config: Final = BedrockRealtimeConfig()
+        self._run(config, self._assistant_text_block("FINAL", "Already spoken.", "END_TURN"))
+        events: Final = self._run(config, self._assistant_text_block("SPECULATIVE", "Next turn.", "PARTIAL_TURN"))
+        assert [e["type"] for e in events][:1] == ["response.created"]
+        assert [e["text"] for e in events if e["type"] == "response.text.done"] == ["Next turn."]
+
+    def test_barge_in_marker_in_final_block_is_dropped_and_the_next_answer_still_flows(self):
+        events: Final = self._run(
+            BedrockRealtimeConfig(),
+            self._assistant_text_block("SPECULATIVE", "I can't check live weather.", "PARTIAL_TURN")
+            + self._assistant_audio_block("INTERRUPTED")
+            + self._assistant_text_block("FINAL", '{ "interrupted" : true }', "INTERRUPTED")
+            + self._assistant_text_block("FINAL", "I can't check live weather.", "INTERRUPTED")
+            + self._assistant_text_block("SPECULATIVE", "Still no live weather.", "PARTIAL_TURN")
+            + self._assistant_audio_block("END_TURN"),
+        )
+        types: Final = [e["type"] for e in events]
+        assert [e["text"] for e in events if e["type"] == "response.text.done"] == [
+            "I can't check live weather.",
+            "Still no live weather.",
+        ]
+        assert types.count("response.created") == 1
+        assert types.count("response.done") == 1
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])