Preserve text_format through HttpStream's intermediate and final activities

examples/stream/README.md

@@ -4,6 +4,7 @@ A test application that demonstrates streaming functionality.
 
 - Send any message for the normal single-stream demo with suggested actions.
 - Send `simple-card` to send a minimal Adaptive Card outside the streaming flow.
+- Send `extended-markdown` to stream a message where every chunk sets `text_format="extendedmarkdown"` — task-list checkboxes and strikethrough render as they stream, which plain markdown can't do. The leading informative update uses the new `stream.update(text, "markdown")` overload.
 - Send `multi-stream` to test emitting an Adaptive Card as part of the first stream final message, finalizing with `close()`, and then reusing `ctx.stream` for another streamed response.
 
 ## Running

examples/stream/src/main.py

@@ -44,12 +44,25 @@
     "[stream 2] The app processor will close this stream when the handler returns.",
 ]
 
+EXTENDED_MARKDOWN_MESSAGES = [
+    "**Extended markdown stream** — rendering features plain markdown can't:\n\n",
+    '- [x] Sent with `text_format="extendedmarkdown"`\n',
+    "- [x] Task list items render as real checkboxes\n",
+    "- [ ] ~~Under plain markdown these would be literal `[ ]` text~~\n",
+    "- [x] Strikethrough renders too\n",
+]
+
 
 def should_run_multi_stream(text: str | None) -> bool:
     normalized = (text or "").lower().replace("-", " ")
     return "multi stream" in normalized
 
 
+def should_run_extended_markdown(text: str | None) -> bool:
+    normalized = (text or "").lower().replace("-", " ")
+    return "extended markdown" in normalized or "extendedmarkdown" in normalized
+
+
 def should_send_simple_card(text: str | None) -> bool:
     normalized = (text or "").lower().replace("-", " ")
     return "simple card" in normalized
@@ -101,6 +114,15 @@ async def handle_message(ctx: ActivityContext[MessageActivity]):
             ctx.stream.emit(message)
         return
 
+    if should_run_extended_markdown(ctx.activity.text):
+        ctx.stream.update("Starting the *extended* markdown stream...", "markdown")
+        await asyncio.sleep(1)
+
+        for message in EXTENDED_MARKDOWN_MESSAGES:
+            await asyncio.sleep(0.5)
+            ctx.stream.emit(MessageActivityInput(text=message).with_text_format("extendedmarkdown"))
+        return
+
     ctx.stream.update("Stream starting...")
     await asyncio.sleep(1)
 

packages/api/src/microsoft_teams/api/activities/typing.py

@@ -5,7 +5,7 @@
 
 from typing import Literal, Optional, Self
 
-from ..models import ActivityBase, ActivityInputBase, ChannelData, CustomBaseModel, StreamInfoEntity
+from ..models import ActivityBase, ActivityInputBase, ChannelData, CustomBaseModel, StreamInfoEntity, TextFormat
 
 
 class _TypingBase(CustomBaseModel):
@@ -18,6 +18,12 @@ class _TypingBase(CustomBaseModel):
     The text content of the message.
     """
 
+    text_format: Optional[TextFormat] = None
+    """
+    Format of the `text` field (ex. `'extendedmarkdown'`). Omit or leave as ``None``
+    to use the Teams default (``'markdown'``).
+    """
+
 
 class TypingActivity(_TypingBase, ActivityBase):
     """Output model for received typing activities with required fields and read-only properties."""
@@ -31,6 +37,11 @@ def with_text(self, value: str) -> Self:
         self.text = value
         return self
 
+    def with_text_format(self, value: TextFormat) -> Self:
+        """Set the format of the `text` field."""
+        self.text_format = value
+        return self
+
     def add_text(self, text: str) -> Self:
         """Append text."""
         if self.text is None:

packages/api/tests/unit/test_typing.py

@@ -43,3 +43,16 @@ def test_should_build_with_text(self, user: Account, bot: Account, chat: Convers
         )
         assert activity.type == "typing"
         assert activity.text == "testing123"
+
+    def test_should_build_with_text_format(self, user: Account, bot: Account, chat: ConversationAccount) -> None:
+        """Test activity construction with text_format set via the builder."""
+        activity = TypingActivityInput(id="1", from_=user, conversation=chat, recipient=bot).with_text_format(
+            "extendedmarkdown"
+        )
+        assert activity.type == "typing"
+        assert activity.text_format == "extendedmarkdown"
+
+    def test_text_format_defaults_to_none(self, user: Account, bot: Account, chat: ConversationAccount) -> None:
+        """Test that text_format is unset unless explicitly provided."""
+        activity = TypingActivityInput(id="1", from_=user, conversation=chat, recipient=bot)
+        assert activity.text_format is None

packages/apps/src/microsoft_teams/apps/http_stream.py

@@ -15,6 +15,7 @@
     ConversationReference,
     MessageActivityInput,
     SentActivity,
+    TextFormat,
     TypingActivityInput,
 )
 from microsoft_teams.common import EventEmitter
@@ -147,14 +148,19 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str])
             # Schedule a flush immediately when no timeout is set (first emit)
             self._pending = asyncio.create_task(self._flush())
 
-    def update(self, text: str) -> None:
+    def update(self, text: str, text_format: Optional[TextFormat] = None) -> None:
         """
         Send status updates before emitting (ex. "Thinking...").
 
         Args:
             text: The status text to send.
