fix(teams): address PR #88 review findings

- Add a same-thread concurrent _handle_message_activity test that exercises
  the realistic _active_streams race (two near-simultaneous DM webhooks for
  the same thread). Pins upstream's plain-Map clobber semantics: the second
  registration overwrites the first, both in-flight handlers observe the
  later session, and the registry ends empty after both finish. The original
  distinct-threads test is renamed to make clear it covers session
  ISOLATION, not the registry race.

- Empty-id final-send fallback: when Teams accepts streaming chunks but
  returns id="" on the first activity, _close_stream_session now ships the
  final message anyway (omitting streamId from channelData) instead of
  skipping and leaving the streaming UI spinning until Teams times the
  session out client-side. Mirrors upstream's looser check (text non-empty
  → ship the final). Adds a regression test and a non-parity row in
  docs/UPSTREAM_SYNC.md.

- _chained_wait_until: resolve our internal processing_done gate BEFORE
  invoking the caller-supplied waitUntil, so the deadlock-immunity argument
  is trivially obvious (a misbehaving upstream callback can't starve the
  await on processing_done).

- _TeamsStreamSession: add a public read-only `text` property so external
  callers (now _close_stream_session) read through it instead of the
  underscore-prefixed _text attribute. _stream_via_emit retains the direct
  _text write as the canonical mutator.

Author: claude

docs/UPSTREAM_SYNC.md

