From 71de2f3a1c4668750f67606b63f748050142de84 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 17 Aug 2026 10:18:07 +0100 Subject: [PATCH 1/4] feat: support multi-utterance persona messages The engine can split a single persona turn into multiple utterances, each tagged with an utterance_id. Message.content stays the full concatenated turn text for backward compatibility; Message.utterances now exposes an ordered, per-utterance breakdown so consumers can render each utterance as its own chat bubble. MessageStreamEvent also carries utterance_id per chunk. Both fields are omitted/None for engines that don't send utterance ids, and utterances are persona-only. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GoTTaKSGows6cTfFitX1ck --- src/anam/__init__.py | 2 + src/anam/client.py | 33 ++++++++++++++++ src/anam/types.py | 20 ++++++++++ tests/test_client.py | 92 ++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 147 insertions(+) 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..542389c 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 = msg_data.get("utterance_id") or None # 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,26 @@ 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 _append_utterance( + utterances: list[MessageUtterance] | None, event: MessageStreamEvent + ) -> list[MessageUtterance] | None: + """Fold a persona chunk into the message's per-utterance breakdown. + + The engine prepends a joining space to the first chunk of each new utterance so the + turn-level content concatenates correctly; it is not part of the utterance itself, so + it is stripped here. + """ + 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] + return previous + [MessageUtterance(id=event.utterance_id, content=event.content.lstrip())] + 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 +396,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..da2f8ad 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,97 @@ 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 + + class TestPersonaConfig: """Tests for PersonaConfig.""" From 27f54a897e89dc0260d2f6159a55d290c6c7b913 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 17 Aug 2026 10:35:15 +0100 Subject: [PATCH 2/4] fix: validate utterance_id is a string before use Mirrors _extract_correlation_id's type check so a non-string truthy value from the backend can't violate MessageStreamEvent's str | None contract. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GoTTaKSGows6cTfFitX1ck --- src/anam/client.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/anam/client.py b/src/anam/client.py index 542389c..4face82 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -313,7 +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 = msg_data.get("utterance_id") or None + 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}" @@ -365,6 +365,12 @@ 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 From 05dd6760b89fecd860d29ec10c51ffac0d5fe49a Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 17 Aug 2026 14:12:16 +0100 Subject: [PATCH 3/4] fix: preserve utterance leading whitespace Only the single joining space prepended before a subsequent utterance is stripped; the first utterance in a turn, and any other leading whitespace, is now preserved verbatim. Mirrors the JS SDK's a1b80df. Also adds a regression test confirming a previously published Message snapshot isn't mutated by later chunks for the same turn (the Python port already builds fresh Message/utterance objects on every update, so it never had the mutation bug the JS SDK introduced and fixed in its own perf pass). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GoTTaKSGows6cTfFitX1ck --- src/anam/client.py | 13 ++++++++---- tests/test_client.py | 49 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 4 deletions(-) diff --git a/src/anam/client.py b/src/anam/client.py index 4face82..f243a00 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -377,9 +377,9 @@ def _append_utterance( ) -> list[MessageUtterance] | None: """Fold a persona chunk into the message's per-utterance breakdown. - The engine prepends a joining space to the first chunk of each new utterance so the - turn-level content concatenates correctly; it is not part of the utterance itself, so - it is stripped here. + 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 @@ -389,7 +389,12 @@ def _append_utterance( id=previous[-1].id, content=previous[-1].content + event.content ) return previous[:-1] + [merged_last] - return previous + [MessageUtterance(id=event.utterance_id, content=event.content.lstrip())] + 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.""" diff --git a/tests/test_client.py b/tests/test_client.py index da2f8ad..b75f783 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -454,6 +454,55 @@ async def test_history_shape_unchanged_without_utterance_ids(self) -> None: 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.""" From dee888594b42bb0779fc52c2956d1c320f7770e8 Mon Sep 17 00:00:00 2001 From: sr-anam Date: Mon, 17 Aug 2026 17:33:30 +0100 Subject: [PATCH 4/4] lint --- src/anam/client.py | 6 +----- tests/test_client.py | 8 ++------ uv.lock | 2 +- 3 files changed, 4 insertions(+), 12 deletions(-) diff --git a/src/anam/client.py b/src/anam/client.py index f243a00..3012e52 100644 --- a/src/anam/client.py +++ b/src/anam/client.py @@ -389,11 +389,7 @@ def _append_utterance( 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 - ) + 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: diff --git a/tests/test_client.py b/tests/test_client.py index b75f783..44c4e47 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -401,9 +401,7 @@ async def test_missing_or_empty_utterance_id_omitted(self) -> None: 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="") - ) + 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 @@ -413,9 +411,7 @@ 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="Hel", utterance_id="uuid-a")) await client._handle_data_message( self._persona_chunk(content="lo.", content_index=1, utterance_id="uuid-a") ) 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" },