Support resetting response streams

examples/stream/README.md

@@ -2,6 +2,10 @@
 
 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 `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
 
 ```bash

examples/stream/src/main.py

@@ -9,6 +9,7 @@
 
 from microsoft_teams.api import CardAction, CardActionType, MessageActivity, MessageActivityInput, SuggestedActions
 from microsoft_teams.apps import ActivityContext, App
+from microsoft_teams.cards import AdaptiveCard, TextBlock
 
 # Surface SDK INFO/WARNING logs (including the anonymous-mode startup warning
 # emitted when CLIENT_ID / CLIENT_SECRET / TENANT_ID are not configured).
@@ -31,11 +32,75 @@
     "✨ Stream test complete!",
 ]
 
+FIRST_STREAM_MESSAGES = [
+    "[stream 1] Starting the first streamed response. ",
+    "[stream 1] This is using the default ctx.stream instance. ",
+    "[stream 1] Next the handler will close the current streamed message.",
+]
+
+SECOND_STREAM_MESSAGES = [
+    "[stream 2] Reusing ctx.stream after emit reopens the closed stream. ",
+    "[stream 2] This should render after stream 1's final Adaptive Card message. ",
+    "[stream 2] The app processor will close this stream when the handler returns.",
+]
+
+
+def should_run_multi_stream(text: str | None) -> bool:
+    normalized = (text or "").lower().replace("-", " ")
+    return "multi stream" in normalized
+
+
+def should_send_simple_card(text: str | None) -> bool:
+    normalized = (text or "").lower().replace("-", " ")
+    return "simple card" in normalized
+
+
+def create_simple_card() -> AdaptiveCard:
+    return AdaptiveCard(
+        version="1.4",
+        body=[
+            TextBlock(text="Simple Adaptive Card", weight="Bolder", size="Large", wrap=True),
+            TextBlock(text="If you can see this card, basic Adaptive Card delivery is working.", wrap=True),
+        ],
+    )
+
 
 @app.on_message
 async def handle_message(ctx: ActivityContext[MessageActivity]):
     """Stream messages to the user on any message activity."""
 
+    if should_send_simple_card(ctx.activity.text):
+        sent_card = await ctx.send(
+            MessageActivityInput(text="Sending a simple Adaptive Card.").add_card(create_simple_card())
+        )
+        logger.info("Sent simple adaptive card: %s", sent_card.id)
+        return
+
+    if should_run_multi_stream(ctx.activity.text):
+        ctx.stream.update("Starting stream 1...")
+        await asyncio.sleep(1)
+
+        for message in FIRST_STREAM_MESSAGES:
+            await asyncio.sleep(0.5)
+            ctx.stream.emit(message)
+
+        card_message = MessageActivityInput(text="Adaptive Card emitted as part of stream 1.").add_card(
+            create_simple_card()
+        )
+        ctx.stream.emit(card_message)
+        sent_message = await ctx.stream.close()
+        if sent_message:
+            logger.info("Sent stream 1 final message with adaptive card: %s", sent_message.id)
+        await asyncio.sleep(2)
+
+        ctx.stream.update("Starting stream 2...")
+        await asyncio.sleep(1)
+
+        for message in SECOND_STREAM_MESSAGES:
+            await asyncio.sleep(0.5)
+            ctx.stream.emit(message)
+        return
+
     ctx.stream.update("Stream starting...")
     await asyncio.sleep(1)
 

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

@@ -62,17 +62,22 @@ def __init__(self, client: ApiClient, ref: ConversationReference):
         self._state_changed = asyncio.Event()
 
         self._canceled = False
-        self._reset_state()
+        self._reset_current_stream()
 
-    def _reset_state(self) -> None:
-        """Reset the stream state to initial values."""
+    def _reset_current_stream(self) -> None:
+        """Reset per-message state for the current stream cycle."""
         self._index = 1
         self._id: Optional[str] = None
         self._text: str = ""
         self._channel_data: ChannelData = ChannelData()
         self._final_activity: Optional[MessageActivityInput] = None
         self._queue: deque[Union[MessageActivityInput, TypingActivityInput, str]] = deque()
 
