feat(teams): native streaming for DMs via emit (vercel/chat#416)

[patrick-chinchill]

Summary

Ports upstream vercel/chat#416 (commit ed46bae) — feat(adapter-teams): native streaming for DMs via emit — into the Python Teams adapter.

DM threads now dispatch chunks through the Bot Framework streaming protocol (typing activities with channelData.streamType=streaming + streamSequence, then a final message activity with streamType=final) instead of routing through Thread._fallback_stream. Group chats / channels accumulate the stream and post a single message — matching upstream's post-#416 behavior of avoiding the post+edit flicker where Teams doesn't support native streaming.

Lifecycle (hazard #5: async task lifecycle)

• _handle_message_activity registers a _TeamsStreamSession for DMs and awaits the chat handler so the session stays alive through stream(). The handler is fire-and-forget for non-DM threads.
• _stream_via_emit emits cumulative text per chunk; streamId is captured from the first send and threaded through subsequent chunks (hazard fix: launch must-fix items — security, perf, docs #7 — the first chunk omits the key entirely).
• _close_stream_session posts the final activity. Skipped on cancellation or zero-chunk streams (no orphan finalizers).
• Iterator exceptions cancel the session and re-raise; transient send failures (e.g. 429 mid-stream) cancel the session and return the partial RawMessage instead of bubbling up.

Non-parity updates

docs/UPSTREAM_SYNC.md: scoped existing "Fallback streaming" rows to non-Teams adapters, and added two new Teams-specific divergences:

• Group chats accumulate-and-post (vs upstream's chunked-edit fallback).
• Native streaming soft-cancels on transient send failure (returns partial RawMessage rather than re-raising).

Tests

tests/test_teams_native_streaming.py (665 LOC, new) covers:

• Wire-format invariants (channelData.streamType / streamSequence shape, omitted vs present streamId).
• Dispatch decisions (DM → native, group chat → accumulate-and-post).
• Cancellation paths (caller cancels mid-stream, indicator clears).
• Error mid-stream (iterator raises, send fails 429).
• Very-short streams (one chunk).
• Two concurrent DMs (no cross-contamination of streamId / sequence).
• End-to-end webhook → process_message → stream() → _close_stream_session lifecycle.

Existing TestStream cases in test_teams_coverage.py and test_teams_extended.py updated for the new accumulate-and-post semantics on group chats.

Test plan

• uv run ruff check src/ tests/ scripts/
• uv run ruff format --check src/ tests/ scripts/
• uv run python scripts/audit_test_quality.py
• uv run pytest tests/ --tb=short -q

Upstream ref: vercel/chat#416 (commit ed46bae)

https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj

---

Generated by Claude Code

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 55f6e504-5c97-47db-a677-cff1de53b4e7

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch claude/port-teams-native-dm-streaming-J7S7H

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[gemini-code-assist]

Code Review

This pull request implements native streaming for Microsoft Teams DMs using the Bot Framework protocol, while transitioning group chats and channels to an accumulate-and-post strategy to eliminate UI flicker. It introduces a session management system for tracking in-flight DM streams and includes comprehensive test coverage for the new streaming lifecycle. Feedback suggests simplifying the stream finalization logic by removing a redundant check for empty content, as the presence of a stream ID already implies that content was sent.

[gemini-code-assist]

The check not session._text is redundant here. session.stream_id is only set when a chunk with content is sent, which also guarantees session._text will be non-empty. Therefore, checking for session.stream_id is None is sufficient.

Suggested change

	if session.stream_id is None or not session._text: # noqa: SLF001
	if session.stream_id is None:

[patrick-chinchill]

Code review — feat(teams): native streaming for DMs via emit (vercel/chat#416)

Compared HEAD (58b7fc7, including the follow-up _text-check refactor) to upstream ed46bae at f55378a (packages/adapter-teams/src/index.ts). Walked the porting hazards (especially #5/#7) and the SELF_REVIEW adversarial checks (pass-interaction, emit/parse symmetry, sentinels).

✅ Looks good

• Hazard #7 (omit vs None) is honored: the first chunk's channelData is constructed without a streamId key (if session.stream_id is not None: channel_data["streamId"] = ...), and test_first_chunk_omits_stream_id pins the wire shape with assert "streamId" not in first_payload["channelData"].
• streamSequence starts at 1 and increments per emitted chunk (session.sequence += 1 before building the payload, after the empty-skip), matching the Bot Framework protocol and pinned by test_subsequent_chunks_carry_stream_id_and_increment_sequence.
• DM detection matches upstream byte-for-byte — both Python is_dm and TS isDM use not conversationId.startsWith("19:") (not conversationType == "personal" as the brief implied). Group-chat IDs (19:...@thread.skype) correctly fall through to accumulate-and-post.
• Invariant in the follow-up refactor holds: stream_id is only set after a successful _teams_send AND only when result.get("id") is non-empty (line 1174–1178); empty chunks are skipped before accumulated += text. So stream_id is not None ⇒ _text non-empty. The not session._text check was indeed redundant.
• Cleanup-path safety: the finally in _handle_message_activity uses if current is session before popping (so a concurrent re-registration on the same DM thread isn't clobbered), and _close_stream_session is wrapped in try/except so its failure can't leak the registry entry. Cancel/iterator-exception paths both call session.cancel() before re-raising, so _close_stream_session correctly skips the final activity.
• Soft-cancel-on-send-failure divergence is documented in docs/UPSTREAM_SYNC.md non-parity table with rationale and tested by test_emit_send_failure_cancels_session (verifies no exception bubbles, result.raw["text"] carries pre-failure content).
• Group-chat accumulate-and-post is a parity behavior with upstream post-#416, not a divergence — the doc table row marks it "no divergence at the adapter level" for clarity, which is the right framing.

🟡 Medium

• Concurrent-DM test doesn't exercise _active_streams race — test_two_concurrent_dm_streams_have_independent_sessions constructs two sessions for two distinct thread IDs and calls _stream_via_emit directly with explicit session args, bypassing _active_streams entirely. The realistic race (two near-simultaneous webhooks for the same DM thread overwriting _active_streams[tid]) isn't covered. This race exists in upstream too (activeStreams.set(threadId, ctx.stream) overwrites identically), so it's parity behavior — but the test name oversells what's being verified. Suggest renaming or adding a same-thread test that drives two _handle_message_activity calls under asyncio.gather.
• First-chunk send returns id="": if Teams accepts the typing activity but returns an empty id, session.stream_id stays None (line 1177 guard), _text becomes non-empty, and _close_stream_session skips because of the (now-tightened) stream_id is None check. Net effect: user sees streamed chunks but no final message activity — the streaming UI stays running until Teams times it out client-side. Upstream has the same shape (messageId = "") but would still attempt the final send. Worth a one-line comment at the close-skip site explaining this, and ideally a test pinning the empty-id behavior.

🔵 Nit

• _chained_wait_until invokes upstream_wait_until(task) before _resolve_processing(task). If the caller-supplied wait_until raises, the exception propagates back through process_message, hits the adapter's outer try, and falls into finally (no hang — await processing_done isn't reached). That's fine, but reversing the order — resolve first, then call upstream — would make the deadlock-immunity argument trivially obvious to a future reader.
• _TeamsStreamSession._text is read from outside the class via session._text with # noqa: SLF001 — a public text attribute (or method) would be cleaner since the leading underscore isn't really enforcing privacy.

---

Posted by an automated reviewer agent. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj

---

Generated by Claude Code

[patrick-chinchill]

Re-review of fixes in 2e96fbb

Verified the four fixes against ed46bae (vercel/chat#416) and the broader adapter surface.

Verified — fixes land cleanly

1. test_same_thread_concurrent_handlers_clobber_active_stream (tests/test_teams_native_streaming.py:722) — drives two _handle_message_activity tasks for the same thread under a barrier, snapshots the registry post-overlap, and pins [second_session, second_session]. Matches upstream's Map clobber semantics. The renamed test_distinct_dm_threads_each_have_isolated_session_state makes the isolation-vs-race split explicit.
2. Empty-id final-send (adapter.py:1239–1271, test 317–362) — _close_stream_session now ships the final message when text is non-empty even if stream_id is None, omitting streamId from channelData (rather than serializing None — Hazard #7). Regression test asserts "streamId" not in final_payload["channelData"]. Non-parity row at UPSTREAM_SYNC.md:492 is accurate.
3. _chained_wait_until order (adapter.py:462–471) — _resolve_processing(task) runs before upstream_wait_until(task), with a comment explaining the deadlock-immunity argument.
4. Public text property (adapter.py:95–103, 1243, 1260) — read-only text accessor exposed; reads in _close_stream_session go through it; only the canonical mutator at adapter.py:1216 retains # noqa: SLF001. Tests still poke _text directly (lines 176, 300) — fine, they're inside the test module's own laxer scope and lint-clean.

Findings

Nit (parity gap, pre-existing across multiple adapters)
_stream_via_emit and the accumulate-and-post fallback both check isinstance(chunk, dict) and chunk.get("type") == "markdown_text" (adapter.py:1081, 1139). A real MarkdownTextChunk dataclass instance silently drops to "" because isinstance(MarkdownTextChunk(...), dict) is False. Other adapters (google_chat, whatsapp, slack, github) use hasattr(chunk, "type") for this. Same gap exists in discord/linear, so out of scope here, but worth a follow-up issue — the StreamChunk union in types.py:798 is dataclass-typed.

Nit (divergence not in UPSTREAM_SYNC.md)
_chained_wait_until resolves the internal gate before invoking the caller-supplied wait_until. Upstream TS does the opposite (baseOptions?.waitUntil?.(task) first, then task.then(...)). Functionally equivalent (both register callbacks; neither blocks), but it IS a deliberate ordering divergence and the commit message frames it as a hardening defense — consider a one-line entry in the non-parity table so future syncs don't "correct" it back.

Nit (forward-looking)
chained_options = WebhookOptions(wait_until=_chained_wait_until) (adapter.py:473) drops any other fields that may exist on the inbound options. Today WebhookOptions only has wait_until, so no observable bug — but upstream uses {...baseOptions, waitUntil: ...} for a reason. A replace(options, wait_until=...) (or explicit comment "WebhookOptions has only wait_until — re-spread if it grows") would future-proof.

Hunt results — no other issues

• Wire format: streamSequence (1-indexed, ++ per emit), streamType ("streaming"/"final"), streamId (omitted on first chunk + when server returned empty id), parallel streaminfo entity — all match the Bot Framework streaming protocol that upstream's IStreamer emits under the hood.
• Cleanup: try/finally in _handle_message_activity mirrors upstream's try { await processingDone } finally { activeStreams.delete }; asyncio.CancelledError cancels the session and re-raises (covered by test_cancelled_error_propagates_and_marks_session_canceled).
• DM detection: is_dm checks only not conversationId.startswith("19:") — matches thread-id.ts:30 exactly.
• Test fidelity: upstream PR #416 added zero new it() blocks in index.test.ts; only assertion swaps in replay-streaming.test.ts. Python's 23 native-streaming tests + 159 teams-streaming-related tests all pass; no missing TS coverage.
• Hazard #5: no untracked tasks in the new code. process_message tracks via _active_tasks, _chained_wait_until only calls add_done_callback, no bare create_task / ensure_future.

Re-review verdict: PASS

The three nits are real but non-blocking (one is pre-existing, one is hardening with no observable behavior change, one is forward-looking). Fixes for the previous round all landed correctly.

Posted by an automated re-reviewer agent. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj

---

Generated by Claude Code