[#535] StreamingResponse: mutating the finalized "final" stream activity (delete or update) after end_stream() re-hangs the Teams client's typing/streaming indicator

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/typing_indicator.py

@@ -217,7 +217,17 @@ async def _on_send_activities_handler(ctx, activities, next_handler):
                     self._stop_loop()
                 return await next_handler()
 
+            async def _on_update_activity_handler(ctx, activity, next_handler):
+                self._stop_loop()
+                return await next_handler()
+
+            async def _on_delete_activity_handler(ctx, reference, next_handler):
+                self._stop_loop()
+                return await next_handler()
+
             self._context.on_send_activities(_on_send_activities_handler)
+            self._context.on_update_activity(_on_update_activity_handler)
+            self._context.on_delete_activity(_on_delete_activity_handler)
             self._hook_registered = True
 
     def _stop_loop(self) -> None:

tests/hosting_core/app/test_typing_indicator.py

@@ -46,9 +46,20 @@ def __init__(self, should_raise: bool = False, channel_id: str = "test") -> None
             recipient={"id": "user"},
         )
         self._on_send_handlers = []
+        self._on_update_handlers = []
+        self._on_delete_handlers = []
 
     def on_send_activities(self, handler):
         self._on_send_handlers.append(handler)
+        return self
+
+    def on_update_activity(self, handler):
+        self._on_update_handlers.append(handler)
+        return self
+
+    def on_delete_activity(self, handler):
+        self._on_delete_handlers.append(handler)
+        return self
 
     @property
     def sent_activities(self):
@@ -334,6 +345,47 @@ async def _next_handler():
     await indicator._stop_async()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "handlers_attribute",
+    [
+        "_on_update_handlers",
+        "_on_delete_handlers",
+    ],
+)
+async def test_mutation_hook_stops_before_initial_typing_send(handlers_attribute):
+    """Update and delete operations should cancel typing before the mutation."""
+    context = StubTurnContext()
+    opts = _fast_options(initial_delay_ms=50, interval_ms=10)
+    indicator = TypingIndicator(context, typing_options=opts)
+    indicator.start()
+
+    mutation = Activity(type=ActivityTypes.message)
+    if handlers_attribute == "_on_delete_handlers":
+        mutation = context.activity.get_conversation_reference()
+        mutation.activity_id = "activity-id"
+
+    handlers = getattr(context, handlers_attribute)
+    assert len(handlers) == 1
+
+    next_handler_called = False
+
+    async def _next_handler():
+        nonlocal next_handler_called
+        next_handler_called = True
+        return "mutation-result"
+
+    result = await handlers[0](context, mutation, _next_handler)
+    await asyncio.sleep(0.075)
+
+    assert result == "mutation-result"
+    assert next_handler_called
+    assert indicator._stopped
+    assert context.sent_activities == []
+
+    await indicator._stop_async()
+
+
 # ---------------------------------------------------------------------------
 # Per-channel strategy tests
 # ---------------------------------------------------------------------------