+            text_format: Format of ``text`` (ex. ``'extendedmarkdown'``). Omit or pass
+                ``None`` to use the Teams default (``'markdown'``).
         """
-        self.emit(TypingActivityInput().with_text(text).with_channel_data(ChannelData(stream_type="informative")))
+        typing_update = TypingActivityInput().with_text(text).with_channel_data(ChannelData(stream_type="informative"))
+        if text_format is not None:
+            typing_update.with_text_format(text_format)
+        self.emit(typing_update)
 
     def clear_text(self) -> None:
         """
@@ -323,13 +329,19 @@ async def _flush(self) -> None:
             if self._timed_out:
                 return
 
-            # Send informative updates immediately
+            # Streamed text chunks use last-emitted-message-wins for text_format (same as
+            # attachments/entities/etc.), so they render like the final message.
+            text_format = self._final_activity.text_format if self._final_activity else None
+
+            # Send informative updates immediately. Each carries its own text_format, so keep the update's own value.
             for typing_update in informative_updates:
                 await self._send_activity(typing_update)
 
             # Send the combined text chunk
             if self._text:
                 to_send = TypingActivityInput(text=self._text)
+                if text_format:
+                    to_send.with_text_format(text_format)
                 await self._send_activity(to_send)
 
             # If more queued, schedule another flush

packages/apps/src/microsoft_teams/apps/plugins/streamer.py

@@ -6,7 +6,7 @@
 import asyncio
 from typing import Awaitable, Callable, Literal, Optional, Protocol, Union
 
-from microsoft_teams.api import MessageActivityInput, SentActivity, TypingActivityInput
+from microsoft_teams.api import MessageActivityInput, SentActivity, TextFormat, TypingActivityInput
 
 StreamerEvent = Literal["chunk", "close"]
 
@@ -91,12 +91,14 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str])
         """
         ...
 
-    def update(self, text: str) -> None:
+    def update(self, text: str, text_format: Optional[TextFormat] = None) -> None:
         """
         Send status updates before emitting (ex. "Thinking...").
 
         Args:
             text: The status text to send.
