Skip to content

fix(aws): nest endpointingSensitivity under turnDetectionConfiguration for Nova Sonic 2 #6201

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
St-Luciferr wants to merge 1 commit into
base: main
Choose a base branch
from
+103 −14
Open

fix(aws): nest endpointingSensitivity under turnDetectionConfiguration for Nova Sonic 2 #6201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 28 additions & 14 deletions livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,9 +95,16 @@ class ToolConfiguration(BaseModel):
tools: list[Tool]


class TurnDetectionConfiguration(BaseModel):
endpointingSensitivity: TURN_DETECTION


class SessionStart(BaseModel):
inferenceConfiguration: InferenceConfiguration
endpointingSensitivity: TURN_DETECTION | None = "MEDIUM"
# Nova Sonic 1 used a flat field; Nova Sonic 2 requires it nested under
# turnDetectionConfiguration. Exactly one is populated per model.
endpointingSensitivity: TURN_DETECTION | None = None
turnDetectionConfiguration: TurnDetectionConfiguration | None = None


class InputTextContentStart(BaseModel):
Expand Down Expand Up @@ -227,9 +234,12 @@ class Event(BaseModel):


class SonicEventBuilder:
def __init__(self, prompt_name: str, audio_content_name: str):
def __init__(
self, prompt_name: str, audio_content_name: str, model: str = "amazon.nova-2-sonic-v1:0"
):
self.prompt_name = prompt_name
self.audio_content_name = audio_content_name
self._nova_sonic_2 = "nova-2-sonic" in model

