diff --git a/CHANGELOG.md b/CHANGELOG.md index a5275ee..5a63bdd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ Sync wave from `chat@4.31.0` to `chat@4.41.1` (tracking #184). `UPSTREAM_PARITY` stays `4.31.0` until #203. +- **Teams outbound: reactions, targeted ephemeral messages, placeholder-aware native streaming** (#219). Ports vercel/chat `5eb8b846` (#734), `160140e3` (#737) and `93a58af5` (#709) from chat@4.35.0, and `85089037` (#951, chat@4.41.0). A live Teams check is pending. + - **Consumer-visible (Teams):** `add_reaction` / `remove_reaction` now call the Teams conversations reactions API instead of logging a warning, and can raise (`AuthenticationError`, `AdapterPermissionError`, `NetworkError`, `AdapterRateLimitError`). `check`, `eyes`, `pin`, `rocket`, `thinking`, `thumbs_up` and `x` map to their Teams IDs; any other name is sent as a native Teams reaction ID (`like`, `heart`, `1f440_eyes`, ...). + - **Consumer-visible (Teams):** `TeamsAdapter.post_ephemeral` is new. In channels and group chats, `thread.post_ephemeral(user, ...)` now sends a native targeted message that only `user` sees (`used_fallback=False`), instead of opening a DM. The app must be installed in that conversation. In a 1:1 chat it posts normally with `used_fallback=True`. Edits and deletes of a targeted message go through the targeted endpoint: the sent id is recorded in the state adapter under `teams:targetedActivity:{conversationId}:{messageId}` for 24 hours (the key upstream uses), and state errors fall back to the plain endpoint. + - **Consumer-visible (Teams streaming):** only apps that set `fallback_streaming_placeholder_text` explicitly see a change. In a DM with a native streamer, the placeholder is sent as the Teams informative status line before the first chunk. In group chats, channels and proactive messages, `TeamsAdapter.stream()` returns `None`, so core posts the placeholder and edits it as text arrives, instead of posting one buffered message. `""` counts as set. Unset or `None` keeps the previous behaviour exactly. `TeamsAdapter.stream()` now returns `RawMessage | None`. + - **Python-specific (divergence from upstream):** a reaction ID that is not `[A-Za-z0-9_-]+` raises `ValidationError` before any request, because the Python SDK puts it in the URL path unescaped. `_stream_via_emit` waits at most 30 s (`STREAM_FIRST_CHUNK_ID_TIMEOUT_S`) for the first chunk's id, then settles the stream and, if nothing was delivered and the user did not cancel, posts the accumulated text as one ordinary message: on `microsoft-teams-apps` 2.0.16+, a terminal 403 on the first flush (for example "Content stream is not allowed") never delivers a chunk and never sets `canceled`, so the wait used to hang the DM handler, and the reply was never delivered. See `docs/UPSTREAM_SYNC.md`. + - Works on `microsoft-teams-apps` 2.0.13 to 2.1: reactions use `conversations.add_reaction` / `delete_reaction` where the SDK has them (2.0.16+) and `ApiClient.reactions` otherwise. - **Turn cancellation, typing lifecycle and agent-session events** (#201; core half of vercel/chat `2ce2be00` #862, chat@4.39.0). No default behavior change: state polling runs only for adapters that set `supports_turn_cancellation` (none in-repo yet; Slack lands in #215). - **New `thread.signal`** (a `TurnSignal`: `aborted`, `await signal.wait()`, `add_listener(cb)`), aborted when the turn is stopped. Pass it to model calls so a stop also stops generation. Streams posted from the turn stop being consumed when it is aborted: the reply is finalized with what was already streamed, and the source generator is closed. Outside a message handler the signal is never aborted. - **New `chat.abort_turn(thread_id)`**: aborts the thread's running turn in this process and, when the turn's adapter opted in, in any process sharing the state backend (`active-turn:` / `abort-turn:` keys, 1 h TTL, polled every 250 ms per running turn). diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index 1ec5334..28a9de9 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -1449,6 +1449,57 @@ and the Teams part of `4cc3445c` (#779, chat@4.37.0). `d4c52cad` (#652) is - **Divergences** (rows in the non-parity table): the per-URL client cache and `TeamsFormatConverter.to_ast` using the bounded stripper. +### Teams outbound: reactions, targeted ephemeral, placeholder streaming (chat@4.35–4.41, #219) + +Ports `5eb8b846` (vercel/chat#734), `160140e3` (#737) and `93a58af5` (#709) +from chat@4.35.0, and `85089037` (#951, chat@4.41.0). Two divergence-table +rows: the reaction-ID check and the bounded first-chunk wait. + +- **Reactions (`5eb8b846`).** `add_reaction` / `remove_reaction` resolve the + emoji through `_TEAMS_REACTION_ALIASES` (`check`, `eyes`, `pin`, `rocket`, + `thinking`, `thumbs_up`, `x`); other names pass through as native Teams IDs. + The call goes to the thread's service-URL client (`_api_for`, #216) after + the SSRF check, and failures go through `_handle_teams_error(error, + "addReaction" / "removeReaction")`. SDK adaptation: `microsoft-teams-api` + 2.0.16+ has `conversations.add_reaction` / `delete_reaction` (upstream's + `conversations.addReaction`), and 2.0.13 has only `ApiClient.reactions.add` + / `delete`, which 2.0.16+ deprecates. Both send `PUT` / `DELETE + /v3/conversations/{id}/activities/{id}/reactions/{type}`. +- **Targeted ephemeral (`160140e3`).** `post_ephemeral(thread_id, user_id, + message, *, options=None)` checks `is_dm` first, so a 1:1 chat (including a + `19:` ID with an explicit `personal` type) posts normally with + `used_fallback=True` and never reaches the SDK's personal-chat guard. + Otherwise it builds the same text or card payload as `post_message`, calls + `with_recipient(Account(id=user_id, name=user_id, role="user"), True)` and + sends it through `_send_to`. The SDK then posts to + `activities?isTargetedActivity=true` (`ActivitySender.send` on 2.0.x, + `send_or_update_activity` on 2.1). `role` is a field on 2.1's `Account` and + an extra field on 2.0.x, so it is on the wire with both. `raw` is the + activity dict, as for `post_message`. `options` is ignored, as upstream. + The SDK's `with_recipient(..., True)` emits its own `ExperimentalWarning` + (a `FutureWarning`) for the preview targeted API. It is not suppressed. +- **Targeted mutation (`85089037`).** A targeted send records + `teams:targetedActivity:{conversationId}:{messageId}` = `"1"` with a 24 h + TTL (Teams deletes targeted messages after 24 hours). `edit_message` / + `delete_message` read it and call `activities(conv).update_targeted` / + `delete_targeted` instead of `update` / `delete`. A successful delete + forgets the record; a failed one keeps it. Every state access is best + effort, as upstream: a read error means "not targeted", and write or delete + errors are ignored. The key format is upstream's, so state can be shared + with the TS SDK. +- **Placeholder streaming (`93a58af5`).** `stream()` reads + `options.fallback_streaming_placeholder_text` (the #199 `UNSET` / `None` / + `str` value). With an active, non-canceled streamer it is passed to + `_stream_via_emit`, which calls the synchronous `stream.update(text)` before + the first chunk when it is a `str`. With no streamer and a `str`, `stream()` + returns `None` before reading the stream, and core's post+edit fallback + posts the placeholder. `UNSET` and `None` keep the previous paths. A + canceled streamer still falls through to the buffered post, as upstream. + `""` is a `str`, so it takes the new paths too (upstream `typeof === + "string"`). +- **First-chunk wait.** The issue comment on #219 (found while reviewing #262) + is fixed here: see the divergence row. + ### Teams installation lifecycle and bot joins (chat@4.40–4.41, #217) Ports `aaeede70` (vercel/chat#899, chat@4.40.0) and the Teams handler half of @@ -2140,7 +2191,7 @@ Of the adapter halves, only Slack's no-recipient guard is ported here: threads (`chat.thread(id)`, `open_dm`, action/reaction threads) reply through core's post+edit fallback instead of raising `ValidationError`. The rest of Slack's half is in "Slack streaming fallback (#207)" below. Teams informative -status is #219. +status is ported in #219. - **Lightweight threads (`438f5513`).** `_create_thread` takes `Message | None`. `chat.thread(id)` without `current_message`, `open_dm`, @@ -3044,21 +3095,23 @@ stay explicit instead of being rediscovered in code review. | Teams inbound activity routing (issue #93 PR 1) | `BridgeHttpAdapter.dispatch` feeds webhooks through the SDK `HttpServer` (JWT validation), but the adapter **overrides `app.server.on_request`** with `_dispatch_activity` instead of letting the SDK's default router run. Our callback dumps the lenient `CoreActivity` to a camelCase dict and routes by `type` to the existing handler logic. | Upstream `@chat-adapter/teams@4.30.0` registers `app.on("message" / "card.action" / …)` and lets `@microsoft/teams.apps` route via its typed dispatcher; handlers receive `ctx.activity` as a strongly-typed `IMessageActivity` etc. | The Python SDK's default `on_request` (`App._process_activity_event` → `ActivityProcessor.process_activity`) runs `ActivityTypeAdapter.validate_python` (strict per-activity validation — `recipient`/`id` required) **and** a live `api.users.token.get` network call inside `_build_context` before any handler fires. Minimal serverless webhook payloads (and the adapter's dict-based handler logic) can't survive strict validation, and the token fetch would make an unwanted outbound call per inbound activity. We keep the SDK as the **auth + transport** layer (JWT genuinely validated by its `TokenValidator`) but route the already-authenticated activity ourselves through the lenient `CoreActivity`, preserving the exact pre-migration handler behavior. The SDK `on_message`/`on_card_action`/… decorators are still registered for parity/forward-compat. Regression coverage: `tests/test_teams_extended.py::TestActivityTypes`, `tests/test_teams_coverage.py::TestSdkInboundAuth`, `tests/test_teams_bridge.py`. | | Teams dialog/modal inbound (issue #93 PR 1) | `on_dialog_open` / `on_dialog_submit` are registered on the SDK App but only cache user context (no `process_modal_submit`, no task-module response) | Upstream `@chat-adapter/teams@4.30.0` `handleDialogOpen`/`handleDialogSubmit` drive `chat.processModalSubmit` + `modalToAdaptiveCard` | The pre-migration Python Teams adapter never implemented modal/dialog inbound processing, so PR 1 (inbound + auth plumbing) preserves that behavior rather than introducing new modal handling. Wiring dialogs to `chat.process_modal_submit` is tracked as a later wave of the #93 migration. | | Teams dialog-open action options (4.41 wave, #191) | N/A: dialog-open inbound is not ported (see the row above), so there is no `process_action` call with `on_open_modal` to spread `WebhookOptions` into or to guard against a rejected action task | `c21ccbc0` spreads `...webhookOptions` into `handleDialogOpen`'s `processAction` call; `91683e52` resolves the empty dialog response when that action rejects | Lands with the dialog inbound wave of the #93 migration. The DM message path does spread the caller's options (see the native streaming row). | -| Fallback streaming with whitespace-only streams (non-Teams adapters) | Placeholder cleared to `" "` on final edit | Placeholder left visible (`"..."` stuck) | Upstream 4.26 guards against empty edits but leaves the placeholder stranded on the message. We issue one final `edit_message(" ")` so the placeholder disappears when no real content was produced. Teams does not route through `_fallback_stream` (DMs stream natively through the SDK `IStreamer`; group chats accumulate-and-post), so this divergence applies only to Slack / Discord / GitHub / Google Chat / Linear / WhatsApp. Telegram has streamed every chat through its own post-and-edit loop since the 4.41 wave (#226, vercel/chat#822); like upstream's, that loop leaves the `"..."` placeholder on a whitespace-only stream. | +| Fallback streaming with whitespace-only streams (adapters using core post+edit, including Teams with an explicit placeholder) | Placeholder cleared to `" "` on final edit | Placeholder left visible (`"..."` stuck) | Upstream 4.26 guards against empty edits but leaves the placeholder stranded on the message. We issue one final `edit_message(" ")` so the placeholder disappears when no real content was produced. It applies to Slack / Discord / GitHub / Google Chat / Linear / WhatsApp, and to Teams when `stream()` returns `None` (no active streamer and an explicit `str` `fallback_streaming_placeholder_text`, #219: group chats, channels, proactive DMs). Teams DMs with a streamer stream natively through the SDK `IStreamer`, and Teams without a placeholder accumulates and posts once, so neither reaches `_fallback_stream`. Telegram has streamed every chat through its own post-and-edit loop since the 4.41 wave (#226, vercel/chat#822); like upstream's, that loop leaves the `"..."` placeholder on a whitespace-only stream. | | Google Chat `` round-trip | `to_ast()` / `extract_plain_text()` parse the custom-label syntax back to a link node / bare label | `toAst()` / `extractPlainText()` leave `` as raw text (or parse the whole string as an autolink with a malformed URL) | Upstream 4.26 emits `` in `from_ast` but never taught the reverse direction to parse it. A message posted with `[label](url)` then read back through `fetch_messages` comes back as unstructured text (or worse, a link node with the full `url\|text` as its URL) in upstream. We close the round-trip via an AST placeholder substitution: each `` is extracted to a private-use sentinel, Markdown is parsed on the rest, and link nodes are injected where the sentinels landed. This avoids the Markdown parser's incomplete handling of balanced-parens link destinations, so URLs like `https://en.wikipedia.org/wiki/Foo_(bar)` round-trip intact. | | Restored thread/channel ownership (4.41 wave, #199) | `ThreadImpl.from_json` / `ChannelImpl.from_json` resolve ownership when called: an explicit `chat` with an explicit `adapter` it did not register raises `RuntimeError` ("does not belong to this Chat instance") before anything is built; an explicit `adapter` alone is matched against the active Chat at once (and again on access); modal-context restore binds the event's adapter with `chat=self` when this Chat registered it under any key, never the active singleton | `ChatBinding` resolves ownership on first access: the same error is thrown by the first state read or write, and the singleton is consulted only then | `chat.activate()` is scoped to a `contextvars.Context`, so a thread restored inside the block and used after it would find no Chat, or a different one, if ownership were resolved lazily. Raising in `from_json` also leaves an existing instance untouched on an idempotent rebind. Regression tests: `tests/test_chat_resolver.py::TestContextVarActivation::test_explicit_adapter_restored_inside_activate_keeps_its_owner`, `tests/test_serialization.py::TestThreadFromJsonFaithful::test_should_raise_before_rebinding_when_chat_does_not_own_the_adapter`, `::TestRestoredRuntimeOwnershipFixes::test_throws_for_an_explicit_chat_that_does_not_own_the_explicit_adapter` and `::TestRevivedStreamingConfiguration::test_modal_context_for_an_adapter_under_a_custom_key_stays_with_its_chat`. | | `from_json(data, adapter=X)` → `_adapter_name` | Updated to `X.name` so `to_json()` reflects the bound adapter | Kept at `json.adapterName`, so re-serialization can emit a name that no longer matches the actual adapter | Upstream TS has the same gap but only exposes it via the `fromJSON(json, adapter?)` overload. In Python we lean on this API more (explicit `chat=` / explicit `adapter=` is preferred over the singleton). We sync the name on rebind so runtime and serialize agree. | | Google Chat link labels with `\|` / `>` / `]` / newline, empty labels, URLs without a scheme, or URLs containing `\|` / `>` | Fall back to `text (url)` (or bare URL for empty labels) when the `` form can't round-trip safely | Always ``, producing malformed or un-parseable output | Google Chat's `` has no escape for `\|` or `>`; `]` breaks our own `to_ast()` regex (which converts `` to Markdown `[text](url)`, and Markdown closes the label at the first `]`); newline breaks the single-line form; schemeless URLs and URLs containing `\|`/`>` don't match our reverse parser. Upstream emits the malformed form regardless; we fall back to the pre-4.26 `text (url)` form (or the bare URL for empty labels) so the label/URL stays intact and Google Chat's auto-link detection still fires for http(s). | | Google Chat heading rendering | `#`-headings emit as `*text*` (bold) so they're visually distinct | Falls through to default node-to-text (plain concatenation) | Google Chat has no heading syntax; emitting plain text loses the visual hierarchy. Bold is the closest approximation the platform supports. | | Google Chat image rendering | Images emit as `{alt} ({url})` or bare `url` | No image branch — falls through to default which concatenates children only, dropping the URL | Upstream silently drops image URLs when rendering to Google Chat text. We preserve the URL so the message content isn't lost. | -| Fallback streaming stream-exception capture (non-Teams, non-Telegram adapters) | `_fallback_stream` captures exceptions from the stream iterator, flushes whatever content was already rendered, awaits `pending_edit`, and re-raises after cleanup | `try/finally` only — exception propagates immediately, `pendingEdit` is un-awaited, and the placeholder is stranded as `"..."` | Upstream leaves a hard UX failure when streams crash mid-flight (common: LLM connection drops): placeholder visible forever, orphan background task. We flush + clean up before re-raising so the caller still sees the original error and users see the partial content instead of a spinner. This divergence does not apply to Teams: Teams DMs stream natively through the SDK `IStreamer` (`_stream_via_emit`), and a non-cancel iterator exception propagates straight to the caller while the SDK closes the streamer after the handler returns. Nor does it apply to Telegram: since the 4.41 wave (#226, vercel/chat#822) every Telegram chat streams through the adapter's own post-and-edit loop, where, as in upstream (`adapter-telegram/src/index.ts` `postAndEditStream`), a text-stream exception propagates without a partial flush, so a stream that fails before the first paced edit leaves the `"..."` placeholder visible. | +| Fallback streaming stream-exception capture (non-Telegram adapters using core post+edit, including Teams with an explicit placeholder) | `_fallback_stream` captures exceptions from the stream iterator, flushes whatever content was already rendered, awaits `pending_edit`, and re-raises after cleanup | `try/finally` only — exception propagates immediately, `pendingEdit` is un-awaited, and the placeholder is stranded as `"..."` | Upstream leaves a hard UX failure when streams crash mid-flight (common: LLM connection drops): placeholder visible forever, orphan background task. We flush + clean up before re-raising so the caller still sees the original error and users see the partial content instead of a spinner. It applies to Teams only when `stream()` returns `None` (no active streamer and an explicit `str` placeholder, #219), which routes the reply through `_fallback_stream`. Teams DMs with a streamer stream natively through the SDK `IStreamer` (`_stream_via_emit`), where a non-cancel iterator exception propagates straight to the caller while the SDK closes the streamer after the handler returns; Teams without a placeholder accumulates and posts once. Nor does it apply to Telegram: since the 4.41 wave (#226, vercel/chat#822) every Telegram chat streams through the adapter's own post-and-edit loop, where, as in upstream (`adapter-telegram/src/index.ts` `postAndEditStream`), a text-stream exception propagates without a partial flush, so a stream that fails before the first paced edit leaves the `"..."` placeholder visible. | | Slack `stream()` on Enterprise Grid (`chat.startStream` `team_not_found`) | Threads the workspace `team_id` into `client.chat_stream(...)` (= `options.recipient_team_id`, the `team.id` extracted on the inbound path), which slack_sdk forwards into `_stream_args` → `chat.startStream`. `chat.appendStream`/`chat.stopStream` don't receive `_stream_args` and don't need `team_id`. Harmless on non-Grid workspaces — a correct `team_id` is always valid. Since #208 every stream segment (each `chat_stream` call when a long reply rotates) gets the same `team_id`. Since #207 it is sent only when `recipient_team_id` is truthy (a DM streams natively without recipient context); a Grid DM without one that hits `team_not_found` is delivered by the first-call post+edit fallback. (chat-sdk-python#95) | Builds the `chat.startStream` args from `channel`/`threadTs`/`recipientUserId`/`recipientTeamId`/`taskDisplayMode` only (`adapter-slack/src/index.ts` `stream()`); never passes a workspace `team_id`. On Grid orgs `chat.startStream` then fails with `team_not_found` (the per-workspace bot token alone isn't sufficient to disambiguate the team), even though `chat.postMessage` on the same workspace succeeds without it. | Upstream has the same gap — its `stream()` never threads `team_id`, so streaming is broken on Grid while non-streaming posts work. `chat.startStream` requires `team_id` for Grid disambiguation; `chat.postMessage` does not, which is why only streaming regresses. We source `team_id` from the already-plumbed `recipient_team_id` (the workspace where the interaction happened = the streaming target workspace). Live verification needs a real Grid workspace; the unit regression (`tests/test_slack_api.py::TestStream::test_stream_threads_team_id_to_chat_stream_for_grid` + the `team_not_found` mutation guard `test_stream_without_team_id_on_grid_falls_back_to_post_and_edit`) simulates Grid by raising `team_not_found` from the streamer's lazy `chat.startStream` when `team_id` is absent. Tracked for contribution upstream. Re-checked for #268: upstream 4.35+ injects an org-wide event's `team_id` through `withToken`, but its `chatStream` call does not go through `withToken`, and ours does not go through `_with_token_kwargs`, so `chat_stream` keeps `team_id = recipient_team_id` and gets no `client_context_team_id` under an org-wide context (`test_stream_keeps_the_recipient_team_id_under_an_org_wide_context`). | | Slack `stream()` fallback with an empty reply (#207) | When the post+edit fallback engaged but nothing was ever posted (e.g. an empty stream whose `stop()` failed), returns `RawMessage(id="", thread_id=thread_id, raw=None)` | Returns `fallback.message`, which is `null` when nothing was posted (`adapter-slack/src/index.ts` `stream()` fallback returns) | Returning `None` after the stream was consumed would make core run its own post+edit fallback (`thread.py` `_handle_stream`) on an exhausted iterator, posting the `"..."` placeholder and a `" "` final message for a reply that had no content. `RawMessage(id="")` keeps core's "adapter handled it" path. Pinned by `tests/test_slack_api.py::TestNativeStreamingFallback::test_warns_about_skipped_stream_blocks_and_returns_an_empty_reply_marker`. | -| 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 DMs stream through the SDK `IStreamer` and the Teams accumulate-and-post path ships raw `accumulated` via `post_message`, matching upstream; 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` (SDK-backed) instead of post+edit, even for group chats and channel threads | Same (`@chat-adapter/teams@4.30.0`: `if (activeStream && !activeStream.canceled) … else { accumulate; postMessage }`) — no divergence at the adapter level | Documented for clarity: the Python port matches upstream's behavior of avoiding the post+edit flicker where Teams doesn't support native streaming. The buffered fallback routes through the same SDK send path (`_send_to`) as a normal `post_message`. | -| Teams native streaming via the SDK `IStreamer` (DMs) | `TeamsAdapter._handle_message_activity` captures a Teams SDK `IStreamer` (`microsoft_teams.apps.StreamerProtocol` / `HttpStream`) for DMs via `app.activity_sender.create_stream(ref)` on `microsoft-teams-apps` 2.0.x, or `HttpStream(app.api.from_service_url(ref.service_url), ref)` on 2.1.x, which removed `ActivitySender` (#250), registers it in `_active_streams`, and `await`s a `processing_done` gate (a wrapped `wait_until` shim) so the streamer stays alive while the handler streams. The shim builds its options with `dataclasses.replace(options or WebhookOptions(), wait_until=...)`, so `propagate_handler_errors` / `deduplicate` reach `Chat.process_message` (upstream `{ ...baseOptions, waitUntil }`). Since #191 `wait_until` receives Chat's error-swallowing wrapper Task by default (the raw handler task with `propagate_handler_errors=True`). The gate's `add_done_callback` hooks the handler task that `process_message` returns, not the handed wrapper: a host can cancel the wrapper while the shielded handler keeps streaming, whereas upstream's `.catch` promise cannot be cancelled and always settles with the handler. The handed task is only the fallback when `process_message` returns no Task (pinned by `tests/test_teams_native_streaming.py::TestHandleMessageActivityWithRealChat`). `stream()` → `_stream_via_emit` calls `stream.emit(text)` per chunk and NEVER calls `close()`; the adapter's `_handle_message_activity` `finally` calls `stream.close()` once (the lifecycle-owner role the SDK App's `process_activity` plays upstream). | `@chat-adapter/teams@4.30.0` `index.ts` does exactly this: `this.activeStreams.set(threadId, ctx.stream)`, build `processingDone` + wrapped `waitUntil`, `await processingDone`, `streamViaEmit` calls `stream.emit(text)` and never `close()` (the SDK App auto-closes after the handler returns). | **No adapter-level divergence.** The only mechanical difference is the close call site: upstream lets the SDK `App` auto-close `ctx.stream` because the SDK owns dispatch; our bridge overrides `server.on_request`, so we own dispatch and reproduce the close in `_handle_message_activity`'s `finally`. The SDK `HttpStream.close` no-ops when the stream was canceled or had no content, so closing in both success and cancel paths is safe (matching the SDK App, which closes in both its success and `StreamCancelledError` branches). Cancellation is detected via `stream.canceled` (checked before each emit) and by catching `StreamCancelledError` (other exceptions re-raise). The first chunk id is captured via `on_chunk` and awaited only when text was emitted and the stream was not canceled. Replaces the prior hand-rolled wire format, the 1500ms emit throttle, and the `RawMessage.text` / `update_interval_ms` divergences (all unwound in #93 PR 3). | +| Fallback streaming final SentMessage content (adapters using core post+edit, including Teams with an explicit placeholder) | 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. It applies to every adapter whose `stream()` returns `None`, including Teams with no active streamer and an explicit `str` placeholder (#219: the final edit of the placeholder is remend'd). Teams DMs with a streamer stream through the SDK `IStreamer`, and Teams without a placeholder ships raw `accumulated` in one `post_message`, matching upstream. | +| Teams group-chat / channel streaming via accumulate-and-post | Without a native streamer, `TeamsAdapter.stream` accumulates the full text and issues a single `post_message` (SDK-backed) instead of post+edit, unless `fallback_streaming_placeholder_text` is an explicit `str` (`""` included): then it returns `None` before reading the stream and core posts the placeholder and edits it (#219). `UNSET` and `None` keep the buffered post | Same (chat@4.41.1 `stream`: `if (activeStream && !activeStream.canceled) streamViaEmit(…)`; `if (!activeStream && typeof options?.fallbackStreamingPlaceholderText === "string") return null`; else accumulate and `postMessage`) — no divergence at the adapter level | Documented for clarity: the Python port matches upstream's behavior of avoiding the post+edit flicker where Teams doesn't support native streaming, unless the app asked for a placeholder. The buffered fallback routes through the same SDK send path (`_send_to`) as a normal `post_message`. | +| Teams native streaming via the SDK `IStreamer` (DMs) | `TeamsAdapter._handle_message_activity` captures a Teams SDK `IStreamer` (`microsoft_teams.apps.StreamerProtocol` / `HttpStream`) for DMs via `app.activity_sender.create_stream(ref)` on `microsoft-teams-apps` 2.0.x, or `HttpStream(app.api.from_service_url(ref.service_url), ref)` on 2.1.x, which removed `ActivitySender` (#250), registers it in `_active_streams`, and `await`s a `processing_done` gate (a wrapped `wait_until` shim) so the streamer stays alive while the handler streams. The shim builds its options with `dataclasses.replace(options or WebhookOptions(), wait_until=...)`, so `propagate_handler_errors` / `deduplicate` reach `Chat.process_message` (upstream `{ ...baseOptions, waitUntil }`). Since #191 `wait_until` receives Chat's error-swallowing wrapper Task by default (the raw handler task with `propagate_handler_errors=True`). The gate's `add_done_callback` hooks the handler task that `process_message` returns, not the handed wrapper: a host can cancel the wrapper while the shielded handler keeps streaming, whereas upstream's `.catch` promise cannot be cancelled and always settles with the handler. The handed task is only the fallback when `process_message` returns no Task (pinned by `tests/test_teams_native_streaming.py::TestHandleMessageActivityWithRealChat`). `stream()` → `_stream_via_emit` calls `stream.emit(text)` per chunk and calls `close()` only after the bounded first-chunk wait expires (see the first-chunk row); the adapter's `_handle_message_activity` `finally` calls `stream.close()` once (the lifecycle-owner role the SDK App's `process_activity` plays upstream). | `@chat-adapter/teams@4.30.0` `index.ts` does exactly this: `this.activeStreams.set(threadId, ctx.stream)`, build `processingDone` + wrapped `waitUntil`, `await processingDone`, `streamViaEmit` calls `stream.emit(text)` and never `close()` (the SDK App auto-closes after the handler returns). | **No adapter-level divergence.** The only mechanical difference is the close call site: upstream lets the SDK `App` auto-close `ctx.stream` because the SDK owns dispatch; our bridge overrides `server.on_request`, so we own dispatch and reproduce the close in `_handle_message_activity`'s `finally`. The SDK `HttpStream.close` no-ops when the stream was canceled or had no content, so closing in both success and cancel paths is safe (matching the SDK App, which closes in both its success and `StreamCancelledError` branches). An explicit `str` placeholder (`fallback_streaming_placeholder_text`, #219) is sent first with the synchronous `stream.update(text)` as the informative status; `UNSET` / `None` send nothing. Cancellation is detected via `stream.canceled` (checked before each emit) and by catching `StreamCancelledError` (other exceptions re-raise; a cancel raised by `update` is caught the same way). The first chunk id is captured via `on_chunk` and awaited only when text was emitted and the stream was not canceled, for at most `STREAM_FIRST_CHUNK_ID_TIMEOUT_S` (see the first-chunk row). Replaces the prior hand-rolled wire format, the 1500ms emit throttle, and the `RawMessage.text` / `update_interval_ms` divergences (all unwound in #93 PR 3). | | Teams: `microsoft-teams-apps` 2.0.x and 2.1.x both supported (#250) | The `[teams]` extra allows `>=2.0.13,<2.2` for `microsoft-teams-{apps,api,cards,common}` (common is imported directly and apps leaves it unbounded, so it is declared and capped too). SDK differences are feature-detected, never version-sniffed. **Streaming:** `_create_streamer` uses `app.activity_sender.create_stream(ref)` when the App has one (2.0.x) and otherwise builds `HttpStream` on `app.api.from_service_url(ref.service_url)` (2.1.x, mirroring 2.1's `ActivityContext.stream`). Either way the stream owns a client pinned to the inbound service URL, separate from the per-URL clients outbound calls use (`_api_for`, #216), so an outbound call cannot redirect an in-flight stream. An SDK with neither entry point falls back to buffered posting. **Edit/delete:** `_api_for(url).conversations.activities(id).update/delete` exists on every supported version; on 2.1 it routes through `conversations.update_activity(..., service_url=None)`, which uses that client's own URL. `_api_for` builds the per-URL client with `ApiClient.from_service_url` on 2.1 and `ApiClient(url, app.api.http, ...)` on 2.0.x (#216). **Sends:** `_send_to` calls `app.activity_sender.send(activity, ref)` on 2.0.x and `send_or_update_activity(api, activity, ref)` on 2.1.x (#216). The flattened `update_activity`/`delete_activity` are not used because 2.0.13.4 (the floor) lacks them. **Inbound auth:** 2.1 replaced the Bot Framework-only `TokenValidator.for_service` with `InboundActivityTokenValidator`, which also accepts Entra ID (Agent 365 "Agent ID") tokens whose audience is the app id, from any tenant (issuer taken from the token's `tid`) and without the `serviceurl` claim check. It also picks the Entra branch from the token's *unverified* `iss`, building a per-`tid` validator and fetching that tenant's JWKS (blocking) before the signature is checked. So `BridgeHttpAdapter.dispatch` calls the adapter's `_rejects_before_auth` hook before the SDK route handler: it decodes the Bearer token without verification and answers 401 unless `iss == app.cloud.token_issuer` (an unreadable token is refused too; the SDK would refuse it). A token that passes still gets the SDK's full validation. `_dispatch_activity` repeats the issuer check on the validated `JsonWebToken` as defence in depth. On 2.0.x the SDK already enforces the issuer; the pre-check only spares the Bot Framework JWKS fetch for tokens it would reject. Requests with no `Bearer` header, and all requests in `dangerously_allow_unauthenticated_requests` / `skip_auth` mode (the SDK ignores the header there), go to the SDK unchanged. That mode includes a configured `webhook_verifier` (#221): the bridge runs the verifier first and the App is built with the skip-auth flag, so the verifier replaces this pre-check just as it replaces the SDK's JWT validation (pinned by `tests/test_teams_connect.py::TestConnectWebhookDispatch::test_verifier_replaces_the_bot_framework_issuer_precheck`) | `@chat-adapter/teams@4.41.1` still depends on `@microsoft/teams.*` `^2.0.14` and calls `ctx.stream` / `app.api.conversations.activities(...)` | Python-only SDK compatibility, not a behavior divergence: on either SDK line the adapter streams, edits and authenticates as it did on 2.0.x. Agent ID activities are not supported by this adapter, so accepting their tokens would only widen who can deliver activities. Pinned by `TestCreateStreamer` (both SDK shapes plus an unstubbed real-SDK test that builds a per-URL client mid-stream), `TestOutboundServiceUrlRouting` / `TestTeamsAppRouting` (edit/delete and a real send checked at the HTTP boundary) and `TestInboundTokenIssuer` (real bridge → SDK validator path with RS256 test-key tokens, only JWKS key resolution stubbed: Entra tokens refused with no JWKS fetched, Bot Framework tokens accepted and still audience/`serviceurl`-checked, sovereign-cloud issuer via `CLOUD=USGov`, the post-validation guard, and unauthenticated mode) and `TestSdkDependencyDeclarations` (every imported `microsoft_teams.*` package declared and capped). CI installs 2.1.x only; 2.0.16 was run locally. **A live Teams check of native DM streaming, edit and delete on 2.1 is still owed before release.** | | Teams per-service-URL client cache (`_api_for`, #216; upstream `TeamsApp.apiFor`, chat@4.41.0) | `_api_for(url)` returns one cached `ApiClient` per normalized (trailing-slash-stripped) service URL, derived from the current `App.api` (the cache is emptied when `App.api` changes) and bounded to 32 entries (oldest evicted). `App.api` itself is returned for an empty URL, its own URL, or when `api_url` / `TEAMS_API_URL` pins every call | `apiFor` builds a `new ConnectorClient(target, this.client.clone({ token }))` on every call | Avoids rebuilding a client (and its sub-clients) for every edit/delete on a regional or sovereign host. Each cached client shares the App's token source (2.0.x: the same token-bearing `HttpClient`; 2.1: `from_service_url`'s token provider), so tokens stay fresh. `_api_for` never awaits, so the cache adds no shared state across an `await` (the race #216 removes). Pinned by `tests/test_teams_adapter.py::TestTeamsAppRouting` (`test_targets_other_service_urls_with_a_dedicated_client`, `test_client_cache_follows_the_app_api_it_was_built_from`, `test_client_cache_is_bounded`). | +| Teams reaction-ID check (#219) | `add_reaction` / `remove_reaction` raise `ValidationError` before any request when the resolved reaction ID does not fully match `[A-Za-z0-9_-]+` (aliases resolve first) | `resolveTeamsReactionType` passes any string through (chat@4.41.1 `adapter-teams/src/index.ts`) | The Python SDK's `ReactionClient` / `conversations.add_reaction` interpolate the type into the URL path unescaped (`api/clients/reaction/client.py`), so `like/../x` or `like?x` would change the request. The SDK's listed reaction types (`like`, `heart`, `1f440_eyes`, `2705_whiteheavycheckmark`, `launch`, `1f4cc_pushpin`) and the alias targets all match; `-` is allowed so hyphenated IDs still pass. | +| Teams first native-stream chunk wait (#219) | `_stream_via_emit` awaits the first `chunk` id with `asyncio.wait_for(…, STREAM_FIRST_CHUNK_ID_TIMEOUT_S)` (30 s, the SDK `HttpStream` close wait); on timeout, unless the user canceled, it settles the stream with `stream.close()` (which waits for an in-flight flush and finalizes a stream that got through; the first chunk's id is still captured, since the capture future is shielded from the timeout and Teams returns the id only on the first response) and, only when that reports nothing delivered, no flush is running or scheduled (`HttpStream._lock` / `_pending` / `_timeout`; the SDK has no public signal, and a chunk a failed flush left queued does not count) and the stream is still not canceled, logs a warning and posts the accumulated text with one buffered `post_message` (returning that message) | `messageId = await idCaptured` with no bound (chat@4.41.1 `streamViaEmit`) | `microsoft-teams-apps` 2.0.16+ `HttpStream._send` sets `canceled` only for a real cancel; a terminal 403 (e.g. "Content stream is not allowed") raises out of the background flush without emitting `chunk`, so an unbounded await hangs the DM handler and the webhook never completes (2.0.13.4 set `canceled` on every 403). Settling with `close()` first means a slow first flush that eventually succeeds is not also posted, and a cancel 403 on the first flush, or one raised while `close()` finalizes, posts nothing (the `StreamCancelledError` from `close()` is caught; it subclasses `asyncio.CancelledError`, so only that subclass is caught). A stream finalized this way, or one whose text was posted instead (the SDK keeps that text buffered), is removed from `_active_streams`, so a later post in the same handler takes the no-streamer path (2.0.13.4 does not reopen a closed stream on emit). The handler's later `close()` is then a no-op (idempotent once closed, or nothing to send). Known cost: in a tenant that refuses streaming, each DM reply waits the full 30 s before the post. A first flush still in flight after `close()`'s own 30 s wait (the SDK client sets no request timeout) gets no fallback post, so it is never sent twice; if that flush never completes, the reply is lost, as upstream (which waits on it forever). Pinned by `tests/test_teams_native_streaming.py::TestFirstChunkIdWait` (real SDK `HttpStream` over an `httpx` mock transport, 2.0.x and 2.1). Found while reviewing #262. | | Teams installation token service URL (#217) | When an `installationUpdate` has no `serviceUrl`, the validated token's `service_url` is checked against the SSRF allow-list (`_validate_service_url`) before it is encoded into `InstallationEvent.channel_id` (a root-path endpoint gets back the trailing `/` the SDK token strips, so `https://host/` is not rejected as `https://host`); a disallowed one logs `"Ignoring disallowed token serviceUrl"` and leaves `channel_id` `None` | `handleInstallationUpdate` encodes `activity.serviceUrl || ctx.ref.serviceUrl` unchecked | `channel_id` is documented as safe to persist for later proactive sends, and every Python send path validates the thread's service URL, so a disallowed URL would only fail later. Not reachable with a Bot Framework-signed token in practice; the activity's own `serviceUrl` is encoded unchecked, exactly like message thread IDs. Pinned by `tests/test_teams_installation.py::TestTeamsInstallationTokenServiceUrl::test_ignores_a_disallowed_token_service_url` and `::test_restores_the_root_slash_the_token_strips_from_a_root_path_endpoint`. | | Teams `TeamsFormatConverter.to_ast` tag stripping (#216) | Strips leftover tags with the bounded `strip_html_tags` primitive (`<[^>]{1,2048}>` until stable), like the other Teams HTML-to-text paths | `markdown.ts` `toAst` still loops an unbounded `/<[^>]+>/g` until stable (only `format/` and `graph/messages.ts` adopted `stripHtmlTags` in chat@4.37.0) | Keeps every Teams inbound HTML-to-text path on one bounded scan, as the issue scoped. Only a "tag" longer than 2048 characters differs: it stays as text. Pinned by `tests/test_teams_format.py::TestToAst::test_strips_tags_with_the_bounded_stripper`. | | Teams streaming throttle / Bot Framework wire format ownership | The SDK `HttpStream` owns the entire Bot Framework streaming wire format (`streamType`/`streamSequence`/`streamId`), the inter-flush throttle, and 429 retry. We hand it text via `emit()` and read back the assigned id via `on_chunk`. | Same — `@microsoft/teams.apps`'s `IStreamer` owns all of this in the JS SDK. | **THROTTLE PARITY (verified against the installed SDK source — `microsoft-teams-apps==2.0.13.4`):** the SDK throttles and is 429-safe, so we don't regress to rate-limit errors: (1) `http_stream.py:266` — after a flush, if more is queued, the next flush is scheduled via `call_later(0.5, …)`, i.e. a 500ms inter-flush delay (the module docstring at `http_stream.py:39-41` states this is "to ensure we dont hit API rate limits with Microsoft Teams"); (2) `http_stream.py:283,290` — `add_stream_update(self._index)` stamps the Bot Framework `streamSequence` and `self._index` increments per stream activity; (3) `http_stream.py:285-288` — each chunk send goes through `retry(..., RetryOptions(max_delay=4.0, max_attempts=8))`, so transient 429s are retried with backoff; (4) `http_stream.py:180-201` — `close()` waits for the queue to drain (`_wait_for_id_and_queue`) and the final `add_stream_final()` send also goes through `retry()`. **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.** | diff --git a/src/chat_sdk/adapters/teams/adapter.py b/src/chat_sdk/adapters/teams/adapter.py index 8daafb8..e501d3b 100644 --- a/src/chat_sdk/adapters/teams/adapter.py +++ b/src/chat_sdk/adapters/teams/adapter.py @@ -10,6 +10,7 @@ import asyncio import base64 +import contextlib import dataclasses import inspect import json @@ -65,6 +66,7 @@ ValidationError, ) from chat_sdk.types import ( + UNSET, ActionEvent, AdapterPostableMessage, Attachment, @@ -72,6 +74,7 @@ ChannelInfo, ChatInstance, EmojiValue, + EphemeralMessage, FetchOptions, FetchResult, FileUpload, @@ -84,6 +87,7 @@ Message, MessageMetadata, PostableMarkdown, + PostEphemeralOptions, RawMessage, ReactionEvent, StreamOptions, @@ -91,6 +95,7 @@ ThreadSummary, TypingOptions, UninstalledEvent, + Unset, UserInfo, WebhookOptions, _parse_iso, @@ -112,12 +117,67 @@ # Bound on the sender lookup awaited before dispatch (Python-only; see # docs/UPSTREAM_SYNC.md). On expiry the message is dispatched without email. INCOMING_USER_TIMEOUT_S = 5.0 +# Teams deletes a targeted message 24 hours after it is sent, so a record of +# one is worthless past that point (upstream TARGETED_ACTIVITY_TTL_MS). +TARGETED_ACTIVITY_TTL_MS = 24 * 60 * 60 * 1000 # 24 hours +# Bound on waiting for the first native-stream chunk id. Matches the SDK +# ``HttpStream`` close wait. +# Divergence from upstream — see docs/UPSTREAM_SYNC.md (first-chunk wait). +STREAM_FIRST_CHUNK_ID_TIMEOUT_S = 30.0 +# After a timed-out stream is settled with a delivered chunk, the bound on +# waiting for the SDK's background ``chunk`` handler to record that chunk's id. +STREAM_SETTLED_CHUNK_ID_TIMEOUT_S = 1.0 +# Common Chat SDK names that differ from Teams reaction IDs (upstream +# TEAMS_REACTION_ALIASES). Other names pass through as native Teams IDs. +_TEAMS_REACTION_ALIASES: dict[str, str] = { + "check": "2705_whiteheavycheckmark", + "eyes": "1f440_eyes", + "pin": "1f4cc_pushpin", + "rocket": "launch", + "thinking": "think", + "thumbs_up": "like", + "x": "274c_crossmark", +} +# The SDK interpolates the reaction type into the URL path unescaped, so only +# path-safe IDs (``like``, ``1f440_eyes``, hyphenated names) are sent. +# Divergence from upstream — see docs/UPSTREAM_SYNC.md (reaction-ID check). +_TEAMS_REACTION_TYPE_PATTERN = re.compile(r"[A-Za-z0-9_-]+") # Strong references for fire-and-forget tasks (user-info cache writes) so the # GC does not collect them mid-flight. _background_tasks: set[asyncio.Task[Any]] = set() +def _stream_send_in_flight(stream: Any) -> bool: + """Whether the SDK ``HttpStream`` has a flush running or scheduled. + + The SDK exposes no public flag for this, so it reads ``HttpStream`` + internals present on every supported ``microsoft-teams-apps`` version: + ``_lock`` (held for the whole flush, including retries), ``_pending`` (a + flush task not yet started) and ``_timeout`` (the next flush's timer). + Queued content alone does not count: after a flush fails, the SDK leaves + later chunks queued with nothing scheduled to send them. A streamer + without these attributes counts as idle. + """ + lock = getattr(stream, "_lock", None) + if isinstance(lock, asyncio.Lock) and lock.locked(): + return True + return getattr(stream, "_pending", None) is not None or getattr(stream, "_timeout", None) is not None + + +def _resolve_teams_reaction_type(emoji: EmojiValue | str) -> str: + """Map an emoji to a Teams reaction ID (upstream ``resolveTeamsReactionType``). + + Raises :class:`ValidationError` for an ID that is not path-safe, before + any request is made. + """ + name = emoji if isinstance(emoji, str) else emoji.name + reaction_type = _TEAMS_REACTION_ALIASES.get(name, name) + if not _TEAMS_REACTION_TYPE_PATTERN.fullmatch(reaction_type): + raise ValidationError("teams", f"Invalid Teams reaction type: {reaction_type!r}") + return reaction_type + + def _pin_task(task: asyncio.Task[Any]) -> None: _background_tasks.add(task) task.add_done_callback(_background_tasks.discard) @@ -2382,6 +2442,166 @@ async def post_message( # Should not reach here due to _handle_teams_error always raising raise # pragma: no cover + async def post_ephemeral( + self, + thread_id: str, + user_id: str, + message: AdapterPostableMessage, + *, + options: PostEphemeralOptions | None = None, + ) -> EphemeralMessage: + """Send a message only ``user_id`` can see, as a Teams targeted message. + + Port of upstream ``postEphemeral`` (vercel/chat#737, chat@4.35.0). In a + 1:1 chat Teams has no targeted messages, so the message is posted + normally and ``used_fallback`` is ``True``; the check runs before the + activity is built because the SDK rejects a targeted send to a + personal conversation. Elsewhere the activity is addressed to + ``user_id`` with ``with_recipient(..., True)`` (the SDK routes it to the + targeted endpoint) and the sent id is recorded so a later edit or + delete uses the targeted endpoint too. The app must be installed in the + conversation. ``options`` is accepted and ignored, as upstream. + """ + if self.is_dm(thread_id): + # Upstream parity (chat@4.41.1 adapter-teams/src/index.ts + # postEphemeral: `if (this.isDM(threadId)) return postMessage(...)`): + # a 1:1 chat's other member is the only possible viewer, and the + # thread ID does not record who that is, so ``user_id`` is not + # checked against it; callers pass the DM's own user. + sent = await self.post_message(thread_id, message) + return EphemeralMessage( + id=sent.id, + thread_id=sent.thread_id, + used_fallback=True, + raw=sent.raw, + ) + + decoded = self.decode_thread_id(thread_id) + + files = extract_files(message) + file_attachments = await self._files_to_attachments(files) if files else [] + + card = extract_card(message) + activity_payload: dict[str, Any] + if card: + activity_payload = { + "type": "message", + "attachments": [ + { + "contentType": "application/vnd.microsoft.card.adaptive", + "content": card_to_adaptive_card(card), + }, + *file_attachments, + ], + } + self._logger.debug( + "Teams API: send (targeted adaptive card)", + { + "conversationId": decoded.conversation_id, + "userId": user_id, + "fileCount": len(file_attachments), + }, + ) + else: + text = convert_emoji_placeholders( + self._format_converter.render_postable(message), + "teams", + ) + activity_payload = { + "type": "message", + "text": text, + "textFormat": "markdown", + } + if file_attachments: + activity_payload["attachments"] = file_attachments + self._logger.debug( + "Teams API: send (targeted message)", + { + "conversationId": decoded.conversation_id, + "userId": user_id, + "textLength": len(text), + "fileCount": len(file_attachments), + }, + ) + + try: + _validate_service_url(decoded.service_url) + activity = self._message_activity_input(activity_payload).with_recipient( + self._create_targeted_recipient(user_id), True + ) + sent = await self._send_to(decoded, activity) + except Exception as error: + self._logger.error( + "Teams API: targeted send failed", + { + "conversationId": decoded.conversation_id, + "userId": user_id, + "error": str(error), + }, + ) + _handle_teams_error(error, "postEphemeral") + raise # unreachable: _handle_teams_error always raises + + sent_id = getattr(sent, "id", "") or "" + await self._remember_targeted_activity(decoded.conversation_id, sent_id) + self._logger.debug("Teams API: targeted send response", {"messageId": sent_id}) + return EphemeralMessage( + id=sent_id, + thread_id=thread_id, + used_fallback=False, + raw=activity_payload, + ) + + @staticmethod + def _create_targeted_recipient(user_id: str) -> Any: + """The recipient account of a targeted message (upstream ``createTargetedRecipient``). + + ``role`` is a field on ``microsoft-teams-api`` 2.1's ``Account``; 2.0.x + keeps it as an extra field, so it is on the wire either way. + """ + from microsoft_teams.api import Account + + return Account(id=user_id, name=user_id, role="user") + + # Teams accepts an update or a delete of a targeted activity only through + # the ``?isTargetedActivity=true`` endpoint; the plain one answers 400. + # Nothing on the wire says afterwards which activities were targeted, and + # ``edit_message`` / ``delete_message`` get only an id, so the ids sent + # targeted are recorded in the Chat state adapter (shared across instances + # and restarts) and read back on mutation. Every access is best effort: a + # lost record costs the 400, never the message. Upstream vercel/chat#951. + + @staticmethod + def _targeted_activity_key(conversation_id: str, message_id: str) -> str: + return f"teams:targetedActivity:{conversation_id}:{message_id}" + + async def _remember_targeted_activity(self, conversation_id: str, message_id: str) -> None: + if not (self._chat and message_id): + return + # A state adapter that cannot write must not fail the send. + with contextlib.suppress(Exception): + await self._chat.get_state().set( + self._targeted_activity_key(conversation_id, message_id), + "1", + TARGETED_ACTIVITY_TTL_MS, + ) + + async def _is_targeted_activity(self, conversation_id: str, message_id: str) -> bool: + if not (self._chat and message_id): + return False + try: + recorded = await self._chat.get_state().get(self._targeted_activity_key(conversation_id, message_id)) + except Exception: + return False + return recorded == "1" + + async def _forget_targeted_activity(self, conversation_id: str, message_id: str) -> None: + if not (self._chat and message_id): + return + # The TTL clears it anyway. + with contextlib.suppress(Exception): + await self._chat.get_state().delete(self._targeted_activity_key(conversation_id, message_id)) + async def edit_message( self, thread_id: str, @@ -2422,21 +2642,23 @@ async def edit_message( "textFormat": "markdown", } + targeted = await self._is_targeted_activity(decoded.conversation_id, message_id) + self._logger.debug( "Teams API: updateActivity", { "conversationId": decoded.conversation_id, "messageId": message_id, + "targeted": targeted, }, ) try: _validate_service_url(decoded.service_url) api = self._api_for(decoded.service_url) - await api.conversations.activities(decoded.conversation_id).update( - message_id, - self._message_activity_input(activity_payload), - ) + activities = api.conversations.activities(decoded.conversation_id) + update = activities.update_targeted if targeted else activities.update + await update(message_id, self._message_activity_input(activity_payload)) except Exception as error: self._logger.error( "Teams API: updateActivity failed", @@ -2455,18 +2677,23 @@ async def delete_message(self, thread_id: str, message_id: str) -> None: """Delete a Teams message.""" decoded = self.decode_thread_id(thread_id) + targeted = await self._is_targeted_activity(decoded.conversation_id, message_id) + self._logger.debug( "Teams API: deleteActivity", { "conversationId": decoded.conversation_id, "messageId": message_id, + "targeted": targeted, }, ) try: _validate_service_url(decoded.service_url) api = self._api_for(decoded.service_url) - await api.conversations.activities(decoded.conversation_id).delete(message_id) + activities = api.conversations.activities(decoded.conversation_id) + delete = activities.delete_targeted if targeted else activities.delete + await delete(message_id) except Exception as error: self._logger.error( "Teams API: deleteActivity failed", @@ -2479,14 +2706,22 @@ async def delete_message(self, thread_id: str, message_id: str) -> None: _handle_teams_error(error, "deleteMessage") raise # unreachable: _handle_teams_error always raises + await self._forget_targeted_activity(decoded.conversation_id, message_id) + async def add_reaction( self, thread_id: str, message_id: str, emoji: EmojiValue | str, ) -> None: - """Add a reaction (not supported by Teams Bot Framework API).""" - self._logger.warn("addReaction is not supported by the Teams Bot Framework API") + """Add the bot's reaction to a message (upstream vercel/chat#734). + + ``check``, ``eyes``, ``pin``, ``rocket``, ``thinking``, ``thumbs_up`` + and ``x`` map to their Teams IDs; any other name is sent as a native + Teams reaction ID. A name that is not path-safe raises + :class:`ValidationError` before any request. + """ + await self._send_reaction(thread_id, message_id, emoji, remove=False) async def remove_reaction( self, @@ -2494,8 +2729,50 @@ async def remove_reaction( message_id: str, emoji: EmojiValue | str, ) -> None: - """Remove a reaction (not supported by Teams Bot Framework API).""" - self._logger.warn("removeReaction is not supported by the Teams Bot Framework API") + """Remove the bot's reaction from a message (upstream vercel/chat#734).""" + await self._send_reaction(thread_id, message_id, emoji, remove=True) + + async def _send_reaction( + self, + thread_id: str, + message_id: str, + emoji: EmojiValue | str, + *, + remove: bool, + ) -> None: + """Add or delete a reaction through the conversations reactions API. + + ``microsoft-teams-api`` 2.0.16+ has ``conversations.add_reaction`` / + ``delete_reaction`` (upstream's ``conversations.addReaction``); 2.0.13 + has only ``ApiClient.reactions.add`` / ``delete``, which 2.0.16+ + deprecates. Both issue the same request. + """ + decoded = self.decode_thread_id(thread_id) + reaction_type = _resolve_teams_reaction_type(emoji) + label = "deleteReaction" if remove else "addReaction" + log_context = { + "conversationId": decoded.conversation_id, + "messageId": message_id, + "reactionType": reaction_type, + } + + self._logger.debug(f"Teams API: {label}", log_context) + + try: + _validate_service_url(decoded.service_url) + api = self._api_for(decoded.service_url) + conversations = api.conversations + if hasattr(conversations, "add_reaction"): + call = conversations.delete_reaction if remove else conversations.add_reaction + else: + call = api.reactions.delete if remove else api.reactions.add + await call(decoded.conversation_id, message_id, reaction_type) + except Exception as error: + self._logger.error(f"Teams API: {label} failed", {**log_context, "error": str(error)}) + _handle_teams_error(error, "removeReaction" if remove else "addReaction") + raise # unreachable: _handle_teams_error always raises + + self._logger.debug(f"Teams API: {label} response", {"ok": True}) async def start_typing( self, thread_id: str, status: str | None = None, *, options: TypingOptions | None = None @@ -2529,23 +2806,31 @@ async def stream( thread_id: str, text_stream: Any, options: StreamOptions | None = None, - ) -> RawMessage: + ) -> RawMessage | None: """Stream responses to a Teams conversation. DMs stream natively via the Teams SDK ``IStreamer.emit()`` when an active streamer exists (captured by :meth:`_handle_message_activity` - from the inbound activity). Group chats / channels / proactive messages - accumulate the stream and post a single message — Teams supports native + from the inbound activity); a placeholder set with + ``fallback_streaming_placeholder_text`` is shown there as the native + status line. Without a streamer (group chats, channels, proactive + messages), a placeholder string returns ``None`` before the stream is + read, so core posts the placeholder and edits it; otherwise the stream + is accumulated and posted as a single message. Teams supports native streaming only in 1:1 chats. - Mirrors upstream ``stream`` in - ``packages/adapter-teams/src/index.ts`` (``@chat-adapter/teams@4.30.0``): - delegate to ``streamViaEmit`` when ``activeStream && !activeStream.canceled``, - else accumulate and ``postMessage``. + Mirrors upstream ``stream`` in ``packages/adapter-teams/src/index.ts`` + (vercel/chat#709, chat@4.35.0). Only an explicit ``str`` (``""`` + included) changes behaviour: ``UNSET`` (not configured) and ``None`` + (no placeholder) keep the native and buffered paths unchanged. """ + placeholder = options.fallback_streaming_placeholder_text if options is not None else UNSET active_stream = self._active_streams.get(thread_id) if active_stream is not None and not active_stream.canceled: - return await self._stream_via_emit(thread_id, text_stream, active_stream) + return await self._stream_via_emit(thread_id, text_stream, active_stream, placeholder) + + if active_stream is None and isinstance(placeholder, str): + return None # No native streamer (group chats, proactive messages, or DMs whose # streamer was already canceled). Accumulate and post once — matching @@ -2579,14 +2864,20 @@ async def _stream_via_emit( thread_id: str, text_stream: Any, stream: StreamerProtocol, + placeholder_text: str | None | Unset = UNSET, ) -> RawMessage: """Native streaming via the Teams SDK ``IStreamer.emit()``. + A ``placeholder_text`` string is sent first with ``stream.update()`` + (an informative status, replaced by the first text chunk); ``None`` + and ``UNSET`` send nothing. + Each non-empty chunk is handed to ``stream.emit(text)``; the SDK ``HttpStream`` owns the Bot Framework streaming wire format (``streamType``/``streamSequence``/``streamId``), the per-flush throttle (~500ms between flushes), and 429 retry/backoff. We never - touch its internals and we never call ``stream.close()`` — the SDK + touch its internals and we call ``stream.close()`` only after the + first-chunk wait below times out — normally the SDK sends the ``streamType: 'final'`` message after the handler returns (see :meth:`_handle_message_activity`'s ``finally`` block, which plays the lifecycle-owner role the SDK App otherwise would). @@ -2601,7 +2892,14 @@ async def _stream_via_emit( We capture the first chunk's server-assigned id via ``on_chunk`` and await it ONLY when text was emitted and the stream was not canceled — awaiting unconditionally would hang forever if no chunk was ever - delivered (empty stream, or canceled before the first flush). + delivered (empty stream, or canceled before the first flush). The wait + is also bounded by ``STREAM_FIRST_CHUNK_ID_TIMEOUT_S`` (Python-only): + ``microsoft-teams-apps`` 2.0.16+ leaves ``canceled`` unset when the + first flush fails terminally (e.g. a 403 that is not a cancel), so no + chunk ever arrives. The stream is then settled with ``close()``; if + nothing was delivered and the user did not cancel, the accumulated + text is sent with one buffered ``post_message`` so the reply is not + lost. Mirrors upstream ``streamViaEmit`` in ``packages/adapter-teams/src/index.ts`` (``@chat-adapter/teams@4.30.0``). @@ -2626,6 +2924,11 @@ async def _on_chunk(activity: Any) -> None: stream.on_chunk(_on_chunk) try: + if isinstance(placeholder_text, str): + # Synchronous on the SDK streamer: it queues an informative + # typing activity for the next flush. + stream.update(placeholder_text) + async for chunk in text_stream: if stream.canceled: self._logger.debug("Teams stream canceled by user", {"threadId": thread_id}) @@ -2662,12 +2965,73 @@ async def _on_chunk(activity: Any) -> None: # canceled before any chunk was delivered (which would hang forever). if accumulated and not stream.canceled: try: - message_id = await id_captured + # Shielded so a timeout leaves the capture alive: a first chunk + # that lands while ``close()`` settles the stream still records + # its id (Teams returns the id only on the first response). + message_id = await asyncio.wait_for(asyncio.shield(id_captured), STREAM_FIRST_CHUNK_ID_TIMEOUT_S) except StreamCancelledError: self._logger.debug( "Teams stream canceled before first chunk delivered", {"threadId": thread_id}, ) + except asyncio.TimeoutError: + # Python-only (see the first-chunk row in docs/UPSTREAM_SYNC.md). + # The first flush may have failed terminally (e.g. a 403 + # StreamNotAllowedError), or may still be retrying. ``close()`` + # settles it: it waits for an in-flight flush and finalizes a + # stream that did get through, and returns ``None`` when nothing + # was delivered. The handler's own ``close()`` is then a no-op. + if not stream.canceled: + try: + settled = await stream.close() + except StreamCancelledError: + # The user canceled while the stream was being + # finalized. Only the SDK subclass is caught; a plain + # task cancel still propagates. + self._logger.debug( + "Teams stream canceled while settling", + {"threadId": thread_id}, + ) + return RawMessage(id="", thread_id=thread_id, raw={"text": accumulated}) + if settled is not None: + # The stream is finalized, so a later post in this + # handler must not reuse it (2.0.13.4 does not reopen a + # closed stream on emit); it takes the no-streamer path. + if self._active_streams.get(thread_id) is stream: + self._active_streams.pop(thread_id, None) + # A settled stream means the first chunk was delivered, + # and its id is the message id (the final response may + # carry none). The SDK runs async ``chunk`` handlers as a + # background task, so give the capture a moment to land. + try: + message_id = await asyncio.wait_for( + asyncio.shield(id_captured), STREAM_SETTLED_CHUNK_ID_TIMEOUT_S + ) + except asyncio.TimeoutError: + message_id = getattr(settled, "id", "") or "" + elif _stream_send_in_flight(stream): + # close() gave up waiting while a flush is still in + # flight (the SDK client has no request timeout). It + # may still deliver, and the handler's close() will + # finalize it, so posting now could send the reply + # twice. + self._logger.warn( + "Teams stream first chunk still in flight; not posting a fallback", + {"threadId": thread_id}, + ) + elif not stream.canceled: + # Nothing reached the user and nothing is in flight: + # deliver the text with one buffered post instead of + # dropping the reply. The failed streamer is retired + # first: the SDK keeps this text buffered, so reusing + # it would resend it ahead of a later reply. + if self._active_streams.get(thread_id) is stream: + self._active_streams.pop(thread_id, None) + self._logger.warn( + "Teams stream delivered no chunk; posting the reply as one message", + {"threadId": thread_id}, + ) + return await self.post_message(thread_id, PostableMarkdown(markdown=accumulated)) return RawMessage(id=message_id, thread_id=thread_id, raw={"text": accumulated}) diff --git a/tests/test_critical_fixes.py b/tests/test_critical_fixes.py index 06a8b9c..57b07cb 100644 --- a/tests/test_critical_fixes.py +++ b/tests/test_critical_fixes.py @@ -4,7 +4,6 @@ 1. Telegram process_action uses snake_case keys 2. Telegram process_reaction uses snake_case keys 3. Google Chat stream() accepts AsyncIterable -4. Teams add_reaction / remove_reaction don't raise (just warn) 5. WhatsApp post_message calls Graph API correctly 6. WhatsApp add_reaction sends correct emoji payload 7. WhatsApp stream accumulates and posts @@ -29,8 +28,6 @@ GoogleChatAdapterConfig, ServiceAccountCredentials, ) -from chat_sdk.adapters.teams.adapter import TeamsAdapter -from chat_sdk.adapters.teams.types import TeamsAdapterConfig from chat_sdk.adapters.telegram.adapter import TelegramAdapter from chat_sdk.adapters.telegram.types import TelegramAdapterConfig from chat_sdk.adapters.whatsapp.adapter import WhatsAppAdapter @@ -51,15 +48,6 @@ def _make_telegram_adapter(**overrides) -> TelegramAdapter: return TelegramAdapter(config) -def _make_teams_adapter(**overrides) -> TeamsAdapter: - config = TeamsAdapterConfig( - app_id=overrides.pop("app_id", "test-app-id"), - app_password=overrides.pop("app_password", "test-password"), - **overrides, - ) - return TeamsAdapter(config) - - def _make_whatsapp_adapter(**overrides) -> WhatsAppAdapter: defaults = { "access_token": "test-token", @@ -355,47 +343,6 @@ async def mixed_stream(): assert posted[0][1].markdown == "Start middle end" -# --------------------------------------------------------------------------- -# 4. Teams add_reaction / remove_reaction don't raise -# --------------------------------------------------------------------------- - - -class TestTeamsReactionsGraceful: - """Verify Teams reactions log a warning instead of raising NotImplementedError.""" - - @pytest.mark.asyncio - async def test_add_reaction_does_not_raise(self): - mock_logger = MagicMock( - debug=MagicMock(), - info=MagicMock(), - warn=MagicMock(), - error=MagicMock(), - ) - adapter = _make_teams_adapter(logger=mock_logger) - - # Should not raise - await adapter.add_reaction("teams:conv123", "msg456", "thumbs_up") - - mock_logger.warn.assert_called_once() - assert "not supported" in mock_logger.warn.call_args[0][0].lower() - - @pytest.mark.asyncio - async def test_remove_reaction_does_not_raise(self): - mock_logger = MagicMock( - debug=MagicMock(), - info=MagicMock(), - warn=MagicMock(), - error=MagicMock(), - ) - adapter = _make_teams_adapter(logger=mock_logger) - - # Should not raise - await adapter.remove_reaction("teams:conv123", "msg456", "thumbs_up") - - mock_logger.warn.assert_called_once() - assert "not supported" in mock_logger.warn.call_args[0][0].lower() - - # --------------------------------------------------------------------------- # 5. WhatsApp post_message calls Graph API correctly # --------------------------------------------------------------------------- diff --git a/tests/test_teams_adapter.py b/tests/test_teams_adapter.py index 3c3da42..edbbc30 100644 --- a/tests/test_teams_adapter.py +++ b/tests/test_teams_adapter.py @@ -1567,6 +1567,8 @@ def _card() -> Any: "edit_message", "delete_message", "start_typing", + "add_reaction", + "remove_reaction", ], ) async def test_every_outbound_op_rejects_a_disallowed_thread_url_before_sending( @@ -1600,6 +1602,8 @@ async def fake_http(_self, url, **_kwargs): "edit_message": lambda: adapter.edit_message(thread_id, "m-1", "hi"), "delete_message": lambda: adapter.delete_message(thread_id, "m-1"), "start_typing": lambda: adapter.start_typing(thread_id), + "add_reaction": lambda: adapter.add_reaction(thread_id, "m-1", "like"), + "remove_reaction": lambda: adapter.remove_reaction(thread_id, "m-1", "like"), } if operation == "start_typing": @@ -2214,24 +2218,552 @@ async def test_sends_typing_activity(self): # --------------------------------------------------------------------------- -# addReaction / removeReaction (not supported) +# addReaction / removeReaction (upstream describe "reactions", vercel/chat#734) # --------------------------------------------------------------------------- +_GROUP_THREAD = TeamsThreadId( + conversation_id="19:abc@thread.tacv2", + service_url="https://smba.trafficmanager.net/teams/", +) + + +class _SdkStatusError(Exception): + """An SDK-style error carrying an HTTP status on ``.status_code``.""" + + def __init__(self, status_code: int, message: str) -> None: + super().__init__(message) + self.status_code = status_code + + +def _reaction_test_adapter() -> tuple[TeamsAdapter, AsyncMock, AsyncMock, str]: + """Upstream ``createReactionTestAdapter``: ``App.api`` exposes only + ``conversations.add_reaction`` / ``delete_reaction`` (the 2.0.16+ SDK shape).""" + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + add_reaction = AsyncMock(return_value=None) + delete_reaction = AsyncMock(return_value=None) + adapter._app.api = SimpleNamespace( # type: ignore[method-assign] + service_url="https://smba.trafficmanager.net/teams", + conversations=SimpleNamespace(add_reaction=add_reaction, delete_reaction=delete_reaction), + ) + return adapter, add_reaction, delete_reaction, adapter.encode_thread_id(_GROUP_THREAD) + class TestReactions: @pytest.mark.asyncio - async def test_add_reaction_warns_instead_of_raising(self): - logger = _make_logger() - adapter = _make_adapter(logger=logger) - await adapter.add_reaction("tid", "mid", "emoji") - assert logger.warn.call_count == 1 + async def test_should_add_a_raw_teams_reaction_id(self): + adapter, add_reaction, _delete, thread_id = _reaction_test_adapter() + + await adapter.add_reaction(thread_id, "message-1", "think") + + add_reaction.assert_awaited_once_with("19:abc@thread.tacv2", "message-1", "think") @pytest.mark.asyncio - async def test_remove_reaction_warns_instead_of_raising(self): - logger = _make_logger() - adapter = _make_adapter(logger=logger) - await adapter.remove_reaction("tid", "mid", "emoji") - assert logger.warn.call_count == 1 + @pytest.mark.parametrize( + ("name", "teams_id"), + [ + ("check", "2705_whiteheavycheckmark"), + ("eyes", "1f440_eyes"), + ("pin", "1f4cc_pushpin"), + ("rocket", "launch"), + ("thinking", "think"), + ("thumbs_up", "like"), + ("x", "274c_crossmark"), + ], + ) + async def test_should_map_name_to_the_teams_reaction_id(self, name: str, teams_id: str): + from chat_sdk.emoji import get_emoji + + adapter, add_reaction, _delete, thread_id = _reaction_test_adapter() + + await adapter.add_reaction(thread_id, "message-1", get_emoji(name)) + + add_reaction.assert_awaited_once_with("19:abc@thread.tacv2", "message-1", teams_id) + + @pytest.mark.asyncio + async def test_should_remove_a_reaction_with_the_teams_conversation_api(self): + from chat_sdk.emoji import get_emoji + + adapter, add_reaction, delete_reaction, thread_id = _reaction_test_adapter() + + await adapter.remove_reaction(thread_id, "message-1", get_emoji("check")) + + delete_reaction.assert_awaited_once_with("19:abc@thread.tacv2", "message-1", "2705_whiteheavycheckmark") + add_reaction.assert_not_called() + + @pytest.mark.asyncio + async def test_should_translate_teams_api_failures(self): + from chat_sdk.shared.errors import AuthenticationError + + adapter, add_reaction, _delete, thread_id = _reaction_test_adapter() + add_reaction.side_effect = _SdkStatusError(401, "Unauthorized") + + with pytest.raises(AuthenticationError, match="addReaction"): + await adapter.add_reaction(thread_id, "message-1", "like") + + @pytest.mark.asyncio + async def test_remove_failure_is_reported_as_remove_reaction(self): + from chat_sdk.shared.errors import AuthenticationError + + adapter, _add, delete_reaction, thread_id = _reaction_test_adapter() + delete_reaction.side_effect = _SdkStatusError(401, "Unauthorized") + + with pytest.raises(AuthenticationError, match="removeReaction"): + await adapter.remove_reaction(thread_id, "message-1", "like") + + @pytest.mark.asyncio + @pytest.mark.parametrize("name", ["like/../../x", "like?x=1", "like#x", "", "like\n", "%2e%2e", "lik e"]) + async def test_rejects_a_reaction_name_that_is_not_path_safe_before_any_call(self, name: str): + """Python-only: the SDK puts the reaction type into the URL path unescaped.""" + adapter, add_reaction, delete_reaction, thread_id = _reaction_test_adapter() + + with pytest.raises(ValidationError, match="Invalid Teams reaction type"): + await adapter.add_reaction(thread_id, "message-1", name) + with pytest.raises(ValidationError, match="Invalid Teams reaction type"): + await adapter.remove_reaction(thread_id, "message-1", name) + + add_reaction.assert_not_called() + delete_reaction.assert_not_called() + + @pytest.mark.asyncio + async def test_accepts_a_hyphenated_reaction_id(self): + adapter, add_reaction, _delete, thread_id = _reaction_test_adapter() + + await adapter.add_reaction(thread_id, "message-1", "yes-tone1") + + add_reaction.assert_awaited_once_with("19:abc@thread.tacv2", "message-1", "yes-tone1") + + @pytest.mark.asyncio + async def test_uses_the_reactions_client_on_sdks_without_conversation_reactions(self): + """``microsoft-teams-api`` 2.0.13 has only ``ApiClient.reactions``.""" + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + add = AsyncMock(return_value=None) + delete = AsyncMock(return_value=None) + adapter._app.api = SimpleNamespace( # type: ignore[method-assign] + service_url="https://smba.trafficmanager.net/teams", + conversations=SimpleNamespace(), + reactions=SimpleNamespace(add=add, delete=delete), + ) + thread_id = adapter.encode_thread_id(_GROUP_THREAD) + + await adapter.add_reaction(thread_id, "message-1", "thumbs_up") + await adapter.remove_reaction(thread_id, "message-1", "thumbs_up") + + add.assert_awaited_once_with("19:abc@thread.tacv2", "message-1", "like") + delete.assert_awaited_once_with("19:abc@thread.tacv2", "message-1", "like") + + @pytest.mark.asyncio + async def test_reaction_hits_the_wire_on_the_threads_service_url(self): + """The real SDK client issues ``PUT``/``DELETE`` on the thread's service URL.""" + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200) + + from microsoft_teams.api import ApiClient + + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + adapter._app.api = ApiClient("https://smba.trafficmanager.net/teams", _bot_http_client(handler)) # type: ignore[method-assign] + thread_id = adapter.encode_thread_id( + TeamsThreadId( + conversation_id="19:abc@thread.tacv2", + service_url="https://smba.trafficmanager.net/emea/", + ) + ) + + await adapter.add_reaction(thread_id, "1700000000000", "eyes") + await adapter.remove_reaction(thread_id, "1700000000000", "eyes") + + assert [(r.method, str(r.url)) for r in requests] == [ + ( + "PUT", + "https://smba.trafficmanager.net/emea/v3/conversations/19:abc@thread.tacv2" + "/activities/1700000000000/reactions/1f440_eyes", + ), + ( + "DELETE", + "https://smba.trafficmanager.net/emea/v3/conversations/19:abc@thread.tacv2" + "/activities/1700000000000/reactions/1f440_eyes", + ), + ] + assert requests[0].headers["authorization"] == "Bearer bot-token" + + +# --------------------------------------------------------------------------- +# postEphemeral (upstream describe "postEphemeral", vercel/chat#737) +# --------------------------------------------------------------------------- + + +class TestPostEphemeral: + @pytest.mark.asyncio + async def test_should_send_a_targeted_text_message_to_the_requested_user(self): + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter, "targeted-msg-123") + thread_id = adapter.encode_thread_id(_GROUP_THREAD) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", {"markdown": "Only you can see this"}) + + assert result.id == "targeted-msg-123" + assert result.thread_id == thread_id + assert result.used_fallback is False + send.assert_awaited_once() + activity, ref = send.call_args.args + assert ref.conversation.id == "19:abc@thread.tacv2" + assert activity.recipient.id == "29:target-user" + assert activity.recipient.name == "29:target-user" + assert activity.recipient.is_targeted is True + assert activity.recipient.role == "user" + assert activity.text == "Only you can see this" + assert activity.text_format == "markdown" + wire = activity.model_dump(by_alias=True, exclude_none=True) + assert wire["recipient"] == { + "id": "29:target-user", + "name": "29:target-user", + "role": "user", + "isTargeted": True, + } + + @pytest.mark.asyncio + async def test_should_send_targeted_adaptive_cards(self): + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter, "targeted-card-123") + thread_id = adapter.encode_thread_id(_GROUP_THREAD) + + result = await adapter.post_ephemeral( + thread_id, + "29:target-user", + {"card": {"type": "card", "title": "Private card", "children": []}}, + ) + + assert result.id == "targeted-card-123" + assert result.used_fallback is False + activity, ref = send.call_args.args + assert ref.conversation.id == "19:abc@thread.tacv2" + assert [a.content_type for a in activity.attachments] == ["application/vnd.microsoft.card.adaptive"] + assert activity.recipient.id == "29:target-user" + assert activity.recipient.is_targeted is True + + @pytest.mark.asyncio + @pytest.mark.parametrize("kind", ["text", "card"]) + async def test_targeted_messages_carry_file_attachments(self, kind: str): + """Python-only coverage (upstream's postEphemeral tests send no files): + files ride on the targeted activity as data-URI attachments, after the + adaptive card when there is one, and the activity stays targeted.""" + from chat_sdk.cards import Card + from chat_sdk.types import FileUpload, PostableCard, PostableMarkdown + + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter, "targeted-file-1") + thread_id = adapter.encode_thread_id(_GROUP_THREAD) + files = [FileUpload(data=b"a,b\n1,2\n", filename="report.csv", mime_type="text/csv")] + message = ( + PostableMarkdown(markdown="your report", files=files) + if kind == "text" + else PostableCard(card=Card(title="Results"), files=files) + ) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", message) + + assert result.used_fallback is False + activity, _ref = send.call_args.args + wire = activity.model_dump(by_alias=True, exclude_none=True) + content_types = [a["contentType"] for a in wire["attachments"]] + if kind == "text": + assert wire["text"] == "your report" + assert content_types == ["text/csv"] + else: + assert content_types == ["application/vnd.microsoft.card.adaptive", "text/csv"] + file_att = wire["attachments"][-1] + assert file_att["name"] == "report.csv" + assert file_att["contentUrl"] == "data:text/csv;base64,YSxiCjEsMgo=" + assert wire["recipient"]["isTargeted"] is True + assert wire["recipient"]["id"] == "29:target-user" + + @pytest.mark.asyncio + async def test_should_handle_targeted_send_failure_by_calling_handle_teams_error(self): + from chat_sdk.shared.errors import AuthenticationError + + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter) + send.side_effect = _SdkStatusError(401, "Unauthorized") + thread_id = adapter.encode_thread_id(_GROUP_THREAD) + + with pytest.raises(AuthenticationError, match="postEphemeral"): + await adapter.post_ephemeral(thread_id, "29:target-user", "Private") + + @pytest.mark.asyncio + async def test_falls_back_to_a_normal_post_in_a_1_1_chat_instead_of_a_targeted_send(self): + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter, "personal-msg-123") + thread_id = adapter.encode_thread_id( + TeamsThreadId( + conversation_id="a:1personal-chat-id", + service_url="https://smba.trafficmanager.net/teams/", + ) + ) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", {"markdown": "Only you can see this"}) + + assert result.id == "personal-msg-123" + assert result.used_fallback is True + activity, _ref = send.call_args.args + assert activity.recipient is None + assert activity.text == "Only you can see this" + + @pytest.mark.asyncio + async def test_an_explicit_personal_type_falls_back_even_with_a_19_id(self): + """``is_dm`` honours the thread ID's explicit type, so the SDK's + personal-chat guard on targeted sends is never reached.""" + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter, "personal-msg-2") + thread_id = adapter.encode_thread_id( + TeamsThreadId( + conversation_id="19:personal@unq.gbl.spaces", + service_url="https://smba.trafficmanager.net/teams/", + conversation_type="personal", + ) + ) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + + assert result.used_fallback is True + activity, ref = send.call_args.args + assert activity.recipient is None + assert ref.conversation.conversation_type == "personal" + + @pytest.mark.asyncio + async def test_an_a_prefixed_group_chat_gets_a_targeted_send(self): + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter, "targeted-group-1") + thread_id = adapter.encode_thread_id( + TeamsThreadId( + conversation_id="a:group-chat", + service_url="https://smba.trafficmanager.net/teams/", + conversation_type="groupChat", + ) + ) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + + assert result.used_fallback is False + activity, _ref = send.call_args.args + assert activity.recipient.is_targeted is True + + @pytest.mark.asyncio + async def test_rejects_a_disallowed_service_url_before_sending(self): + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + send = _mock_app_send(adapter) + thread_id = adapter.encode_thread_id( + TeamsThreadId(conversation_id="19:abc@thread.tacv2", service_url="https://evil.example.com/") + ) + + with pytest.raises(NetworkError): + await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + send.assert_not_called() + + @pytest.mark.asyncio + async def test_targeted_send_edit_and_delete_hit_the_targeted_endpoints_on_the_wire(self): + """Real SDK clients over an in-memory transport: the targeted create, the + later edit and the delete all carry ``isTargetedActivity=true``. The send + goes through 2.1's ``send_or_update_activity`` (2.0.x's + ``ActivitySender`` builds its own client, so that half needs 2.1).""" + pytest.importorskip("microsoft_teams.apps.activity_send") + import json as _json + + from microsoft_teams.api import ApiClient + + from chat_sdk.testing import create_mock_chat_instance + + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.method == "DELETE": + return httpx.Response(200) + return httpx.Response(200, json={"id": "targeted-wire-1"}) + + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + adapter._chat = create_mock_chat_instance() # type: ignore[assignment] + adapter._app.api = ApiClient("https://smba.trafficmanager.net/teams", _bot_http_client(handler)) # type: ignore[method-assign] + thread_id = adapter.encode_thread_id(_GROUP_THREAD) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + await adapter.edit_message(thread_id, result.id, "edited") + await adapter.delete_message(thread_id, result.id) + + base = "https://smba.trafficmanager.net/teams/v3/conversations/19:abc@thread.tacv2/activities" + assert [(r.method, str(r.url)) for r in requests] == [ + ("POST", f"{base}?isTargetedActivity=true"), + ("PUT", f"{base}/targeted-wire-1?isTargetedActivity=true"), + ("DELETE", f"{base}/targeted-wire-1?isTargetedActivity=true"), + ] + sent = _json.loads(requests[0].content) + assert sent["recipient"] == { + "id": "29:target-user", + "name": "29:target-user", + "role": "user", + "isTargeted": True, + } + assert sent["text"] == "hi" + assert "recipient" not in _json.loads(requests[1].content) + + +# --------------------------------------------------------------------------- +# Targeted message mutation (upstream describe "targeted message mutation", +# vercel/chat#951) +# --------------------------------------------------------------------------- + + +def _targeted_adapter(state: Any = None) -> tuple[TeamsAdapter, SimpleNamespace, str, Any]: + """Upstream ``createTargetedAdapter``: a Chat with mock state, a sender that + returns ``targeted-msg-1`` and activity ops for both endpoints.""" + from chat_sdk.testing import create_mock_chat_instance, create_mock_state + + adapter = _make_adapter(app_id="test-app-id", logger=_make_logger()) + resolved_state = state if state is not None else create_mock_state() + adapter._chat = create_mock_chat_instance(state=resolved_state) # type: ignore[assignment] + calls = SimpleNamespace( + update=AsyncMock(return_value=_SentActivity("targeted-msg-1")), + update_targeted=AsyncMock(return_value=_SentActivity("targeted-msg-1")), + delete=AsyncMock(return_value=None), + delete_targeted=AsyncMock(return_value=None), + ) + _mock_app_send(adapter, "targeted-msg-1") + adapter._app.api = SimpleNamespace( # type: ignore[method-assign] + service_url="https://smba.trafficmanager.net/teams", + conversations=SimpleNamespace(activities=MagicMock(return_value=calls)), + ) + return adapter, calls, adapter.encode_thread_id(_GROUP_THREAD), resolved_state + + +class TestTargetedMessageMutation: + @pytest.mark.asyncio + async def test_deletes_a_message_it_sent_targeted_through_the_targeted_endpoint(self): + adapter, calls, thread_id, _state = _targeted_adapter() + await adapter.post_ephemeral(thread_id, "29:target-user", {"markdown": "Only you can see this"}) + + await adapter.delete_message(thread_id, "targeted-msg-1") + + calls.delete_targeted.assert_awaited_once_with("targeted-msg-1") + calls.delete.assert_not_called() + + @pytest.mark.asyncio + async def test_edits_a_message_it_sent_targeted_through_the_targeted_endpoint(self): + adapter, calls, thread_id, _state = _targeted_adapter() + await adapter.post_ephemeral(thread_id, "29:target-user", {"markdown": "Only you can see this"}) + + await adapter.edit_message(thread_id, "targeted-msg-1", {"markdown": "Updated"}) + + calls.update_targeted.assert_awaited_once() + message_id, activity = calls.update_targeted.call_args.args + assert message_id == "targeted-msg-1" + assert activity.text == "Updated" + calls.update.assert_not_called() + + @pytest.mark.asyncio + async def test_leaves_a_message_it_did_not_send_targeted_on_the_plain_endpoint(self): + adapter, calls, thread_id, _state = _targeted_adapter() + + await adapter.delete_message(thread_id, "public-msg-1") + await adapter.edit_message(thread_id, "public-msg-1", {"markdown": "Updated"}) + + calls.delete.assert_awaited_once_with("public-msg-1") + assert calls.update.await_count == 1 + calls.delete_targeted.assert_not_called() + calls.update_targeted.assert_not_called() + + @pytest.mark.asyncio + async def test_stops_treating_a_targeted_message_as_targeted_once_it_is_deleted(self): + adapter, calls, thread_id, _state = _targeted_adapter() + await adapter.post_ephemeral(thread_id, "29:target-user", {"markdown": "Only you can see this"}) + + await adapter.delete_message(thread_id, "targeted-msg-1") + await adapter.delete_message(thread_id, "targeted-msg-1") + + assert calls.delete_targeted.await_count == 1 + assert calls.delete.await_count == 1 + + @pytest.mark.asyncio + async def test_falls_back_to_the_plain_endpoint_when_the_state_adapter_cannot_be_read(self): + from chat_sdk.testing import create_mock_state + + state = create_mock_state() + state.get = AsyncMock(side_effect=RuntimeError("state unavailable")) # type: ignore[method-assign] + adapter, calls, thread_id, _state = _targeted_adapter(state) + + await adapter.delete_message(thread_id, "msg-1") + + calls.delete.assert_awaited_once_with("msg-1") + calls.delete_targeted.assert_not_called() + + @pytest.mark.asyncio + async def test_records_the_targeted_id_under_the_cross_sdk_key_with_a_24h_ttl(self): + from chat_sdk.testing import create_mock_state + + state = create_mock_state() + state.set = AsyncMock(wraps=state.set) # type: ignore[method-assign] + adapter, _calls, thread_id, _state = _targeted_adapter(state) + + await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + + state.set.assert_awaited_once_with( + "teams:targetedActivity:19:abc@thread.tacv2:targeted-msg-1", "1", 24 * 60 * 60 * 1000 + ) + + @pytest.mark.asyncio + async def test_a_state_write_failure_does_not_fail_the_targeted_send(self): + from chat_sdk.testing import create_mock_state + + state = create_mock_state() + state.set = AsyncMock(side_effect=RuntimeError("state unavailable")) # type: ignore[method-assign] + adapter, _calls, thread_id, _state = _targeted_adapter(state) + + result = await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + + assert result.id == "targeted-msg-1" + assert result.used_fallback is False + + @pytest.mark.asyncio + async def test_a_state_delete_failure_does_not_fail_the_targeted_delete(self): + adapter, calls, thread_id, state = _targeted_adapter() + await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + state.delete = AsyncMock(side_effect=RuntimeError("state unavailable")) + + await adapter.delete_message(thread_id, "targeted-msg-1") + + calls.delete_targeted.assert_awaited_once_with("targeted-msg-1") + + @pytest.mark.asyncio + async def test_a_failed_targeted_delete_keeps_the_record(self): + """The record is forgotten only after Teams accepts the delete, so a + retry still uses the targeted endpoint.""" + from chat_sdk.shared.errors import AuthenticationError + + adapter, calls, thread_id, state = _targeted_adapter() + await adapter.post_ephemeral(thread_id, "29:target-user", "hi") + calls.delete_targeted.side_effect = [_SdkStatusError(401, "Unauthorized"), None] + + with pytest.raises(AuthenticationError): + await adapter.delete_message(thread_id, "targeted-msg-1") + await adapter.delete_message(thread_id, "targeted-msg-1") + + assert calls.delete_targeted.await_count == 2 + calls.delete.assert_not_called() + assert "teams:targetedActivity:19:abc@thread.tacv2:targeted-msg-1" not in state.cache + + @pytest.mark.asyncio + async def test_a_1_1_fallback_post_is_not_recorded_as_targeted(self): + adapter, calls, _thread_id, state = _targeted_adapter() + dm_thread = adapter.encode_thread_id( + TeamsThreadId(conversation_id="a:1personal-chat-id", service_url="https://smba.trafficmanager.net/teams/") + ) + + await adapter.post_ephemeral(dm_thread, "29:target-user", "hi") + await adapter.delete_message(dm_thread, "targeted-msg-1") + + assert state.cache == {} + calls.delete.assert_awaited_once_with("targeted-msg-1") + calls.delete_targeted.assert_not_called() # --------------------------------------------------------------------------- diff --git a/tests/test_teams_native_streaming.py b/tests/test_teams_native_streaming.py index 8fa34ea..051c525 100644 --- a/tests/test_teams_native_streaming.py +++ b/tests/test_teams_native_streaming.py @@ -27,6 +27,7 @@ from __future__ import annotations import asyncio +import json from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -38,7 +39,7 @@ from chat_sdk.chat import Chat from chat_sdk.testing import MockLogger, create_mock_state from chat_sdk.thread import ThreadImpl, _ThreadImplConfig -from chat_sdk.types import ChatConfig, Message, RawMessage +from chat_sdk.types import UNSET, ChatConfig, Message, RawMessage, StreamOptions def _make_adapter() -> TeamsAdapter: @@ -107,6 +108,10 @@ def __init__( raise_cancel_after: int | None = None, ) -> None: self.emitted: list[str] = [] + # Status texts passed to ``update()``, and the order of ``update`` / + # ``emit`` calls (``("update", text)`` / ``("emit", text)``). + self.updates: list[str] = [] + self.call_order: list[tuple[str, str]] = [] self._canceled = canceled self.closed = False self.close_calls = 0 @@ -128,8 +133,11 @@ def on_chunk(self, handler: Any) -> None: def on_close(self, handler: Any) -> None: # pragma: no cover - parity shim pass - def update(self, text: str) -> None: # pragma: no cover - parity shim - pass + def update(self, text: str) -> None: + # Synchronous, like the SDK ``HttpStream.update`` (which queues an + # informative typing activity through ``emit``). + self.updates.append(text) + self.call_order.append(("update", text)) def clear_text(self) -> None: # pragma: no cover - parity shim pass @@ -151,6 +159,7 @@ def emit(self, activity: Any) -> None: text = activity if isinstance(activity, str) else getattr(activity, "text", "") self.emitted.append(text) + self.call_order.append(("emit", text)) # Fire on_chunk for the FIRST emitted chunk only, like the SDK # (the first stream activity returns the assigned message id). @@ -540,6 +549,567 @@ async def gen(): assert result.raw["text"] == "" +# --------------------------------------------------------------------------- +# Placeholder-aware streaming (upstream describe "streaming", vercel/chat#709) +# --------------------------------------------------------------------------- + + +class TestPlaceholderStreaming: + @pytest.mark.asyncio + async def test_uses_core_fallback_for_an_explicit_group_chat_placeholder(self): + adapter = _make_adapter() + consumed = False + + async def source(): + nonlocal consumed + consumed = True + yield "Done" + + result = await adapter.stream( + _channel_thread_id(adapter), + source(), + StreamOptions(fallback_streaming_placeholder_text="Working..."), + ) + + assert result is None + assert consumed is False + + @pytest.mark.asyncio + async def test_an_empty_string_placeholder_also_uses_core_fallback(self): + """``""`` is an explicit placeholder (a ``str``), not "unset".""" + adapter = _make_adapter() + consumed = False + + async def source(): + nonlocal consumed + consumed = True + yield "Done" + + result = await adapter.stream( + _channel_thread_id(adapter), + source(), + StreamOptions(fallback_streaming_placeholder_text=""), + ) + + assert result is None + assert consumed is False + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "options", + [None, StreamOptions(), StreamOptions(fallback_streaming_placeholder_text=None)], + ids=["no-options", "unset", "none"], + ) + async def test_preserves_buffered_group_chat_streaming_for_placeholder(self, options: StreamOptions | None): + """No options, ``UNSET`` and ``None`` all keep the single buffered post.""" + adapter = _make_adapter() + tid = _channel_thread_id(adapter) + post_message = AsyncMock(return_value=RawMessage(id="answer-id", thread_id=tid, raw={})) + adapter.post_message = post_message # type: ignore[method-assign] + + async def gen(): + yield "Do" + yield "ne" + + result = await adapter.stream(tid, gen(), options) + + post_message.assert_awaited_once() + posted_tid, postable = post_message.call_args.args + assert posted_tid == tid + assert postable.markdown == "Done" + assert result is not None + assert result.id == "answer-id" + + @pytest.mark.asyncio + async def test_a_canceled_dm_streamer_buffers_even_with_a_placeholder(self): + """Upstream returns ``null`` only when there is no streamer at all; a + canceled one falls through to the buffered post.""" + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + await _register_streamer(adapter, tid, FakeStreamer(canceled=True)) + post_message = AsyncMock(return_value=RawMessage(id="buffered-1", thread_id=tid, raw={})) + adapter.post_message = post_message # type: ignore[method-assign] + + async def gen(): + yield "late" + + result = await adapter.stream(tid, gen(), StreamOptions(fallback_streaming_placeholder_text="Working...")) + + assert result is not None + assert result.id == "buffered-1" + assert post_message.call_args.args[1].markdown == "late" + + @pytest.mark.asyncio + async def test_sends_an_explicit_placeholder_as_native_status_before_the_first_chunk(self): + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = FakeStreamer(chunk_id="answer-id") + + async def gen(): + yield "Do" + yield "ne" + + result = await adapter._stream_via_emit(tid, gen(), streamer, "Working...") # type: ignore[arg-type] + + assert streamer.updates == ["Working..."] + assert streamer.call_order == [("update", "Working..."), ("emit", "Do"), ("emit", "ne")] + assert result.id == "answer-id" + assert result.thread_id == tid + + @pytest.mark.asyncio + async def test_stream_forwards_the_placeholder_to_the_active_dm_streamer(self): + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = FakeStreamer(chunk_id="dm-1") + await _register_streamer(adapter, tid, streamer) + + async def gen(): + yield "hi" + + result = await adapter.stream(tid, gen(), StreamOptions(fallback_streaming_placeholder_text="")) + + assert streamer.call_order == [("update", ""), ("emit", "hi")] + assert result is not None + assert result.id == "dm-1" + + @pytest.mark.asyncio + @pytest.mark.parametrize("placeholder", [UNSET, None], ids=["unset", "none"]) + async def test_unset_or_none_placeholder_sends_no_native_status(self, placeholder: Any): + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = FakeStreamer(chunk_id="dm-2") + await _register_streamer(adapter, tid, streamer) + + async def gen(): + yield "hi" + + result = await adapter.stream(tid, gen(), StreamOptions(fallback_streaming_placeholder_text=placeholder)) + + assert streamer.updates == [] + assert streamer.call_order == [("emit", "hi")] + assert result is not None + assert result.id == "dm-2" + + @pytest.mark.asyncio + async def test_a_cancel_raised_by_the_status_update_is_swallowed(self): + """``update`` goes through ``emit`` on the SDK streamer, so a stream + canceled between the check and the update raises ``StreamCancelledError`` + there; it is handled like a cancel during iteration.""" + from microsoft_teams.apps import StreamCancelledError + + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = FakeStreamer() + streamer.update = MagicMock(side_effect=StreamCancelledError("Stream has been cancelled.")) # type: ignore[method-assign] + pulled = False + + async def gen(): + nonlocal pulled + pulled = True + yield "never" + + result = await adapter._stream_via_emit(tid, gen(), streamer, "Working...") # type: ignore[arg-type] + + assert result.id == "" + assert streamer.emitted == [] + assert pulled is False + + @pytest.mark.asyncio + async def test_group_chat_placeholder_posts_then_edits_through_thread(self): + """End to end through ``ThreadImpl``: an explicit placeholder in a group + chat is posted by core and then edited to the final text.""" + adapter = _make_adapter() + tid = _channel_thread_id(adapter) + adapter.post_message = AsyncMock(return_value=RawMessage(id="ph-1", thread_id=tid, raw={})) # type: ignore[method-assign] + adapter.edit_message = AsyncMock(return_value=RawMessage(id="ph-1", thread_id=tid, raw={})) # type: ignore[method-assign] + thread = ThreadImpl( + _ThreadImplConfig( + id=tid, + channel_id="19:abc@thread.tacv2", + adapter=adapter, # type: ignore[arg-type] + state_adapter=MagicMock(), + fallback_streaming_placeholder_text="Working...", + streaming_update_interval_ms=0, + ) + ) + + async def gen(): + yield "Do" + yield "ne" + + sent = await thread.post(gen()) + + adapter.post_message.assert_awaited_once_with(tid, "Working...") + assert adapter.edit_message.await_args_list[-1].args[:2] == (tid, "ph-1") + assert sent.id == "ph-1" + assert sent.text == "Done" + + +def _sdk_has_terminal_stream_errors() -> bool: + """``microsoft-teams-apps`` 2.0.16+ raises ``StreamNotAllowedError`` / + ``TerminalStreamError`` for a non-cancel 403; 2.0.13.4 marks every 403 as + a cancel.""" + from microsoft_teams.apps.plugins import streamer + + return hasattr(streamer, "StreamNotAllowedError") + + +def _real_sdk_adapter(handler: Any) -> TeamsAdapter: + """An adapter whose real SDK clients (``App.api`` on 2.1, and the 2.0.x + ``ActivitySender`` that builds streams and sends) talk to ``handler`` + through an in-memory ``httpx`` transport.""" + import httpx + from microsoft_teams.api import ApiClient + from microsoft_teams.common import Client, ClientOptions + + client = Client(ClientOptions(token="bot-token")) + client.http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + adapter = _make_adapter() + adapter._app.api = ApiClient("https://smba.trafficmanager.net/teams", client) # type: ignore[method-assign] + sender = getattr(adapter._app, "activity_sender", None) + if sender is not None: + sender._client = client + return adapter + + +def _is_stream_activity(body: dict[str, Any]) -> bool: + return any(e.get("type") == "streaminfo" for e in body.get("entities") or []) + + +class TestFirstChunkIdWait: + """Python-only bound on the first-chunk wait: with ``microsoft-teams-apps`` + 2.0.16+, a terminal 403 on the first flush neither emits ``chunk`` nor + sets ``canceled``, so an unbounded wait would hang the handler. On expiry + the stream is settled with ``close()`` before anything else is sent.""" + + @staticmethod + def _silent_streamer(close_result: Any) -> FakeStreamer: + streamer = FakeStreamer() + streamer.on_chunk = MagicMock() # type: ignore[method-assign] # never fires + streamer.close = AsyncMock(return_value=close_result) # type: ignore[method-assign] + return streamer + + @pytest.mark.asyncio + async def test_posts_the_reply_when_no_chunk_is_ever_delivered(self, monkeypatch: pytest.MonkeyPatch): + from chat_sdk.adapters.teams import adapter as adapter_module + + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.05) + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = self._silent_streamer(None) + send = AsyncMock(return_value=_SentActivity("fallback-1")) + adapter._app.activity_sender = SimpleNamespace(send=send) + + async def gen(): + yield "hello" + + await _register_streamer(adapter, tid, streamer) + result = await asyncio.wait_for(adapter.stream(tid, gen()), 2) + + # close() reported nothing delivered, so the accumulated text is + # delivered with one buffered post instead of being dropped. + assert streamer.emitted == ["hello"] + streamer.close.assert_awaited_once() + send.assert_awaited_once() + activity, ref = send.call_args.args + assert activity.text == "hello" + assert ref.conversation.id == "a:1Abc-DM-conversation-id" + assert result is not None and result.id == "fallback-1" + adapter._logger.warn.assert_called_once() + # The failed streamer is retired (the SDK keeps the posted text in its + # buffer), so a second post in the same handler does not reuse it. + assert tid not in adapter._active_streams + + async def gen2(): + yield "again" + + await adapter.stream(tid, gen2()) + assert streamer.emitted == ["hello"] + assert [c.args[0].text for c in send.await_args_list] == ["hello", "again"] + + @pytest.mark.asyncio + async def test_uses_the_settled_stream_when_the_first_chunk_arrives_late(self, monkeypatch: pytest.MonkeyPatch): + from chat_sdk.adapters.teams import adapter as adapter_module + + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.05) + # The fake never fires ``chunk``, so the settled result's id is used. + monkeypatch.setattr(adapter_module, "STREAM_SETTLED_CHUNK_ID_TIMEOUT_S", 0.05) + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = self._silent_streamer(_SentActivity("late-stream-1")) + send = AsyncMock(return_value=_SentActivity("fallback-1")) + adapter._app.activity_sender = SimpleNamespace(send=send) + + async def gen(): + yield "hello" + + await _register_streamer(adapter, tid, streamer) + result = await asyncio.wait_for(adapter.stream(tid, gen()), 2) + + send.assert_not_awaited() + assert result is not None + assert result.id == "late-stream-1" + assert result.raw == {"text": "hello"} + # The finalized streamer is retired, so a second post in the same + # handler is sent as a plain message instead of reusing it. + assert tid not in adapter._active_streams + + async def gen2(): + yield "again" + + second = await adapter.stream(tid, gen2()) + assert second is not None and second.id == "fallback-1" + assert streamer.emitted == ["hello"] + + @pytest.mark.asyncio + async def test_sends_nothing_more_when_the_user_cancels_during_the_wait(self, monkeypatch: pytest.MonkeyPatch): + from chat_sdk.adapters.teams import adapter as adapter_module + + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.05) + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = self._silent_streamer(None) + send = AsyncMock(return_value=_SentActivity("fallback-1")) + adapter._app.activity_sender = SimpleNamespace(send=send) + + async def gen(): + yield "hello" + + task = asyncio.ensure_future(adapter._stream_via_emit(tid, gen(), streamer)) # type: ignore[arg-type] + await asyncio.sleep(0) + streamer._canceled = True # the SDK marks a cancel 403 on the background flush + result = await asyncio.wait_for(task, 2) + + streamer.close.assert_not_awaited() + send.assert_not_awaited() + assert result.id == "" + + @pytest.mark.asyncio + @pytest.mark.skipif(not _sdk_has_terminal_stream_errors(), reason="this SDK treats every streaming 403 as a cancel") + async def test_real_sdk_stream_refused_with_a_terminal_403_posts_the_reply(self, monkeypatch: pytest.MonkeyPatch): + import httpx + + from chat_sdk.adapters.teams import adapter as adapter_module + + stream_requests: list[dict[str, Any]] = [] + plain_posts: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + if _is_stream_activity(body): + stream_requests.append(body) + return httpx.Response(403, json={"error": {"message": "Content stream is not allowed"}}) + plain_posts.append(body) + return httpx.Response(200, json={"id": "fallback-post-1"}) + + adapter = _real_sdk_adapter(handler) + tid = _dm_thread_id(adapter) + stream = adapter._create_streamer(_dm_activity(), tid) + assert stream is not None + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.5) + + async def gen(): + yield "hello" + + result = await asyncio.wait_for(adapter._stream_via_emit(tid, gen(), stream), 10) + + # The streamed chunk was refused, so the reply arrives as one plain post. + assert len(stream_requests) == 1 + assert [p["text"] for p in plain_posts] == ["hello"] + assert result.id == "fallback-post-1" + assert result.raw["text"] == "hello" + # The handler's later close() sends nothing more. + assert await stream.close() is None + assert len(stream_requests) == 1 and len(plain_posts) == 1 + + @pytest.mark.asyncio + @pytest.mark.skipif(not _sdk_has_terminal_stream_errors(), reason="this SDK treats every streaming 403 as a cancel") + async def test_real_sdk_chunk_stranded_by_a_failed_flush_is_posted(self, monkeypatch: pytest.MonkeyPatch): + """A chunk emitted while the first flush awaits its 403 stays queued + with no flush scheduled: that is not an in-flight send, so the whole + reply is posted.""" + import httpx + + from chat_sdk.adapters.teams import adapter as adapter_module + + stream_requests: list[dict[str, Any]] = [] + plain_posts: list[dict[str, Any]] = [] + first_flush_started = asyncio.Event() + + async def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + if _is_stream_activity(body): + stream_requests.append(body) + first_flush_started.set() + await asyncio.sleep(0.1) + return httpx.Response(403, json={"error": {"message": "Content stream is not allowed"}}) + plain_posts.append(body) + return httpx.Response(200, json={"id": "fallback-post-1"}) + + adapter = _real_sdk_adapter(handler) + tid = _dm_thread_id(adapter) + stream = adapter._create_streamer(_dm_activity(), tid) + assert stream is not None + stream._total_wait_timeout = 0.05 # type: ignore[attr-defined] # the SDK's 30 s close() wait + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.5) + + async def gen(): + yield "hello" + await first_flush_started.wait() + yield " world" + + result = await asyncio.wait_for(adapter._stream_via_emit(tid, gen(), stream), 10) + + assert len(stream_requests) == 1 + assert stream.count == 1 # " world" stranded in the SDK queue + assert [p["text"] for p in plain_posts] == ["hello world"] + assert result.id == "fallback-post-1" + + @pytest.mark.asyncio + async def test_real_sdk_stream_canceled_on_the_first_flush_posts_nothing(self, monkeypatch: pytest.MonkeyPatch): + import httpx + + from chat_sdk.adapters.teams import adapter as adapter_module + + requests: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + requests.append(body) + if _is_stream_activity(body): + return httpx.Response(403, json={"error": {"message": "Content stream was cancelled by user"}}) + return httpx.Response(200, json={"id": "fallback-post-1"}) + + adapter = _real_sdk_adapter(handler) + tid = _dm_thread_id(adapter) + stream = adapter._create_streamer(_dm_activity(), tid) + assert stream is not None + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.5) + + async def gen(): + yield "hello" + + result = await asyncio.wait_for(adapter._stream_via_emit(tid, gen(), stream), 10) + + assert stream.canceled is True + assert [_is_stream_activity(r) for r in requests] == [True] + assert result.id == "" + + @pytest.mark.asyncio + async def test_real_sdk_cancel_while_settling_returns_normally(self, monkeypatch: pytest.MonkeyPatch): + """The slow first flush succeeds, then the user cancels while the + stream is being finalized: the cancel is swallowed like every other + native-stream cancel, and nothing is posted.""" + import httpx + + from chat_sdk.adapters.teams import adapter as adapter_module + + requests: list[dict[str, Any]] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + requests.append(body) + if len(requests) == 1: + await asyncio.sleep(0.3) + return httpx.Response(200, json={"id": "stream-msg-1"}) + return httpx.Response(403, json={"error": {"message": "Content stream was cancelled by user"}}) + + adapter = _real_sdk_adapter(handler) + tid = _dm_thread_id(adapter) + stream = adapter._create_streamer(_dm_activity(), tid) + assert stream is not None + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.05) + + async def gen(): + yield "hello" + + result = await asyncio.wait_for(adapter._stream_via_emit(tid, gen(), stream), 10) + + assert stream.canceled is True + assert [_is_stream_activity(r) for r in requests] == [True, True] + assert result.id == "" + assert result.raw == {"text": "hello"} + + @pytest.mark.asyncio + async def test_real_sdk_slow_first_flush_is_not_duplicated(self, monkeypatch: pytest.MonkeyPatch): + """The first flush is still in flight when the bound expires: the + reply is finalized as a stream, never also posted, and keeps the id + Teams assigned to the first chunk.""" + import httpx + + from chat_sdk.adapters.teams import adapter as adapter_module + + requests: list[dict[str, Any]] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + requests.append(body) + if len(requests) == 1: + await asyncio.sleep(0.3) + return httpx.Response(200, json={"id": "stream-msg-1"}) + # Teams returns the id only on the first streaming response. + return httpx.Response(200, json={}) + + adapter = _real_sdk_adapter(handler) + tid = _dm_thread_id(adapter) + stream = adapter._create_streamer(_dm_activity(), tid) + assert stream is not None + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.05) + + async def gen(): + yield "hello" + + result = await asyncio.wait_for(adapter._stream_via_emit(tid, gen(), stream), 10) + + # One streaming chunk, then the stream's final message; no plain post. + assert [_is_stream_activity(r) for r in requests] == [True, True] + assert [r.get("text") for r in requests] == ["hello", "hello"] + assert result.id == "stream-msg-1" + assert await stream.close() is not None + assert len(requests) == 2 + + @pytest.mark.asyncio + async def test_real_sdk_flush_outlasting_close_is_not_also_posted(self, monkeypatch: pytest.MonkeyPatch): + """The first flush is still in flight after both the first-chunk bound + and ``close()``'s own wait: no buffered post is sent, because the SDK + can still deliver the stream (the SDK client has no request timeout), + and the handler's later ``close()`` finalizes it.""" + import httpx + + from chat_sdk.adapters.teams import adapter as adapter_module + + requests: list[dict[str, Any]] = [] + release = asyncio.Event() + + async def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + requests.append(body) + if len(requests) == 1: + await release.wait() + return httpx.Response(200, json={"id": "stream-msg-1"}) + return httpx.Response(200, json={}) + + adapter = _real_sdk_adapter(handler) + tid = _dm_thread_id(adapter) + stream = adapter._create_streamer(_dm_activity(), tid) + assert stream is not None + stream._total_wait_timeout = 0.05 # type: ignore[attr-defined] # the SDK's 30 s close() wait + monkeypatch.setattr(adapter_module, "STREAM_FIRST_CHUNK_ID_TIMEOUT_S", 0.05) + + async def gen(): + yield "hello" + + result = await asyncio.wait_for(adapter._stream_via_emit(tid, gen(), stream), 10) + + assert [_is_stream_activity(r) for r in requests] == [True] + assert result.id == "" + # The flush completes later; the handler's close() finalizes the stream. + release.set() + stream._total_wait_timeout = 5 # type: ignore[attr-defined] + assert await stream.close() is not None + assert [_is_stream_activity(r) for r in requests] == [True, True] + + # --------------------------------------------------------------------------- # Webhook-level lifecycle (end-to-end through _handle_message_activity) # ---------------------------------------------------------------------------