+            text_format: Format of ``text`` (ex. ``'extendedmarkdown'``). Omit or pass
+                ``None`` to use the Teams default (``'markdown'``).
         """
         ...
 

packages/apps/tests/test_http_stream.py

@@ -14,6 +14,7 @@
     ApiClient,
     CardAction,
     CardActionType,
+    ChannelData,
     ConversationAccount,
     ConversationReference,
     MessageActivityInput,
@@ -600,6 +601,161 @@ async def mock_send(*args):
             assert len(result.activity_params.suggested_actions.actions) == 2
             assert result.activity_params.suggested_actions.actions[0].title == "Option A"
 
+    @pytest.mark.asyncio
+    async def test_text_format_retained_on_intermediate_chunks_and_final_message(
+        self, mock_api_client, conversation_reference, patch_loop_call_later
+    ):
+        """Last emitted message's text_format applies to intermediate typing chunks and the
+        final message (matches the attachments/entities/suggested_actions last-wins behavior)."""
+        loop = asyncio.get_running_loop()
+        patcher, scheduled = patch_loop_call_later(loop)
+        with patcher:
+            stream = HttpStream(mock_api_client, conversation_reference)
+
+            # First message carries no text_format.
+            stream.emit(MessageActivityInput(text="hello "))
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            # A later message sets text_format; last-message-wins semantics apply.
+            stream.emit(MessageActivityInput(text="world").with_text_format("extendedmarkdown"))
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            result = await stream.close()
+
+            sent = mock_api_client.sent_activities
+            # First intermediate chunk (sent before text_format was ever emitted) has none.
+            assert sent[0].type == "typing"
+            assert sent[0].text_format is None
+
+            # Second intermediate chunk, sent after the text_format-carrying message, retains it.
+            assert sent[1].type == "typing"
+            assert sent[1].text_format == "extendedmarkdown"
+
+            # Final message also carries the last emitted text_format.
+            assert result is not None
+            assert result.activity_params.type == "message"
+            assert result.activity_params.text_format == "extendedmarkdown"
+            assert result.activity_params.text == "hello world"
+
+    @pytest.mark.asyncio
+    async def test_informative_update_sources_its_own_text_format(
+        self, mock_api_client, conversation_reference, patch_loop_call_later
+    ):
+        """An informative update keeps its own text_format and is NOT overwritten by the
+        last emitted message's format (the previous behavior incorrectly applied
+        _final_activity.text_format to informative updates)."""
+        loop = asyncio.get_running_loop()
+        patcher, scheduled = patch_loop_call_later(loop)
+        with patcher:
+            stream = HttpStream(mock_api_client, conversation_reference)
+
+            # Informative update with its own text_format, plus a message carrying a DIFFERENT
+            # format in the same flush cycle. The informative chunk must keep its own value.
+            stream.emit(
+                TypingActivityInput()
+                .with_text("Checking the release status...")
+                .with_channel_data(ChannelData(stream_type="informative"))
+                .with_text_format("extendedmarkdown")
+            )
+            stream.emit(MessageActivityInput(text="body").with_text_format("markdown"))
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            sent = mock_api_client.sent_activities
+            informative = [a for a in sent if getattr(a.channel_data, "stream_type", None) == "informative"]
+            assert len(informative) == 1
+            assert informative[0].type == "typing"
+            assert informative[0].text == "Checking the release status..."
+            assert informative[0].text_format == "extendedmarkdown"
+
+    @pytest.mark.asyncio
+    async def test_update_with_text_format_sends_informative_chunk_with_that_format(
+        self, mock_api_client, conversation_reference, patch_loop_call_later
+    ):
+        """update(text, text_format) sends an informative typing chunk carrying that format."""
+        loop = asyncio.get_running_loop()
+        patcher, scheduled = patch_loop_call_later(loop)
+        with patcher:
+            stream = HttpStream(mock_api_client, conversation_reference)
+            stream.update("Thinking...", "extendedmarkdown")
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            sent = mock_api_client.sent_activities
+            assert sent[0].type == "typing"
+            assert sent[0].channel_data is not None
+            assert sent[0].channel_data.stream_type == "informative"
+            assert sent[0].text_format == "extendedmarkdown"
+
+    @pytest.mark.asyncio
+    async def test_update_without_text_format_omits_it(
+        self, mock_api_client, conversation_reference, patch_loop_call_later
+    ):
+        """update(text) and update(text, None) omit text_format (Teams default: markdown)."""
+        loop = asyncio.get_running_loop()
+        patcher, scheduled = patch_loop_call_later(loop)
+        with patcher:
+            stream = HttpStream(mock_api_client, conversation_reference)
+            stream.update("no format")
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+            stream.update("explicit none", None)
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            sent = mock_api_client.sent_activities
+            assert sent[0].type == "typing"
+            assert sent[0].text_format is None
+            assert sent[1].text_format is None
+
+    @pytest.mark.asyncio
+    async def test_final_send_timeout_retains_text_format(
+        self, mock_api_client, conversation_reference, patch_loop_call_later
+    ):
+        """The timeout fallback (sendFinal-equivalent) still carries the last emitted text_format
+        since it reuses the same buffered final activity."""
+        create_calls = 0
+        updates: list[dict] = []
+        loop = asyncio.get_running_loop()
+        patcher, scheduled = patch_loop_call_later(loop)
+        with patcher:
+
+            async def mock_create(conversation_id, activity):
+                nonlocal create_calls
+                create_calls += 1
+                if create_calls == 2:
+                    raise HTTPStatusError(
+                        "Forbidden",
+                        request=Request("POST", "https://example.com"),
+                        response=Response(
+                            403,
+                            json={"error": {"message": "Content stream finished due to exceeded streaming time."}},
+                        ),
+                    )
+                return SentActivity(id="stream-1", activity_params=activity)
+
+            async def mock_update(conversation_id, activity_id, activity):
+                updates.append({"id": activity_id, "text": activity.text, "text_format": activity.text_format})
+                return SentActivity(id=activity_id, activity_params=activity)
+
+            mock_api_client.conversations.create_activity = mock_create
+            mock_api_client.conversations.update_activity = mock_update
+            stream = HttpStream(mock_api_client, conversation_reference)
+
+            stream.emit(MessageActivityInput(text="Final answer").with_text_format("extendedmarkdown"))
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            result = await stream.close()
+
+            assert stream._timed_out is True
+            assert len(updates) == 1
+            assert updates[0]["text"] == "Final answer"
+            assert updates[0]["text_format"] == "extendedmarkdown"
+            assert result is not None
+
     @pytest.mark.asyncio
     async def test_close_waits_for_flush_to_complete(self, mock_api_client, conversation_reference):
         """close() must not send the final message while a flush is still mid-await."""