feat(teams): migrate native streaming to the SDK IStreamer + unwind transitional divergences (#93 PR 3/4) (#145)

* feat(teams): migrate native streaming to the SDK IStreamer + unwind transitional divergences (#93 PR 3/4)

Replace the hand-rolled Bot Framework streaming wire format with the
Teams SDK's native streamer (microsoft-teams-apps IStreamer / HttpStream),
mirroring upstream adapter-teams@chat@4.30.0 index.ts (streamViaEmit DM
path). Atomically unwind the two transitional public-type divergences
this enables.

Streaming flow (DMs):
- _handle_message_activity captures an IStreamer for DMs via
  app.activity_sender.create_stream(ref), registers it in _active_streams,
  and awaits a processing_done gate (wrapped wait_until shim) so the
  streamer stays alive while the handler streams.
- stream() -> _stream_via_emit calls stream.emit(text) per chunk; it NEVER
  calls close(). The handler's finally calls stream.close() once (the
  lifecycle-owner role the SDK App's process_activity plays upstream;
  our bridge owns dispatch via server.on_request).
- Cancellation is detected two ways: stream.canceled (checked before each
  emit) and catching StreamCancelledError (other exceptions re-raise).
- The first chunk's server-assigned id is captured via on_chunk and
  awaited only when text was emitted and the stream was not canceled.
- Group / channel / proactive threads fall back to a single buffered
  post_message (SDK-backed from PR 2).

Removed: _TeamsStreamSession, _stream_via_emit's hand-rolled wire format,
_emit_streaming_activity, _close_stream_session, _teams_send, the 1500ms
emit throttle, the clock/sleep injectables, and the
native_stream_min_emit_interval_ms config field.

Divergence unwinds (must land together — touching one without the other
corrupts recorded message history):
- types.py: removed the RawMessage.text override field (RawMessage is now
  {id, thread_id, raw}, matching upstream packages/chat/src/types.ts).
- thread.py: _handle_stream now seeds StreamOptions.update_interval_ms
  with the thread default (upstream parity, vercel/chat#340) and records
  the local accumulator (no adapter-side text override).

Throttle parity: the SDK HttpStream owns the Bot Framework streaming wire
format (streamType/streamSequence/streamId), a 500ms inter-flush throttle,
and 429 retry/backoff — verified against microsoft-teams-apps==2.0.13
http_stream.py (see docs/UPSTREAM_SYNC.md). A live Teams check is flagged
for reviewers/maintainer.

Tests: rewrote tests/test_teams_native_streaming.py to mock ctx.stream as
a StreamerProtocol double; added history-fidelity tests proving the
SentMessage->Message recording path and update_interval_ms seeding after
the unwind.

* docs(teams): clarify close-on-raw-cancel intent + pin SDK version in throttle note

Review fast-follow for PR 3 (native streaming): the finally-block comment
now distinguishes SDK-detected cancel (HttpStream.close no-ops, _canceled
set) from a raw asyncio.CancelledError (close flushes a final activity for
the accumulated text — intentional, and the except-Exception wrapper does
not catch CancelledError so cancellation still propagates). Also pins the
throttle-parity note to the installed microsoft-teams-apps==2.0.13.4 so a
future SDK bump that changes the 0.5s cadence is caught against the cited
baseline. No logic change.

Author: patrick-chinchill

docs/UPSTREAM_SYNC.md

@@ -0,0 +1,63 @@
+"""Teams SDK streamer plumbing.
+
+Builds the :class:`~microsoft_teams.api.ConversationReference` the Teams SDK
+``IStreamer`` (``microsoft_teams.apps.StreamerProtocol`` /
+``HttpStream``) needs, from the inbound Bot Framework activity dict the adapter
+already parses. Kept in its own module so the SDK model construction stays
+isolated from ``adapter.py`` and importable lazily (Port Rule: optional/SDK
+deps imported inside functions, not at module top).
+
+Mirrors what the Teams SDK's own ``ActivityContext`` does in
+``microsoft_teams/apps/app_process.py`` ``_build_context`` (it builds a
+``ConversationReference`` from the activity and calls
+``ActivitySender.create_stream(ref)`` to expose ``ctx.stream``). Our bridge
+owns dispatch, so we reproduce just the reference-building step.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from microsoft_teams.api import ConversationReference
+
+
+def build_conversation_reference(activity: dict[str, Any], *, bot_app_id: str) -> ConversationReference:
+    """Build a :class:`ConversationReference` from an inbound activity dict.
+
+    The streamer reads ``ref.conversation.id`` and ``ref.service_url`` to
+    target the Bot Framework streaming endpoint, and ``ref.bot`` to populate
+    the outgoing activity's ``from`` account. We source these from the inbound
+    activity: the bot is the activity ``recipient`` (falling back to a
+    synthetic account carrying ``bot_app_id`` when the recipient is absent),
+    the conversation is the activity ``conversation``, and ``channelId`` /
+    ``serviceUrl`` come straight off the activity.
+
+    Raises if required fields are missing — the caller catches and falls back
+    to buffered posting.
+    """
+    from microsoft_teams.api import Account, ConversationAccount, ConversationReference
+
+    recipient = activity.get("recipient") or {}
+    bot = Account(
+        id=recipient.get("id") or bot_app_id or "",
+        name=recipient.get("name"),
+    )
+
+    conversation_raw = activity.get("conversation") or {}
+    conversation = ConversationAccount(
+        id=conversation_raw.get("id") or "",
+        conversation_type=conversation_raw.get("conversationType"),
+        tenant_id=conversation_raw.get("tenantId"),
+        name=conversation_raw.get("name"),
+        is_group=conversation_raw.get("isGroup"),
+    )
+
+    return ConversationReference(
+        service_url=activity.get("serviceUrl") or "",
+        activity_id=activity.get("id"),
+        bot=bot,
+        channel_id=activity.get("channelId") or "msteams",
+        conversation=conversation,
+        locale=activity.get("locale"),
+    )

src/chat_sdk/adapters/teams/adapter.py

@@ -72,17 +72,6 @@ class TeamsAdapterConfig:
     logger: Logger | None = None
     # Override bot username (optional).
     user_name: str | None = None
-    # Minimum interval between native DM streaming activities, in
-    # milliseconds. Bot Framework's streaming endpoint is throttled to
-    # roughly 1 request/second; Microsoft recommends buffering tokens
-    # for 1.5-2 seconds to avoid 429s mid-response. We default to 1500ms
-    # per https://learn.microsoft.com/microsoftteams/platform/bots/streaming-ux.
-    # Chunks that arrive within this window after the previous emit are
-    # accumulated locally and shipped together on the next emit (or in
-    # the final ``message`` activity if the stream ends inside the
-    # window). A caller-supplied ``StreamOptions.update_interval_ms``
-    # overrides this default for a single stream.
-    native_stream_min_emit_interval_ms: int = 1500
 
 
 # =============================================================================

src/chat_sdk/adapters/teams/streamer.py

@@ -690,23 +690,15 @@ async def _handle_stream(
         # Build text-only stream from raw_stream
         text_stream = _from_full_stream(raw_stream)
 
-        # Build streaming options from current message context.
-        #
-        # Divergence from upstream — see docs/UPSTREAM_SYNC.md. Upstream
-        # (vercel/chat#340) seeds ``updateIntervalMs`` with the thread-level
-        # default (``this._streamingUpdateIntervalMs``) before spreading
-        # caller options, so adapters always see a concrete interval. We
-        # deliberately leave ``update_interval_ms`` as ``None`` unless the
-        # caller supplied one: the hand-rolled Teams native streaming path
-        # treats any non-``None`` value as a caller override of its 1500ms
-        # quota-protecting emit throttle (upstream's Teams adapter ignores
-        # the field entirely, so the seed is harmless there). Adapters that
-        # consume the field apply their own default when it is ``None``
-        # (Telegram: ``TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS``), and
-        # ``_fallback_stream`` already falls back to
-        # ``self._streaming_update_interval_ms`` — same observable behavior
-        # as upstream's seed for the fallback path.
-        options = StreamOptions()
+        # Build streaming options from current message context + caller
+        # options. Upstream parity (vercel/chat#340): seed
+        # ``update_interval_ms`` with the thread-level default
+        # (``self._streaming_update_interval_ms``) before merging caller
+        # options, so adapters always see a concrete interval. The Teams
+        # adapter now delegates throttling to the SDK ``IStreamer`` (it no
+        # longer owns a quota throttle), so the seed is harmless there —
+        # matching upstream, whose Teams adapter ignores the field.
+        options = StreamOptions(update_interval_ms=self._streaming_update_interval_ms)
         if self._current_message is not None:
             options.recipient_user_id = self._current_message.author.user_id
             # recipient_team_id is only consumed by the Slack adapter; other
@@ -744,21 +736,16 @@ async def _wrapped_stream() -> AsyncGenerator[str | StreamChunk, None]:
             wrapped_stream = _wrapped_stream()
             raw_result = await self.adapter.stream(self._id, wrapped_stream, options)  # type: ignore[union-attr]
             if raw_result is not None:
-                # Adapters can override the recorded text via the optional
-                # ``text`` field on ``RawMessage`` when their internal state
-                # (cancellation, throttling, partial commits) makes the local
-                # ``accumulated`` buffer diverge from what the platform
-                # actually accepted. Default ``None`` falls back to the local
-                # buffer — backward-compatible for adapters that don't need
-                # the override (Slack, Discord, GitHub, Google Chat,
-                # Telegram, Linear, WhatsApp). The Teams native streaming
-                # path sets it on cancellation to short-circuit the buffered
-                # suffix that was coalesced into the throttle window but
-                # never emitted.
-                recorded_text = raw_result.text if raw_result.text is not None else accumulated
+                # Record the locally-accumulated text. Matches upstream
+                # thread.ts, which builds the ``SentMessage`` from the text
+                # collected by the wrapping iterator (``accumulated``), not
+                # from the adapter's return value. The Teams adapter now
+                # emits each chunk through the SDK ``IStreamer`` as it is
+                # yielded, so ``accumulated`` and what the platform shipped
+                # stay in lockstep — no adapter-side text override needed.
                 sent = self._create_sent_message(
                     raw_result.id,
-                    PostableMarkdown(markdown=recorded_text),
+                    PostableMarkdown(markdown=accumulated),
                     raw_result.thread_id,
                 )
                 if self._thread_history is not None:

src/chat_sdk/adapters/teams/types.py

@@ -855,26 +855,6 @@ class RawMessage:
     id: str
     thread_id: str
     raw: Any
-    # Optional adapter-authoritative text snapshot. When set, callers
-    # like ``Thread.stream`` MUST prefer this over their own local
-    # accumulator when constructing the recorded ``SentMessage`` body /
-    # message-history entry. Used by adapters whose internal state
-    # (cancellation, throttling, partial commits) makes the local
-    # accumulator diverge from what the platform actually accepted —
-    # the Teams native streaming path sets this when a session is
-    # canceled mid-flight so ``Thread.stream`` records only the text
-    # Teams shipped, not the buffered suffix the user canceled out of.
-    # ``None`` means "use the caller's existing logic" — backward
-    # compatible for adapters that don't need this override.
-    #
-    # Divergence from upstream — see docs/UPSTREAM_SYNC.md. Upstream's
-    # ``RawMessage`` interface (packages/chat/src/types.ts) has only
-    # ``id``, ``raw``, ``threadId``; the override is Python-only because
-    # we hand-roll Teams native streaming (upstream uses
-    # ``@microsoft/teams.apps``'s ``IStreamer.emit`` which owns the
-    # cancellation-text reconciliation internally). Will simplify or
-    # disappear once we migrate to ``microsoft-teams-apps`` (Python).
-    text: str | None = None
 
 
 @dataclass

src/chat_sdk/thread.py

@@ -983,47 +983,11 @@ async def test_start_typing_failure_swallowed(self):
 
 
 # ---------------------------------------------------------------------------
-# _teams_send HTTP helper (retained for the still-hand-rolled streaming path)
+# Adapter lifecycle helpers
 # ---------------------------------------------------------------------------
 
 
 class TestTeamsHTTPHelpers:
-    async def test_teams_send_success(self):
-        adapter = _make_adapter(logger=_make_logger())
-
-        token_resp = _mock_aiohttp_response({"access_token": "t", "expires_in": 3600})
-        send_resp = _mock_aiohttp_response({"id": "sent-1"})
-
-        mock_session = _MockSession(default_response=send_resp)
-        original_post = mock_session.post
-
-        def routed_post(url, **kwargs):
-            if "oauth2" in url:
-                return mock_session._make_cm(token_resp)
-            return original_post(url, **kwargs)
-
-        mock_session.post = routed_post
-
-        decoded = TeamsThreadId(
-            conversation_id="19:abc@thread.tacv2",
-            service_url="https://smba.trafficmanager.net/teams/",
-        )
-
-        with patch("aiohttp.ClientSession", return_value=mock_session):
-            result = await adapter._teams_send(decoded, {"type": "message", "text": "hi"})
-            assert result["id"] == "sent-1"
-
-    async def test_teams_send_invalid_service_url_raises(self):
-        adapter = _make_adapter(logger=_make_logger())
-
-        decoded = TeamsThreadId(
-            conversation_id="19:abc@thread.tacv2",
-            service_url="https://evil.com/",
-        )
-
-        with pytest.raises(ValidationError):
-            await adapter._teams_send(decoded, {"type": "message"})
-
     async def test_disconnect_is_noop(self):
         adapter = _make_adapter(logger=_make_logger())
         result = await adapter.disconnect()
@@ -1244,16 +1208,11 @@ def test_handle_message_action_no_chat(self):
 
 
 class TestStream:
+    """Group-chat / non-DM fallback: accumulate and post one SDK message."""
+
     async def test_stream_dict_chunks(self):
         adapter = _make_adapter(logger=_make_logger())
-        send_call_count = 0
-
-        async def mock_send(decoded, payload):
-            nonlocal send_call_count
-            send_call_count += 1
-            return {"id": f"msg-{send_call_count}"}
-
-        adapter._teams_send = mock_send
+        send = _mock_app_send(adapter, "msg-1")
 
         tid = adapter.encode_thread_id(
             TeamsThreadId(
@@ -1267,14 +1226,14 @@ async def text_stream():
             yield {"type": "markdown_text", "text": "World"}
 
         result = await adapter.stream(tid, text_stream())
-        # Group chat: accumulate → single send.
+        # Group chat: accumulate → single SDK send.
         assert result.id == "msg-1"
         assert result.raw["text"] == "Hello World"
-        assert send_call_count == 1
+        send.assert_called_once()
 
     async def test_stream_string_chunks(self):
         adapter = _make_adapter(logger=_make_logger())
-        adapter._teams_send = AsyncMock(return_value={"id": "s1"})
+        send = _mock_app_send(adapter, "s1")
 
         tid = adapter.encode_thread_id(
             TeamsThreadId(
@@ -1290,11 +1249,11 @@ async def text_stream():
         result = await adapter.stream(tid, text_stream())
         assert "Hello World" in result.raw["text"]
         # Group chat: single accumulate-and-post send (no per-chunk edits).
-        assert adapter._teams_send.call_count == 1
+        send.assert_called_once()
 
     async def test_stream_empty_chunks_skipped(self):
         adapter = _make_adapter(logger=_make_logger())
-        adapter._teams_send = AsyncMock(return_value={"id": "s1"})
+        send = _mock_app_send(adapter, "s1")
 
         tid = adapter.encode_thread_id(
             TeamsThreadId(
@@ -1309,7 +1268,7 @@ async def text_stream():
 
         result = await adapter.stream(tid, text_stream())
         assert result.id == ""  # nothing sent
-        assert adapter._teams_send.call_count == 0
+        send.assert_not_called()
 
 
 # ---------------------------------------------------------------------------
@@ -1368,37 +1327,6 @@ def test_extract_card_title_body_not_list(self):
 # tests/test_teams_bridge.py.
 
 
-# ---------------------------------------------------------------------------
-# _teams_send error path (retained for the still-hand-rolled streaming path)
-# ---------------------------------------------------------------------------
-
-
-class TestTeamsHTTPErrorPaths:
-    async def test_teams_send_non_ok_response(self):
-        adapter = _make_adapter(logger=_make_logger())
-
-        token_resp = _mock_aiohttp_response({"access_token": "t", "expires_in": 3600})
-        error_resp = _mock_aiohttp_response({"error": "bad"}, status=500)
-
-        mock_session = _MockSession(default_response=error_resp)
-        original_post = mock_session.post
-
-        def routed_post(url, **kwargs):
-            if "oauth2" in url:
-                return mock_session._make_cm(token_resp)
-            return original_post(url, **kwargs)
-
-        mock_session.post = routed_post
-
-        decoded = TeamsThreadId(
-            conversation_id="19:abc@thread.tacv2",
-            service_url="https://smba.trafficmanager.net/teams/",
-        )
-
-        with patch("aiohttp.ClientSession", return_value=mock_session), pytest.raises(NetworkError):
-            await adapter._teams_send(decoded, {"type": "message"})
-
-
 # ---------------------------------------------------------------------------
 # fetch_channel_info with channel context
 # ---------------------------------------------------------------------------

src/chat_sdk/types.py

@@ -736,11 +736,13 @@ class TestStream:
     async def test_group_chat_stream_accumulates_and_posts_single_message(self):
         """Group chats / channels accumulate the stream and post one message.
 
-        Mirrors upstream after vercel/chat#416: ``streamViaEmit`` is reserved
-        for DMs; non-DM threads no longer post+edit (which produced flicker).
+        Mirrors upstream ``stream`` (adapter-teams@chat@4.30.0): native
+        ``streamViaEmit`` is reserved for DMs (where an ``IStreamer`` exists);
+        non-DM threads accumulate and ``postMessage`` a single message via the
+        SDK ``App.send`` (PR 2-backed).
         """
         adapter = _make_adapter(logger=_make_logger())
-        adapter._teams_send = AsyncMock(return_value={"id": "stream-msg-1"})
+        send = _mock_app_send(adapter, "stream-msg-1")
 
         tid = adapter.encode_thread_id(
             TeamsThreadId(
@@ -755,17 +757,18 @@ async def text_gen():
 
         result = await adapter.stream(tid, text_gen())
         assert result.id == "stream-msg-1"
-        # Single send carrying the full accumulated text — no edits.
-        assert adapter._teams_send.call_count == 1
-        sent_payload = adapter._teams_send.await_args.args[1]
-        assert sent_payload["text"] == "Hello world"
-        assert sent_payload["type"] == "message"
+        # Single SDK send carrying the full accumulated text — no edits.
+        send.assert_called_once()
+        conv_id, activity = send.call_args.args
+        assert conv_id == "19:abc@thread.tacv2"
+        assert activity.text == "Hello world"
+        assert activity.text_format == "markdown"
 
     @pytest.mark.asyncio
     async def test_group_chat_stream_empty_returns_empty(self):
         """Empty streams in a group chat skip the post entirely."""
         adapter = _make_adapter(logger=_make_logger())
-        adapter._teams_send = AsyncMock(return_value={"id": "stream-msg-2"})
+        send = _mock_app_send(adapter, "stream-msg-2")
 
         tid = adapter.encode_thread_id(
             TeamsThreadId(
@@ -782,4 +785,4 @@ async def text_gen():
         # No real text → no send, returned RawMessage carries empty content.
         assert result.id == ""
         assert result.raw["text"] == ""
-        assert adapter._teams_send.call_count == 0
+        send.assert_not_called()