diff --git a/src/anam/__init__.py b/src/anam/__init__.py index 7384389..8dd4fa2 100644 --- a/src/anam/__init__.py +++ b/src/anam/__init__.py @@ -65,6 +65,7 @@ async def consume_audio(): Message, MessageRole, MessageStreamEvent, + MessageUtterance, PersonaConfig, SessionOptions, SessionReplayOptions, @@ -89,6 +90,7 @@ async def consume_audio(): "Message", "MessageRole", "MessageStreamEvent", + "MessageUtterance", "PersonaConfig", "SessionOptions", "SessionReplayOptions", diff --git a/src/anam/client.py b/src/anam/client.py index 484cb68..3012e52 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -26,6 +26,7 @@ Message, MessageRole, MessageStreamEvent, + MessageUtterance, PersonaConfig, SessionInfo, SessionOptions, @@ -312,6 +313,7 @@ async def _handle_data_message(self, data: dict[str, Any]) -> None: end_of_speech = msg_data.get("end_of_speech", False) interrupted = msg_data.get("interrupted", False) timestamp = msg_data.get("timestamp", "") + utterance_id = self._extract_utterance_id(msg_data) # Create message ID similar to JS SDK: "{role}::{message_id}" stream_event_id = f"{role_str}::{message_id}" @@ -333,6 +335,7 @@ async def _handle_data_message(self, data: dict[str, Any]) -> None: end_of_speech=end_of_speech, interrupted=interrupted, correlation_id=correlation_id, + utterance_id=utterance_id, ) await self._emit(AnamEvent.MESSAGE_STREAM_EVENT_RECEIVED, stream_event) @@ -362,6 +365,33 @@ def _extract_correlation_id(data: dict[str, Any]) -> str | None: correlation_id = data.get("user_action_correlation_id") or data.get("correlationId") return correlation_id if isinstance(correlation_id, str) else None + @staticmethod + def _extract_utterance_id(data: dict[str, Any]) -> str | None: + """Extract a persona utterance id from backend event payloads.""" + utterance_id = data.get("utterance_id") + return utterance_id if isinstance(utterance_id, str) and utterance_id else None + + @staticmethod + def _append_utterance( + utterances: list[MessageUtterance] | None, event: MessageStreamEvent + ) -> list[MessageUtterance] | None: + """Fold a persona chunk into the message's per-utterance breakdown. + + A joining space is prepended to each subsequent utterance so the turn-level content + concatenates correctly; remove only that separator while preserving all other leading + whitespace (including on the first utterance of a turn). + """ + if not event.utterance_id: + return utterances + previous = utterances or [] + if previous and previous[-1].id == event.utterance_id: + merged_last = MessageUtterance( + id=previous[-1].id, content=previous[-1].content + event.content + ) + return previous[:-1] + [merged_last] + content = event.content[1:] if previous and event.content.startswith(" ") else event.content + return previous + [MessageUtterance(id=event.utterance_id, content=content)] + def _process_message_stream_event(self, event: MessageStreamEvent, timestamp: str) -> None: """Process a message stream event and update message history.""" # Find existing message with same ID (for both user and persona messages) @@ -373,21 +403,31 @@ def _process_message_stream_event(self, event: MessageStreamEvent, timestamp: st if existing_index is not None: # Update existing message by appending new content existing = self._message_history[existing_index] + utterances = ( + self._append_utterance(existing.utterances, event) + if event.role == MessageRole.ASSISTANT + else existing.utterances + ) self._message_history[existing_index] = Message( id=existing.id, role=existing.role, content=existing.content + event.content, timestamp=existing.timestamp or timestamp, interrupted=existing.interrupted or event.interrupted, + utterances=utterances, ) else: # Add new message (first chunk) + utterances = ( + self._append_utterance(None, event) if event.role == MessageRole.ASSISTANT else None + ) new_message = Message( id=event.id, role=event.role, content=event.content, timestamp=timestamp, interrupted=event.interrupted, + utterances=utterances, ) self._message_history.append(new_message) diff --git a/src/anam/types.py b/src/anam/types.py index 2849380..ed070a9 100644 --- a/src/anam/types.py +++ b/src/anam/types.py @@ -296,6 +296,19 @@ class ClientOptions: environment: dict[str, str] | None = None +@dataclass +class MessageUtterance: + """A single utterance (speech bubble) within a persona message. + + Attributes: + id: Id of the persona utterance, matching MessageStreamEvent.utterance_id. + content: The text content of this utterance. + """ + + id: str + content: str + + @dataclass class Message: """A message in the conversation. @@ -306,6 +319,8 @@ class Message: content: The text content of the message. timestamp: When the message was sent (ISO format). interrupted: Whether the message was interrupted (for persona messages). + utterances: Per-utterance breakdown of content, in speaking order. Persona-only; + present only when the engine sends utterance ids. """ id: str @@ -313,6 +328,7 @@ class Message: content: str timestamp: str = "" interrupted: bool = False + utterances: list[MessageUtterance] | None = None @dataclass @@ -331,6 +347,9 @@ class MessageStreamEvent: interrupted: Whether the message was interrupted (for persona messages). correlation_id: Correlation ID for this turn, matching early user speech events when provided by the backend. + utterance_id: Id of the persona utterance this chunk belongs to. Chunks sharing an + utterance_id form one utterance; a turn can contain several. None on user chunks + and older engines. """ id: str @@ -340,6 +359,7 @@ class MessageStreamEvent: end_of_speech: bool interrupted: bool = False correlation_id: str | None = None + utterance_id: str | None = None @dataclass diff --git a/tests/test_client.py b/tests/test_client.py index 9000769..44c4e47 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -16,6 +16,7 @@ EgressOptions, MessageRole, MessageStreamEvent, + MessageUtterance, PersonaConfig, Session, SessionOptions, @@ -363,6 +364,142 @@ async def test_speech_text_exposes_correlation_id_on_stream_event(self) -> None: assert stream_event.correlation_id == "corr-123" +class TestAnamClientUtterances: + """Tests for multi-utterance persona message handling.""" + + @staticmethod + def _persona_chunk(**overrides: Any) -> dict[str, Any]: + chunk = { + "message_id": "turn-1", + "content_index": 0, + "content": "Hello", + "role": "persona", + "end_of_speech": False, + "interrupted": False, + } + chunk.update(overrides) + return {"messageType": "speechText", "data": chunk} + + @pytest.mark.asyncio + async def test_utterance_id_exposed_on_stream_event(self) -> None: + """A persona chunk's utterance_id is passed through onto the stream event.""" + client = AnamClient(api_key="test-key", persona_id="test-persona") + handler = AsyncMock() + client.add_listener(AnamEvent.MESSAGE_STREAM_EVENT_RECEIVED, handler) + + await client._handle_data_message(self._persona_chunk(utterance_id="uuid-a")) + + stream_event = handler.await_args.args[0] + assert stream_event.utterance_id == "uuid-a" + assert stream_event.id == "persona::turn-1" + + @pytest.mark.asyncio + async def test_missing_or_empty_utterance_id_omitted(self) -> None: + """Both a missing and an empty utterance_id collapse to None on the stream event.""" + client = AnamClient(api_key="test-key", persona_id="test-persona") + handler = AsyncMock() + client.add_listener(AnamEvent.MESSAGE_STREAM_EVENT_RECEIVED, handler) + + await client._handle_data_message(self._persona_chunk()) + await client._handle_data_message(self._persona_chunk(content_index=1, utterance_id="")) + + assert handler.await_args_list[0].args[0].utterance_id is None + assert handler.await_args_list[1].args[0].utterance_id is None + + @pytest.mark.asyncio + async def test_history_splits_utterances_keeps_turn_shape(self) -> None: + """Consecutive chunks merge per-utterance; a new utterance id starts a new entry.""" + client = AnamClient(api_key="test-key", persona_id="test-persona") + + await client._handle_data_message(self._persona_chunk(content="Hel", utterance_id="uuid-a")) + await client._handle_data_message( + self._persona_chunk(content="lo.", content_index=1, utterance_id="uuid-a") + ) + # New utterance in the same turn: the engine prepends the joining space, which must + # survive verbatim in turn-level content but is stripped from the utterance itself. + await client._handle_data_message( + self._persona_chunk( + content=" Second thought.", + content_index=2, + utterance_id="uuid-b", + end_of_speech=True, + ) + ) + + history = client.get_message_history() + assert len(history) == 1 + message = history[0] + assert message.content == "Hello. Second thought." + assert message.utterances == [ + MessageUtterance(id="uuid-a", content="Hello."), + MessageUtterance(id="uuid-b", content="Second thought."), + ] + + @pytest.mark.asyncio + async def test_history_shape_unchanged_without_utterance_ids(self) -> None: + """Messages built from chunks with no utterance_id keep utterances as None.""" + client = AnamClient(api_key="test-key", persona_id="test-persona") + + await client._handle_data_message(self._persona_chunk(content="Hello")) + await client._handle_data_message( + self._persona_chunk(content=" there.", content_index=1, end_of_speech=True) + ) + + history = client.get_message_history() + assert len(history) == 1 + assert history[0].content == "Hello there." + assert history[0].utterances is None + + @pytest.mark.asyncio + async def test_history_preserves_non_separator_whitespace(self) -> None: + """Only a single leading joining space on non-first utterances is stripped.""" + client = AnamClient(api_key="test-key", persona_id="test-persona") + + await client._handle_data_message( + self._persona_chunk(content=" Leading", utterance_id="uuid-a") + ) + await client._handle_data_message( + self._persona_chunk( + content=" Second", + content_index=1, + utterance_id="uuid-b", + end_of_speech=True, + ) + ) + + history = client.get_message_history() + assert history[0].content == " Leading Second" + assert history[0].utterances == [ + MessageUtterance(id="uuid-a", content=" Leading"), + MessageUtterance(id="uuid-b", content=" Second"), + ] + + @pytest.mark.asyncio + async def test_published_message_is_not_mutated_by_later_chunks(self) -> None: + """A Message snapshot handed to consumers stays frozen once later chunks arrive.""" + client = AnamClient(api_key="test-key", persona_id="test-persona") + + await client._handle_data_message( + self._persona_chunk(content="Hello", utterance_id="uuid-a", end_of_speech=True) + ) + published_message = client.get_message_history()[0] + + await client._handle_data_message( + self._persona_chunk( + content=" again", + content_index=1, + utterance_id="uuid-a", + end_of_speech=True, + ) + ) + + assert published_message.content == "Hello" + assert published_message.utterances == [MessageUtterance(id="uuid-a", content="Hello")] + history = client.get_message_history() + assert history[0].content == "Hello again" + assert history[0].utterances == [MessageUtterance(id="uuid-a", content="Hello again")] + + class TestPersonaConfig: """Tests for PersonaConfig.""" diff --git a/uv.lock b/uv.lock index 6ab357c..e5f5934 100644 --- a/uv.lock +++ b/uv.lock @@ -188,7 +188,7 @@ wheels = [ [[package]] name = "anam" -version = "0.9.0a2" +version = "0.10.0" source = { editable = "." } dependencies = [ { name = "aiohttp" },