fix(teams): prevent webhook deadlock when no chat task is scheduled

Address CodeRabbit Critical on PR #88. ``Chat.process_message`` invokes
``wait_until`` synchronously only if a chat task is actually scheduled.
If the message is deduped, dropped by the concurrency strategy, or
otherwise short-circuited, no task is created and ``wait_until`` is
never called — leaving ``processing_done`` unresolved and ``await
processing_done`` hanging the webhook indefinitely.

After ``process_message`` returns, check whether ``processing_done`` was
resolved (which can only happen via ``_chained_wait_until``, which is
only invoked when a task is scheduled). If not, resolve immediately —
there's no in-flight handler to wait on, no streaming will happen, and
``await processing_done`` should fall through to the ``finally`` cleanup.

Also addresses Codex P2 (lines 471): caller-supplied
``WebhookOptions.wait_until`` raising synchronously now logs and
continues instead of escaping through ``Chat.process_message``,
preventing the same-class deadlock-via-finally issue.

Both fixes together close the two paths where the streaming session
could be torn down while the chat task was still scheduled (or never
scheduled): a synchronous wait_until exception, and a deduped/dropped
message.

Tests:
- New ``test_caller_wait_until_raise_does_not_kill_native_streaming``
  verifies the wait_until-raise path keeps the session alive long
  enough for the chat task to call ``thread.stream()`` natively.
- Existing 23 native-streaming tests still pass.

Skipped CodeRabbit's other comment (``_active_streams`` thread-id
keying) — same as Codex P2 #2, matches upstream TS behavior and would
be a heavy-lift Python-only divergence.

https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj

Author: claude

src/chat_sdk/adapters/teams/adapter.py

@@ -468,12 +468,34 @@ def _chained_wait_until(task: Awaitable[Any]) -> None:
             # cannot starve due to a misbehaving caller-supplied hook.
             _resolve_processing(task)
             if upstream_wait_until is not None:
-                upstream_wait_until(task)
+                # Catch synchronous failures in the caller's hook. If we
+                # let it escape, ``Chat.process_message`` propagates the
+                # exception, the outer ``try`` skips ``await processing_done``,
+                # and the ``finally`` tears down the session while the
+                # underlying chat task is still scheduled — handlers that
+                # later call ``thread.stream()`` would then miss native
+                # streaming and fall back to a normal post. Logging keeps
+                # the failure visible without breaking the streaming path.
+                try:
+                    upstream_wait_until(task)
+                except Exception as exc:
+                    self._logger.warn(
+                        "Caller-supplied WebhookOptions.wait_until raised",
+                        {"threadId": thread_id, "error": str(exc)},
+                    )
 
         chained_options = WebhookOptions(wait_until=_chained_wait_until)
 
         try:
             self._chat.process_message(self, thread_id, message, chained_options)
+            # If ``process_message`` returned without invoking
+            # ``wait_until`` synchronously, no chat task was scheduled
+            # (deduped, dropped by the concurrency strategy, or the
+            # message wasn't admitted for handling). Resolve the gate
+            # immediately so ``await processing_done`` doesn't hang
+            # forever — there is no in-flight handler to wait on.
+            if not processing_done.done():
+                processing_done.set_result(None)
             try:
                 await processing_done
             except asyncio.CancelledError:

tests/test_teams_native_streaming.py

@@ -540,7 +540,79 @@ class TestHandleMessageActivityLifecycle:
     """Verify the message-activity → process_message → stream → close flow."""
 
     @pytest.mark.asyncio
-    async def test_dm_message_registers_session_and_closes_after_processing(self):
+    async def test_caller_wait_until_raise_does_not_kill_native_streaming(self):
+        """A caller-supplied ``WebhookOptions.wait_until`` that raises must
+        NOT tear down the DM streaming session before the chat task runs.
+
+        What to fix if this fails: in
+        ``src/chat_sdk/adapters/teams/adapter.py`` ``_chained_wait_until``,
+        the call to the upstream ``wait_until`` must be wrapped in
+        ``try/except`` (and logged). Otherwise the synchronous raise
+        escapes through ``Chat.process_message``, the outer ``try`` skips
+        ``await processing_done``, and the ``finally`` removes the session
+        while the chat task is still scheduled. The handler's later
+        ``thread.stream()`` call would then miss native streaming and
+        fall back to a normal post.
+        """
+        adapter = _make_adapter()
+        adapter._teams_send = AsyncMock(return_value={"id": "id-1"})
+
+        tid = _dm_thread_id(adapter)
+
+        # Build a chat that schedules the streaming task AND invokes
+        # a deliberately-raising upstream wait_until.
+        chat = MagicMock()
+        chat.get_state = MagicMock(return_value=None)
+        stream_calls: list[str] = []
+
+        def process_message(adapter_arg, thread_id, message, options):
+            async def _do_stream():
+                async def gen():
+                    yield "hi"
+
+                # Snapshot whether native streaming is still wired up at
+                # the moment the chat task runs.
+                stream_calls.append(
+                    "native" if thread_id in adapter_arg._active_streams else "fallback"
+                )
+                await adapter_arg.stream(thread_id, gen())
+
+            task = asyncio.get_running_loop().create_task(_do_stream())
+            # Caller-supplied wait_until raises synchronously. The chained
+            # wrapper must swallow this so processing_done still resolves.
+            options.wait_until(task)
+
+        chat.process_message = process_message
+        adapter._chat = chat
+
+        # Inject a raising upstream wait_until via WebhookOptions.
+        from chat_sdk.types import WebhookOptions
+
+        def raising_wait_until(_task: Any) -> None:
+            raise RuntimeError("caller wait_until exploded")
+
+        upstream_options = WebhookOptions(wait_until=raising_wait_until)
+
+        activity = {
+            "type": "message",
+            "id": "incoming-1",
+            "text": "user said something",
+            "from": {"id": "user-1", "name": "User One"},
+            "conversation": {"id": "a:1Abc-DM-conversation-id"},
+            "serviceUrl": "https://smba.trafficmanager.net/teams/",
+        }
+
+        # Should NOT raise — the chained wrapper logs and continues.
+        await adapter._handle_message_activity(activity, upstream_options)
+
+        # The streaming task ran while the session was still registered.
+        assert stream_calls == ["native"], (
+            "Caller wait_until raise tore down the session before the chat "
+            "task ran; the handler fell back to a normal post instead of "
+            "native Teams streaming"
+        )
+        # Session was cleaned up after the task finished.
+        assert tid not in adapter._active_streams
         """A DM message activity registers a session, awaits processing, then drops it."""
         adapter = _make_adapter()
         adapter._teams_send = AsyncMock(return_value={"id": "id-1"})