feat: support multi-utterance persona messages

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",

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)
 

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,13 +319,16 @@ 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
     role: MessageRole
     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

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."""