@classmethod
def get_event_type(cls, json_data: dict) -> str:
Expand Down Expand Up @@ -366,19 +376,23 @@ def create_session_start_event(
temperature: float = 0.7,
endpointing_sensitivity: TURN_DETECTION | None = "MEDIUM",
) -> str:
event = Event(
event=SessionStartEvent(
sessionStart=SessionStart(
inferenceConfiguration=InferenceConfiguration(
maxTokens=max_tokens,
topP=top_p,
temperature=temperature,
),
endpointingSensitivity=endpointing_sensitivity,
)
)
inference = InferenceConfiguration(
maxTokens=max_tokens, topP=top_p, temperature=temperature
)
return event.model_dump_json(exclude_none=False)
if self._nova_sonic_2 and endpointing_sensitivity is not None:
session_start = SessionStart(
inferenceConfiguration=inference,
turnDetectionConfiguration=TurnDetectionConfiguration(
endpointingSensitivity=endpointing_sensitivity
),
)
else:
session_start = SessionStart(
inferenceConfiguration=inference,
endpointingSensitivity=endpointing_sensitivity,
)
Comment on lines +382 to +393

@devin-ai-integration devin-ai-integration Bot Jun 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🚩 Nova Sonic 2 with endpointing_sensitivity=None falls through to the legacy code path

When self._nova_sonic_2 is True but endpointing_sensitivity is None (line 382), the code falls to the else branch which sets endpointingSensitivity=None on SessionStart. With exclude_none=True, both endpointingSensitivity and turnDetectionConfiguration will be excluded from the JSON, resulting in a session start with no turn detection config at all. This is likely the intended behavior for disabling turn detection, but it's worth confirming with the Nova Sonic 2 API docs that omitting both fields is valid and doesn't cause a ValidationException.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

event = Event(event=SessionStartEvent(sessionStart=session_start))
return event.model_dump_json(exclude_none=True) # was exclude_none=False

def create_audio_content_start_event(
self,
Expand Down
2 changes: 2 additions & 0 deletions ...t-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/realtime_model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -511,6 +511,7 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
self._event_builder = seb(
prompt_name=str(uuid.uuid4()),
audio_content_name=str(uuid.uuid4()),
model=self._realtime_model._model,
)
self._input_resampler: rtc.AudioResampler | None = None
self._bstream = utils.audio.AudioByteStream(
Expand Down Expand Up @@ -752,6 +753,7 @@ async def _graceful_session_recycle(self) -> None:
self._event_builder = seb(
prompt_name=str(uuid.uuid4()),
audio_content_name=str(uuid.uuid4()),
model=self._realtime_model._model,
)
self._tool_results_ch = utils.aio.Chan[dict[str, str]]()
self._audio_input_chan = utils.aio.Chan[bytes]()
Expand Down
73 changes: 73 additions & 0 deletions livekit-plugins/livekit-plugins-aws/tests/test_nova_sonic_turn_detection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
"""
Regression tests for Nova Sonic turn-detection serialization in sessionStart.

Nova Sonic 2 (amazon.nova-2-sonic-v1:0) rejects the sessionStart event with a
ValidationException unless the turn-detection setting is nested under
turnDetectionConfiguration. Nova Sonic 1 (amazon.nova-sonic-v1:0) predates
controllable endpointing and uses the legacy flat endpointingSensitivity field.

SonicEventBuilder serializes model-aware: Nova 2 (including cross-region
inference-profile ids such as us.amazon.nova-2-sonic-v1:0) → nested form;
Nova 1 → flat form.
"""

import json
import sys
from typing import Literal
from unittest.mock import MagicMock

# ---------------------------------------------------------------------------
# Stub out the optional AWS Smithy/Bedrock SDK not installed in the base venv.
# Importing the realtime package pulls in realtime_model, which imports the SDK.
# ---------------------------------------------------------------------------
_AWS_STUBS = [
"aws_sdk_bedrock_runtime",
"aws_sdk_bedrock_runtime.client",
"aws_sdk_bedrock_runtime.models",
"aws_sdk_bedrock_runtime.config",
"smithy_aws_core",
"smithy_aws_core.identity",
"smithy_aws_event_stream",
"smithy_aws_event_stream.exceptions",
"smithy_core",
"smithy_core.aio",
"smithy_core.aio.interfaces",
"smithy_core.aio.interfaces.identity",
]
for _mod in _AWS_STUBS:
if _mod not in sys.modules:
sys.modules[_mod] = MagicMock()


def _session_start(model: str, sensitivity: Literal["HIGH", "MEDIUM", "LOW"] = "HIGH") -> dict:
from livekit.plugins.aws.experimental.realtime.events import SonicEventBuilder

builder = SonicEventBuilder("prompt-name", "audio-content-name", model=model)
payload = builder.create_session_start_event(endpointing_sensitivity=sensitivity)
return json.loads(payload)["event"]["sessionStart"]


class TestSessionStartTurnDetection:
"""Model-aware serialization of the turn-detection setting."""
Comment on lines +50 to +51

@devin-ai-integration devin-ai-integration Bot Jun 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Test module missing mandatory pytestmark category marker

AGENTS.md mandates that every test module declares a category marker (pytestmark = pytest.mark.unit, etc.): "Adding a test: give the new module a category marker (pytestmark = pytest.mark.unit, etc.) — collection fails with a hint if it lacks one." The new test file test_nova_sonic_turn_detection.py has no pytestmark declaration. This is a pure unit test (no network/credentials needed) and should be marked pytest.mark.unit.

Prompt for agents 
Add a module-level pytestmark to the test file. Since this test is a fast, hermetic unit test with no external dependencies (it even mocks out the AWS SDK), it should be marked as a unit test.

Add the following near the top of the file (after the imports, before the helper function):

import pytest
pytestmark = pytest.mark.unit

This satisfies the AGENTS.md rule requiring all test modules to declare a category marker.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


def test_nova_sonic_2_nests_under_turn_detection_configuration(self):
ss = _session_start("amazon.nova-2-sonic-v1:0")

assert ss["turnDetectionConfiguration"]["endpointingSensitivity"] == "HIGH"
# the legacy flat field must not be emitted for Nova 2
assert "endpointingSensitivity" not in ss

def test_nova_sonic_1_keeps_flat_field(self):
ss = _session_start("amazon.nova-sonic-v1:0")

assert ss["endpointingSensitivity"] == "HIGH"
# the nested field must not be emitted for Nova 1
assert "turnDetectionConfiguration" not in ss

def test_cross_region_inference_profile_uses_nested_form(self):
# Bedrock cross-region inference profiles prefix the model id with a
# region group (us./eu./apac.); these are still Nova 2 and must nest.
ss = _session_start("us.amazon.nova-2-sonic-v1:0")

assert ss["turnDetectionConfiguration"]["endpointingSensitivity"] == "HIGH"
assert "endpointingSensitivity" not in ss
Loading