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

* test(bedrock): live e2e asserting nova sonic realtime delivers each assistant sentence once

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(bedrock): drop ticket reference from nova sonic e2e docstring

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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

Nova 2 Sonic sends every assistant text block twice, a SPECULATIVE preview
next to the audio and a FINAL transcript once the audio turn has ended. The
realtime bridge forwarded both, so voice clients rendered each sentence twice
and the FINAL copies opened extra responses after response.done, the last of
which never closed. FINAL assistant text blocks are now dropped whole, so a
turn carries each sentence once inside the one response with its audio

* test(bedrock): tag the live Nova Sonic test and type its helpers

* test(bedrock): pin that the Nova Sonic barge-in marker is dropped with its FINAL block

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

litellm/llms/bedrock/realtime/transformation.py

@@ -118,6 +118,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
         # 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 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
                 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 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
         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 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
         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,10 +11,11 @@ models, matching the suite's no-raw-dicts rule.
 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
@@ -22,6 +23,7 @@ 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 @@ class RealtimeSession:
         )
 
 
+    @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 @@ class TestBedrockRealtimeSessionEvents:
 
 
 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 @@ class TestBedrockRealtimeUserEventsAndUsage:
         )
         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"])