+    def _reset_stream_for_next_stream(self) -> None:
+        """Prepare the stream instance to start a new stream cycle."""
+        self._reset_current_stream()
+        self._result = None
+
     @property
     def canceled(self) -> bool:
         """
@@ -83,7 +88,7 @@ def canceled(self) -> bool:
 
     @property
     def closed(self) -> bool:
-        """Whether the final stream message has been sent."""
+        """Whether the current streamed message has been finalized."""
         return self._result is not None
 
     @property
@@ -100,7 +105,7 @@ def on_chunk(self, handler: Callable[[SentActivity], Awaitable[None]]):
         self._events.on("chunk", handler)
 
     def on_close(self, handler: Callable[[SentActivity], Awaitable[None]]):
-        self._events.once("close", handler)
+        self._events.on("close", handler)
 
     def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) -> None:
         """
@@ -113,6 +118,10 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str])
         if self._canceled:
             raise StreamCancelledError("Stream has been cancelled.")
 
+        if self.closed:
+            logger.debug("starting a new streamed message after close")
+            self._reset_stream_for_next_stream()
+
         if isinstance(activity, str):
             activity = MessageActivityInput(text=activity, type="message")
         self._queue.append(activity)
@@ -163,9 +172,16 @@ async def _poll():
             return False
 
     async def close(self) -> Optional[SentActivity]:
-        # wait for lock to be free
-        if self._result is not None:
+        """
+        Finalize the current streamed message.
+
+        Closing is idempotent until the next emit or update. Emitting or
+        updating after close starts a new streamed message using the same
+        stream instance.
+        """
+        if self.closed:
             logger.debug("stream already closed with result")
+            assert self._result is not None
             return self._result
 
         if self._canceled:
@@ -203,8 +219,8 @@ async def close(self) -> Optional[SentActivity]:
         # Emit close event
         self._events.emit("close", res)
 
-        # Reset state
-        self._reset_state()
+        # Reset per-message state; keep event handlers and cancellation state.
+        self._reset_current_stream()
         self._result = res
         logger.debug("stream closed with result: %s", res)
 

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

@@ -30,7 +30,7 @@ def canceled(self) -> bool:
 
     @property
     def closed(self) -> bool:
-        """Whether the final stream message has been sent."""
+        """Whether the current streamed message has been finalized."""
         ...
 
     @property
@@ -59,10 +59,10 @@ def on_chunk(self, handler: Callable[[SentActivity], Awaitable[None]]) -> None:
 
     def on_close(self, handler: Callable[[SentActivity], Awaitable[None]]) -> None:
         """
-        Register a handler for close events.
+        Register a handler for stream close events.
 
         Args:
-            handler: Async function that will be called when the stream closes
+            handler: Async function that will be called each time the stream closes
         """
         ...
 
@@ -89,6 +89,10 @@ def clear_text(self) -> None:
 
     async def close(self) -> Optional[SentActivity]:
         """
-        Close the stream.
+        Finalize the current streamed message.
+
+        Closing is idempotent until the next emit or update. Emitting or
+        updating after close starts a new streamed message using the same
+        stream instance.
         """
         ...

packages/apps/tests/test_http_stream.py

@@ -453,3 +453,49 @@ async def test_close_waits_for_flush_to_complete(self, mock_api_client, conversa
         assert result is not None
         assert mock_api_client.send_call_count == 1
         assert mock_api_client.sent_activities[0].text == "Response text"
+
+    @pytest.mark.asyncio
+    async def test_emit_after_close_reopens_stream_for_next_message(
+        self, mock_api_client, conversation_reference, patch_loop_call_later
+    ):
+        loop = asyncio.get_running_loop()
+        patcher, scheduled = patch_loop_call_later(loop)
+        close_results: list[SentActivity] = []
+        close_event = asyncio.Event()
+
+        async def handle_close(activity: SentActivity) -> None:
+            close_results.append(activity)
+            close_event.set()
+
+        with patcher:
+            stream = HttpStream(mock_api_client, conversation_reference)
+            stream.on_close(handle_close)
+
+            stream.emit("First streamed message")
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            first_result = await stream.close()
+            await close_event.wait()
+            close_event.clear()
+            assert first_result is not None
+            assert stream.closed is True
+
+            repeated_result = await stream.close()
+            assert repeated_result is first_result
+            assert stream.closed is True
+
+            stream.emit("Second streamed message")
+            assert stream.closed is False
+            await asyncio.sleep(0)
+            await self._run_scheduled_flushes(scheduled)
+
+            second_result = await stream.close()
+            await close_event.wait()
+            assert second_result is not None
+            assert stream.closed is True
+
+            assert first_result.id != second_result.id
+            assert first_result.activity_params.text == "First streamed message"
+            assert second_result.activity_params.text == "Second streamed message"
+            assert [result.id for result in close_results] == [first_result.id, second_result.id]