@@ -489,6 +489,7 @@ stay explicit instead of being rediscovered in code review.
 | Fallback streaming final SentMessage content (non-Teams adapters) | SentMessage + final edit carry `final_content` (remend'd — inline markers auto-closed) | SentMessage + final edit carry raw `accumulated` | Narrow UX refinement. If a stream ends with an unclosed `*`/`~~`/etc., upstream ships the unclosed marker; we run `_remend` so the user sees a clean final message. Not observable in the common case where streams close their own markers. Teams native streaming and the Teams accumulate-and-post path both ship raw `accumulated`, matching upstream after #416; this divergence applies only to the remaining adapters that still route through `_fallback_stream`. |
 | Teams group-chat / channel streaming via accumulate-and-post | `TeamsAdapter.stream` accumulates the full text and issues a single `post_message` instead of post+edit, even for group chats and channel threads | Same after vercel/chat#416 (`if (activeStream && !activeStream.canceled) … else { accumulate; postMessage }`) — no divergence at the adapter level | Documented for clarity: the Python port matches upstream's post-#416 behavior of avoiding the post+edit flicker where Teams doesn't support native streaming. The adapter no longer touches `_teams_update` from the streaming path. |
 | Teams native streaming soft-cancel on send failure (DMs) | A `_teams_send` error mid-stream (e.g. 429, 5xx) cancels the streaming session, suppresses the exception, and returns the `RawMessage` with whatever text was already accepted | Upstream's `IStreamer.emit` is fire-and-forget under the SDK; no public hook to fail-soft on per-chunk send errors | Bot Framework streaming activities are advisory — once cancelled, Teams times the streaming UI out client-side. Soft-cancel preserves the chat-handler's ability to continue work after a transient backend hiccup, instead of bubbling the rate-limit through the stream iterator. The `RawMessage` carries the user-visible text so SentMessage history stays consistent. |
+| Teams native streaming final-send when first chunk's `id` was empty (DMs) | `_close_stream_session` sends the final `message` activity whenever `text` is non-empty, even if `stream_id` is `None` (Bot Framework REST response returned `{"id": ""}` on the first chunk). The final activity omits `streamId` from `channelData` rather than serializing `None`. | Upstream's `streamViaEmit` awaits the `chunk` event for the first activity's `id`; if Teams returns an empty id, `messageId` becomes `""` and the SDK's auto-close emits the final activity through `IStreamer` regardless | Without the final `message` activity, the Teams client's streaming UI keeps spinning until the platform times the session out — a stuck-loading-state UX failure with no user workaround. We mirror upstream's looser check (text non-empty → ship the final) so the streaming indicator clears even when the Bot Framework REST response surface returns an empty `id`. |
 | Teams divider rendering | `card_to_adaptive_card` hoists `separator: True` onto the next sibling (or emits a non-empty Container for a trailing divider) | `convertDividerToElement` emits an empty `Container` with `separator: True` | Upstream shares the same bug: Microsoft Teams renders an empty Container at zero height, so the separator line is effectively invisible. Python port fixes locally (issue #45) rather than blocking on upstream. |
 | `SlackAdapter.current_token` / `current_client` | Public `@property` accessors that return the request-context-bound token and a preconfigured `AsyncWebClient` | Not exposed (`getToken()` is private on the TS `SlackAdapter`) | Python-only addition (issue #47). Downstream code that calls Slack Web APIs from inside a handler — email resolution, user profile fetches, reaction bookkeeping — otherwise depends on underscore-prefixed helpers. |
 | `ConcurrencyConfig.max_concurrent` | Enforced via `asyncio.Semaphore` in the `"concurrent"` strategy path; rejects non-integer or `<= 0` values, and rejects any non-`None` `max_concurrent` paired with a non-`"concurrent"` strategy | Accepted into the config type with docstring "Default: Infinity" but never read (3 writes, 0 reads) | Silent correctness bug upstream — consumers setting `max_concurrent=N` with `strategy="concurrent"` reasonably expect an N-way bound on in-flight handlers. We honor the documented contract via a semaphore and fail-fast on misconfiguration so it's never silent. `max_concurrent=None` stays compatible with every strategy (unbounded default). |

src/chat_sdk/adapters/teams/adapter.py

@@ -92,6 +92,16 @@ def cancel(self) -> None:
         """Mark the session canceled. ``stream()`` checks this each chunk."""
         self.canceled = True
 
+    @property
+    def text(self) -> str:
+        """Read-only view of the cumulative streamed text.
+
+        External callers (tests, other adapter helpers) should read this
+        instead of poking at the private ``_text`` attribute. Writes go
+        through ``_stream_via_emit`` which owns the buffer.
+        """
+        return self._text
+
 
 # Bot Framework streaming protocol values for ``channelData.streamType``.
 _STREAM_TYPE_STREAMING = "streaming"
@@ -450,9 +460,15 @@ def _on_done(_t: asyncio.Task[Any]) -> None:
         upstream_wait_until = options.wait_until if options is not None else None
 
         def _chained_wait_until(task: Awaitable[Any]) -> None:
+            # Resolve our own gate FIRST, before invoking the upstream
+            # ``wait_until`` callback. This way, even if the upstream
+            # callback raises, blocks, or never fires, ``processing_done``
+            # is still wired up — making the deadlock-immunity argument
+            # trivially obvious: the await on ``processing_done`` below
+            # cannot starve due to a misbehaving caller-supplied hook.
+            _resolve_processing(task)
             if upstream_wait_until is not None:
                 upstream_wait_until(task)
-            _resolve_processing(task)
 
         chained_options = WebhookOptions(wait_until=_chained_wait_until)
 
@@ -1194,7 +1210,10 @@ async def _stream_via_emit(
 
         # Persist accumulated text on the session so close() can emit the
         # final ``message`` activity with the same content the user saw.
-        session._text = accumulated  # noqa: SLF001 — same module
+        # Direct ``_text`` write is the canonical mutator (the public
+        # ``text`` property is read-only by design); both classes live in
+        # the same module so this isn't a cross-module SLF001.
+        session._text = accumulated  # noqa: SLF001
         return RawMessage(
             id=session.first_chunk_id,
             thread_id=thread_id,
@@ -1208,28 +1227,39 @@ async def _close_stream_session(
     ) -> None:
         """Send the final ``message`` activity to close out a stream.
 
-        No-op if the session was canceled or never emitted any chunks (no
-        ``streamId`` was ever assigned). The final activity replaces the
-        running ``typing`` indicator with a real message containing the full
-        accumulated text — Teams clients clear the streaming UI on receipt.
+        No-op if the session was canceled, or if no chunks were ever
+        emitted (empty ``text``). Otherwise we send the final activity —
+        even if the server never returned an ``id`` for the first chunk
+        (i.e. ``stream_id`` is ``None``), in which case we omit
+        ``streamId`` from ``channelData``. Mirrors upstream's looser
+        check: as long as the user saw streamed text, ship the final
+        ``message`` so the Teams streaming UI clears, instead of leaving
+        it spinning until Teams times the session out client-side.
         """
         if session.canceled:
             return
-        # session.stream_id is only set after a successful chunk send, and
-        # empty chunks are skipped before send — so stream_id-set implies
-        # _text non-empty. Single check is sufficient.
-        if session.stream_id is None:
+        # ``text`` is the cumulative buffer; empty means nothing was ever
+        # emitted (empty stream, or stream canceled before first send).
+        if not session.text:
             return
 
         decoded = self.decode_thread_id(thread_id)
+        channel_data: dict[str, Any] = {
+            "streamType": _STREAM_TYPE_FINAL,
+        }
+        # Hazard #7 — only include ``streamId`` when we actually have one.
+        # The Bot Framework REST response can return ``id=""`` even on a
+        # 200, in which case ``stream_id`` stays ``None`` (see emit guard
+        # in ``_stream_via_emit``); ship the final without a ``streamId``
+        # rather than skipping the send.
+        if session.stream_id is not None:
+            channel_data["streamId"] = session.stream_id
+
         final_activity: dict[str, Any] = {
             "type": "message",
-            "text": session._text,  # noqa: SLF001
+            "text": session.text,
             "textFormat": "markdown",
-            "channelData": {
-                "streamType": _STREAM_TYPE_FINAL,
-                "streamId": session.stream_id,
-            },
+            "channelData": channel_data,
             "entities": [
                 {
                     "type": "streaminfo",

tests/test_teams_native_streaming.py

@@ -313,6 +313,54 @@ async def test_close_session_no_chunks_no_op(self):
         await adapter._close_stream_session(tid, session)
         adapter._teams_send.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_close_session_sends_final_when_first_chunk_returned_empty_id(
+        self,
+    ):
+        """If Teams accepted chunks but never returned an ``id``, still send the final.
+
+        Regression for the empty-``id`` edge case: the Bot Framework REST
+        response can be 200 with ``{"id": ""}`` even on a successful
+        ``typing`` activity send. ``stream_id`` stays ``None`` (the
+        first-chunk guard skips assignment for the empty string), but
+        ``text`` is non-empty because the user already saw the streamed
+        chunks. Without a final ``message`` activity the Teams streaming
+        UI would spin until Teams times the session out client-side —
+        ship the final ``message`` anyway, omitting ``streamId`` from
+        ``channelData``. Mirrors upstream's looser check.
+        """
+        adapter = _make_adapter()
+        # First call (chunk): returns an empty id. Second call (final):
+        # succeeds.
+        adapter._teams_send = AsyncMock(side_effect=[{"id": ""}, {"id": "final-id"}])
+        tid = _dm_thread_id(adapter)
+        session = _TeamsStreamSession()
+        adapter._active_streams[tid] = session
+
+        async def text_gen():
+            yield "hello world"
+
+        await adapter._stream_via_emit(tid, text_gen(), session)
+
+        # Sanity: the chunk send went through, but stream_id is unset
+        # because the server didn't hand us one.
+        assert session.stream_id is None
+        assert session.text == "hello world"
+
+        # Now close: the final ``message`` activity must still be sent
+        # (omitting ``streamId``).
+        await adapter._close_stream_session(tid, session)
+
+        assert adapter._teams_send.await_count == 2
+        final_payload = adapter._teams_send.await_args_list[1].args[1]
+        assert final_payload["type"] == "message"
+        assert final_payload["text"] == "hello world"
+        assert final_payload["channelData"]["streamType"] == _STREAM_TYPE_FINAL
+        # Critical: no streamId key when the server never assigned one,
+        # rather than serializing ``"streamId": None``.
+        assert "streamId" not in final_payload["channelData"]
+        assert final_payload["entities"] == [{"type": "streaminfo", "streamType": _STREAM_TYPE_FINAL}]
+
 
 class TestStreamErrors:
     @pytest.mark.asyncio
@@ -603,12 +651,18 @@ async def _failing():
 
 class TestPassInteraction:
     @pytest.mark.asyncio
-    async def test_two_concurrent_dm_streams_have_independent_sessions(self):
-        """Two streams to two different DM threads must not share state.
-
-        Per docs/SELF_REVIEW.md pass-interaction check: two DMs in flight
-        from the same bot to the same user (one per thread) must each
-        carry their own ``streamId`` and ``streamSequence``.
+    async def test_distinct_dm_threads_each_have_isolated_session_state(self):
+        """Two DM threads streaming in parallel must not share session state.
+
+        This pins the ISOLATION property when sessions are explicitly
+        passed to ``_stream_via_emit`` (the registry is bypassed). Two
+        DMs in flight from the same bot to the same user (one per
+        thread) must each carry their own ``streamId`` and
+        ``streamSequence``.
+
+        Same-thread concurrency (the ``_active_streams`` race) is a
+        DIFFERENT property — see
+        ``test_same_thread_concurrent_handlers_clobber_active_stream``.
         """
         adapter = _make_adapter()
         # Distinct server ids per send so we can verify thread-to-id mapping.
@@ -663,3 +717,95 @@ async def gen_b():
         # and B's to B's.
         for conv_id, payload in send_log:
             assert payload["text"].startswith("A" if "A" in conv_id else "B")
+
+    @pytest.mark.asyncio
+    async def test_same_thread_concurrent_handlers_clobber_active_stream(self):
+        """Two near-simultaneous webhooks for the SAME DM thread.
+
+        Realistic case: a user double-sends, or two webhooks land on the
+        same thread before the first finishes. ``_active_streams`` is a
+        plain ``dict`` keyed by ``thread_id``, so the second registration
+        overwrites the first — pin that behavior here so a future change
+        to add per-thread queueing/locking is a deliberate decision, not
+        an accidental observable change.
+
+        Upstream's ``activeStreams`` is also a plain ``Map`` with the
+        same overwrite semantics; this test mirrors that contract.
+        """
+        adapter = _make_adapter()
+        # Track each session that gets registered, in the order of registration.
+        registered_sessions: list[_TeamsStreamSession] = []
+        # Snapshot the registry contents immediately AFTER each handler's
+        # process_message call so we can pin the clobber.
+        post_registration_snapshots: list[_TeamsStreamSession] = []
+
+        # Block both handlers on a barrier so the second registration races
+        # the first while the first is still "in flight". This pins the
+        # registry behavior under genuine overlap, not just sequential calls.
+        first_registered = asyncio.Event()
+        release_handlers = asyncio.Event()
+
+        adapter._teams_send = AsyncMock(return_value={"id": "send-id"})
+
+        chat = MagicMock()
+        chat.get_state = MagicMock(return_value=None)
+
+        def process_message(adapter_arg, thread_id, message, options):
+            # Snapshot the session that THIS handler call registered.
+            registered_sessions.append(adapter_arg._active_streams[thread_id])
+
+            async def _handler():
+                # Hold both handlers open across the barrier so they truly
+                # overlap. After release, snapshot the registry — by this
+                # point both handlers have registered, and the LATER
+                # registration must have won.
+                if not first_registered.is_set():
+                    first_registered.set()
+                await release_handlers.wait()
+                post_registration_snapshots.append(adapter_arg._active_streams.get(thread_id))
+
+            task = asyncio.get_running_loop().create_task(_handler())
+            options.wait_until(task)
+
+        chat.process_message = process_message
+        adapter._chat = chat
+
+        tid = _dm_thread_id(adapter)
+        activity = {
+            "type": "message",
+            "id": "incoming-same-thread",
+            "text": "user said something",
+            "from": {"id": "user-1", "name": "User One"},
+            "conversation": {"id": "a:1Abc-DM-conversation-id"},
+            "serviceUrl": "https://smba.trafficmanager.net/teams/",
+        }
+
+        async def _drive_two_handlers():
+            # Start the first; wait until it has registered before launching
+            # the second so the second observes (and clobbers) the first's
+            # registry entry. Then release both.
+            t1 = asyncio.create_task(adapter._handle_message_activity(activity))
+            await first_registered.wait()
+            t2 = asyncio.create_task(adapter._handle_message_activity(activity))
+            # Give the second handler a tick to register.
+            await asyncio.sleep(0)
+            release_handlers.set()
+            await asyncio.gather(t1, t2)
+
+        await _drive_two_handlers()
+
+        # Two distinct sessions were created.
+        assert len(registered_sessions) == 2
+        first_session, second_session = registered_sessions
+        assert first_session is not second_session
+        # Pin upstream's plain-Map clobber semantics: BOTH in-flight
+        # handlers, when they look up the registry post-overlap, see the
+        # SECOND session — the first's entry was overwritten in place.
+        # If a future change adds per-thread queueing/locking it must be
+        # a deliberate decision (i.e. update this test).
+        assert post_registration_snapshots == [second_session, second_session]
+        # After both handlers exit, registry is empty. Handler 2's
+        # finally-block matches ``current is session_2`` and pops; handler
+        # 1's finally-block sees the entry already gone (or not its own)
+        # and skips the pop — either way the dict ends empty.
+        assert tid not in adapter._active_streams