feat(teams): migrate native streaming to the SDK IStreamer + unwind transitional divergences (#93 PR 3/4)#145
Conversation
…ransitional 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.
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (11)
📝 WalkthroughWalkthroughReplaces the custom ChangesTeams SDK IStreamer Migration
Sequence Diagram(s)sequenceDiagram
participant BotFramework as Bot Framework
participant Handler as _handle_message_activity
participant CreateStreamer as _create_streamer
participant SDKSend as activity_sender.create_stream
participant Registry as _active_streams
participant StreamViaEmit as _stream_via_emit
participant Thread as Thread._handle_stream
BotFramework->>Handler: DM activity received
Handler->>CreateStreamer: activity, thread_id
CreateStreamer->>SDKSend: build_conversation_reference + create_stream
SDKSend-->>CreateStreamer: StreamerProtocol
CreateStreamer-->>Handler: StreamerProtocol
Handler->>Registry: register[thread_id] = StreamerProtocol
Handler->>Thread: process_message (streamer alive in registry)
Thread->>StreamViaEmit: stream(), finds streamer in _active_streams
loop each text chunk
StreamViaEmit->>SDKSend: stream.emit(text)
end
StreamViaEmit-->>Thread: RawMessage(first_id)
Thread-->>Thread: SentMessage.text = accumulated (not raw_result.text)
Handler->>Registry: remove[thread_id]
Handler->>SDKSend: await stream.close()
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Poem
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| # exits cleanly without sending more chunks. | ||
| session.cancel() | ||
| raise | ||
| await processing_done |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request migrates Microsoft Teams native streaming from a hand-rolled wire format implementation to the official Teams SDK IStreamer (StreamerProtocol). This simplifies the codebase by removing the custom _TeamsStreamSession class, the hand-rolled _teams_send REST helper, and custom throttling logic, aligning the Python SDK closer to upstream parity. Feedback was provided to address a potential issue in _stream_via_emit where id_captured could hang indefinitely if the stream is closed or cancelled before any chunk is delivered, suggesting an on_close callback to safely resolve the future.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| async def _on_chunk(activity: Any) -> None: | ||
| if not id_captured.done(): | ||
| id_captured.set_result(getattr(activity, "id", "") or "") | ||
|
|
||
| stream.on_chunk(_on_chunk) |
There was a problem hiding this comment.
If the stream is closed or cancelled before any chunk is delivered, id_captured will never be resolved, which can cause await id_captured to hang indefinitely. Registering an on_close callback to resolve the future with an empty string ensures that the future always unblocks when the stream terminates.
async def _on_chunk(activity: Any) -> None:
if not id_captured.done():
id_captured.set_result(getattr(activity, "id", "") or "")
async def _on_close(*args: Any, **kwargs: Any) -> None:
if not id_captured.done():
id_captured.set_result("")
stream.on_chunk(_on_chunk)
stream.on_close(_on_close)…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.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 19b8b88038
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| # text. The decision below only governs whether THIS | ||
| # chunk triggers an intermediate ``typing`` send. | ||
| stream.emit(text) | ||
| accumulated += text |
There was a problem hiding this comment.
With the real Teams SDK HttpStream, emit() only queues the activity and the actual send/cancel happens later in the background _flush() task. In a fast stream, this line adds text to accumulated before Teams has accepted it; if the user presses Stop or the SDK gets a 403 before queued chunks flush, close() no-ops for the canceled stream while Thread still records the queued suffix as sent. Please base the recorded text on confirmed delivery, or wait for/observe SDK flush or close state before returning.
Useful? React with 👍 / 👎.
| # emitted (empty stream, or stream canceled before first send). | ||
| if not session.text: | ||
| return | ||
| message_id = await id_captured |
There was a problem hiding this comment.
When the first SDK flush fails or is canceled before any chunk event fires, id_captured is never resolved. Because HttpStream.emit() returns after queuing and the failure occurs later in _flush(), this await can hang forever for a first-chunk 403/network error or immediate Stop, preventing the DM handler from reaching cleanup. Please also resolve/cancel this future from a close/error/cancel signal or add a timeout.
Useful? React with 👍 / 👎.
1 participant
What
PR 3/4 of the Teams adapter migration to
microsoft-teams-apps(#93) — the highest-risk PR. Replaces the hand-rolled Bot Framework streaming wire format with the SDK's native streamer (ctx.stream/StreamerProtocol.emit()), mirroring upstreamadapter-teams@chat@4.30.0index.ts(streamViaEmitDM path), and atomically unwinds the two transitional public-type divergences this enables.PRs 1 (inbound+auth) and 2 (outbound) are already merged to
main; this branch is off thatmain.How streaming now flows through
ctx.streamDMs (native):
_handle_message_activitydetects a DM, calls_create_streamer(activity, thread_id)which builds aConversationReferencefrom the inbound activity and callsapp.activity_sender.create_stream(ref)— the same call the SDK's ownActivityContextmakes to exposectx.stream. The result is a real SDKHttpStream._active_streams[thread_id], and the handlerawaits aprocessing_donegate (a wrappedwait_untilshim, the Python translation of upstream'sprocessingDonepromise + wrappedwaitUntil) so the streamer stays alive while the chat handler streams.Thread.stream→adapter.stream→_stream_via_emit: each non-empty chunk is handed tostream.emit(text). The SDK owns the wire format, throttle, and 429 retry. We never callstream.close().finallycallsstream.close()once — this is the lifecycle-owner role the SDK App'sprocess_activityplays upstream (app_process.py:216). Our bridge overridesserver.on_request, so we own dispatch and reproduce the close. The SDKHttpStream.closeno-ops when canceled or contentless, so closing in both success and cancel paths is safe.stream.on_chunk(...)(the Python SDK exposeson_chunk/on_close, notevents.once) and awaited only when text was emitted and the stream was not canceled.stream.canceledproperty (checked before each emit) and catchingStreamCancelledError.Group / channel / proactive (fallback): no active streamer → accumulate and post a single
post_message(SDK-backed from PR 2), matching upstream'spostMessage(threadId, { markdown: accumulated }).The two divergences, exactly how + where unwound
These had to land together — touching one without the other corrupts recorded message history.
(a)
RawMessage.textoverride — removed insrc/chat_sdk/types.py.RawMessageis now{id, thread_id, raw}, matching upstreampackages/chat/src/types.ts. The consumer insrc/chat_sdk/thread.py_handle_stream(raw_result.text if … else accumulated) is gone; the recordedSentMessageis now always built from the local accumulator (upstreamthread.tsparity). This is correct because the adapter now emits each chunk through the SDKIStreameras it is yielded, so the accumulator and what the platform shipped stay in lockstep — there is no buffered-but-unsent suffix to reconcile.(b)
update_interval_msnon-seeding — restored insrc/chat_sdk/thread.py_handle_stream. Now seedsStreamOptions(update_interval_ms=self._streaming_update_interval_ms)(500ms thread default) before merging caller options, exactly like upstreamthread.ts(vercel/chat#340). Harmless for Teams now because the Teams adapter no longer owns a quota throttle — the SDKIStreamerdoes.Also removed:
native_stream_min_emit_interval_msconfig field (adapters/teams/types.py) and_teams_send(now dead).History-fidelity tests (the #1 risk)
tests/test_thread_faithful.py::test_should_record_local_accumulator_for_native_streaming— recorded text comes from the accumulator, not the adapter return value.tests/test_thread_faithful.py::test_should_fall_back_when_adapterstream_returns_null— now assertsupdate_interval_ms == 500(wasis None).tests/test_teams_native_streaming.py::TestHistoryFidelity— three end-to-end tests driving a realThread.post(stream)against the real Teamsadapter.stream()with aFakeStreamer+ capturing thread-history: full-text recording, the precise cancellation semantics (history records exactly what the wrapping iterator pulled — emitted text plus the single chunk pulled in the iteration that detected cancellation, byte-for-byte upstream parity), andupdate_interval_msseeding.THROTTLE PARITY finding (from the SDK source)
Verified against the installed SDK (
microsoft-teams-apps==2.0.13), the SDKHttpStreamthrottles and is 429-safe, so we do not regress to rate-limit errors:http_stream.py:266— after a flush, if more is queued, the next flush is scheduled viacall_later(0.5, …)(500ms). The module docstring (http_stream.py:39-41) states this is "to ensure we dont hit API rate limits with Microsoft Teams".http_stream.py:283,290—add_stream_update(self._index)stamps the Bot FrameworkstreamSequence;self._indexincrements per stream activity.http_stream.py:285-288— each chunk send goes throughretry(..., RetryOptions(max_delay=4.0, max_attempts=8)).http_stream.py:180-201—close()waits for the queue to drain (_wait_for_id_and_queue) and the finaladd_stream_final()send also goes throughretry().A LIVE Teams check (streaming a real long response without a 429) is out of scope for this build and is flagged for the reviewers/maintainer.
Test rewrite
tests/test_teams_native_streaming.py(was ~82KB of wire-format/throttle-internal assertions) rewritten to mockctx.streamwith aFakeStreamerStreamerProtocoldouble (exposingemit/canceled/closed/on_chunk). Asserts: emit per chunk (string / markdown_text dict / dataclass forms);close()NOT called bystream(); first-chunk-id capture; empty-stream does-not-hang; DM-onlyprocessing_doneblocking + registration/teardown; group/proactive buffered fallback; both cancellation paths; streamer construction from the inbound activity; pass-interaction isolation. Dropped the wire-format/throttle-internal assertions. All other Teams tests updated to the SDK-backed paths and kept passing.SDK-forced divergence / residual risk
ctx.stream; our bridge owns dispatch (server.on_requestoverride from PR 1), so we reproduce the close in the handlerfinally. Behaviorally identical (no double-close — the SDK no-ops a second close).on_chunk/on_closehandler registration, notevents.once("chunk", …). Mapped 1:1 via a Future resolved by the firston_chunkcallback.Gauntlet
ruff check— cleanruff format --check— 257 files formattedaudit_test_quality.py— 0 hard failures (pre-existing cross-adapter duplicate warnings only)pyrefly check— 0 errorspytest— 4773 passed, 3 skippedverify_test_fidelity.py --strict— 732/732 (100%), exit 0Summary by CodeRabbit
Bug Fixes
Documentation
Chores