diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd464a..a5275ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -427,6 +427,13 @@ Ports upstream `32687038` (vercel/chat#830, chat@4.38.1), the Google Chat part o - New exports from `chat_sdk.ai`: `ChatApprovalToolName`, `ReadScope`; `ToolOptions.guard`; `ChatToolsOptions.scope` / `strict_scope`. - **Python-specific (divergence from upstream):** link-metadata bounds count code points, not UTF-16 code units, so a field with emoji keeps up to the limit in code points and never ends on a lone surrogate. See `docs/UPSTREAM_SYNC.md`. - Fidelity: `ai/index.test.ts` 26 → 0 missing, `ai/messages.test.ts` 10 → 9 at `chat@4.41.1`. +- **Slack: Enterprise Grid org-wide installs, `authorizations[]` routing, event retry marker** (#268, split from #213; ports the non-cache half of vercel/chat `907450d7` #724, chat@4.35.0). + - **Consumer-visible: org-wide OAuth installs now succeed.** `handle_oauth_callback` used to raise `missing access_token or team.id` for an org-wide install (`team: null`). It now stores it under `enterprise.id`, the key org-wide webhooks resolve by, and raises `missing access_token or enterprise.id` when that is absent. The result gains `enterprise_id` and `is_enterprise_install`; `team_id` is the storage key. `SlackInstallation` gains `enterprise_id` / `is_enterprise_install`. + - **Consumer-visible: retried events that were already dispatched are dropped.** Each dispatched event writes a `slack:event-delivered:{event_id}` state key (24 h TTL, fire-and-forget). A delivery with `x-slack-retry-num > 0` (or a forwarded socket event with `retryNum > 0`) whose key exists is acked and not processed; first deliveries never read state, and a retry whose original never arrived is still processed. Live Socket Mode retries are still skipped until #283. + - Multi-workspace events resolve their installation from `authorizations[0]` before the top-level `team_id` / `enterprise_id`, so Slack Connect events route to the receiving installation. Socket Mode `events_api`, slash commands and interactive payloads now resolve org-wide installs by enterprise ID, like HTTP. + - Under an org-wide install, the adapter's Web API calls send the event's workspace `team_id`, and calls to the event's channel echo a Slack Connect `context_team_id` as `client_context_team_id`. A `team_id` the caller passes wins. The #95 `chat_stream` `team_id` is unchanged. + - `W…` user ids count as raw user ids in outgoing `@mentions`. `with_bot_token` / `with_bot_token_async` accept keyword-only `installation_id=` to scope installation-owned caches outside webhooks. `RequestContext` gains `team_id`, `context_team_id` and `context_channel`. + - The Socket Mode `events_api` envelope now keeps `is_ext_shared_channel`, as upstream, so shared channels seen over Socket Mode are marked external. - **Slack inbound mentions and authors: self-mention decoding, content-based `is_mention`, bot author ids, `email` / `is_system`** (part (a) of #209; ports vercel/chat `bb7cd124` #716, `80def3ab` #707 (Slack half), `51322dde` #891, `c2b6bff0` #883 and `683eadc1` #947, chat@4.35.0–4.41.0). The mrkdwn normalization, `channel.post` thread ids and Socket Mode retries are split out to #283. - **Breaking/consumer-visible:** - **`message.text` decodes the bot's own mention.** `<@U_BOT> hi` used to read `@U_BOT hi`. It now reads `@BotName hi`, or `@U_BOT hi` when `users.info` fails. `formatted` changes the same way. *Migration:* to detect a mention, use `message.is_mention`, not a search for the bot id in `text`. To remove the mention, strip the bot's display name, or read the id markup from `message.raw["text"]` / `raw["blocks"]`. diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index cfaf4d7..1ec5334 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -1314,6 +1314,78 @@ SDK-free cards-primitives surface, so upstream's two table converters (and the modals to cards (see the "Teams dialog/modal inbound" row in the non-parity table), so only the primitive gains the new children. +### Slack Enterprise Grid: org-wide installs, `authorizations[]`, retry marker (chat@4.35, #268) + +Parity with the non-cache half of upstream `907450d7` (vercel/chat#724, +chat@4.35.0); the installation-scoped caches landed in #205. Code is in +`slack/adapter.py` and `slack/types.py`. + +- **OAuth.** `handle_oauth_callback` keys an org-wide install + (`is_enterprise_install`, `team: null`) by `enterprise.id` and raises + `AuthenticationError("Slack OAuth failed: missing access_token or + enterprise.id")` without it. The result gains `enterprise_id` and + `is_enterprise_install`; `team_id` is always the storage key. + `SlackInstallation` gains `enterprise_id` / `is_enterprise_install`, stored + as `enterpriseId` / `isEnterpriseInstall` only when set (upstream's spread + drops undefined keys), so a plain install's stored shape is unchanged. +- **Event routing.** `_resolve_event_request_context(payload)` returns a + `RequestContext`, `"not-applicable"` or `"unresolved"` and is shared by the + HTTP and Socket Mode `events_api` paths. It prefers `authorizations[0]` + (`is_enterprise_install` via `??`, ids via `||`) over the top-level fields + and records `team_id`, `context_team_id` (an envelope field) and + `context_channel` (`event.channel`). Slash commands go through + `_run_slash_command` on both paths, and Socket Mode interactive payloads + through `_extract_installation_from_interactive_payload`, so all three + resolve org-wide installs by enterprise ID. Socket slash fields that arrive + as JSON booleans become `"true"` / `"false"` (JS `String(v)`). The socket + `events_api` envelope now keeps `authorizations`, `context_team_id`, the + enterprise fields and `is_ext_shared_channel`, as upstream, so shared + channels seen over Socket Mode are marked external too. +- **`team_id` / `client_context_team_id`.** `_with_token_kwargs(**kwargs)` + (upstream `withToken`; the token is already bound to the client) adds + `team_id` under an org-wide context and `client_context_team_id` on calls + whose `channel` is the event's channel. Keys the caller set (not `None`) + win. It wraps the same call sites as upstream; `chat_stream`, + `chat_scheduleMessage` / `chat_deleteScheduledMessage`, `files_upload_v2` + and `oauth_v2_access` stay unwrapped as upstream, so the #95 `chat_stream` + `team_id` (`recipient_team_id`) is unchanged under an org-wide context. +- **Retry marker.** `_process_event_payload` writes + `slack:event-delivered:{event_id}` (24 h TTL) as a fire-and-forget task (a + pinned reference plus a done-callback that logs at debug). + `_is_duplicate_event_delivery(payload, retry_num)` reads it only when + `retry_num > 0` (HTTP `x-slack-retry-num`, a malformed header counts as 0; + socket `retry_attempt`; forwarded `retryNum`), and a state read error means + "process". +- **`W…` user ids.** `SLACK_USER_ID_EXACT_PATTERN` is `^[UW][A-Z0-9]+$`. +- **`with_bot_token(token, fn, *, installation_id=None)`** and + `with_bot_token_async(...)` set `RequestContext.installation_id`. Upstream + passes an async function to `withBotToken`; in Python a coroutine returned + by the sync form runs after the context is reset, so async work belongs in + `with_bot_token_async` (the ported cache tests use it). + +Python-specific notes: + +- **Socket retries before #283.** `_on_socket_request` still acks and skips + envelopes with `retry_attempt > 0`, so on a live socket the marker only + sees first deliveries; it already covers HTTP retries and forwarded socket + events. #283 (split from #209) removes the skip, after which `retry_attempt` reaches the + marker unchanged. +- **Flag normalization.** `is_enterprise_install` counts only as `True` or + `"true"` everywhere (upstream's event path uses `Boolean(...)`, so a + `"false"` string would count there). Slack sends booleans in event JSON, so + this differs only for malformed payloads. +- **`with_token` scope.** Only the adapter's own calls are wrapped; the + public `web_client` / `current_client` are plain clients, as upstream's + `webClient`. + +Regression coverage: `tests/test_slack_enterprise_grid.py` (ports the +`handleOAuthCallback` Grid cases, `socket mode - multi-workspace token +resolution`, `withToken enterprise context injection`, `event delivery +deduplication`, `W-prefixed enterprise user IDs` and `event routing via +authorizations[]`), the two `withBotToken` cases in +`tests/test_slack_webhook.py::TestInstallationScopedCaches`, and +`tests/test_slack_api.py::TestStream::test_stream_keeps_the_recipient_team_id_under_an_org_wide_context` +for #95. ### Teams routing and outbound text (chat@4.36–4.41, #216) Ports `7062c395` (vercel/chat#898, chat@4.40.0), `257a32d0` (#746, @@ -2980,7 +3052,7 @@ stay explicit instead of being rediscovered in code review. | 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. | -| 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. | +| 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`. | diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 3769385..d0bbdde 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -218,7 +218,8 @@ def build_feedback_buttons_block(options: SlackFeedbackButtonsOptions | None = N # --------------------------------------------------------------------------- SLACK_USER_ID_PATTERN = re.compile(r"^[A-Z0-9_]+$") -SLACK_USER_ID_EXACT_PATTERN = re.compile(r"^U[A-Z0-9]+$") +# Enterprise Grid users can have W-prefixed IDs anywhere a U-prefixed ID appears +SLACK_USER_ID_EXACT_PATTERN = re.compile(r"^[UW][A-Z0-9]+$") SLACK_MESSAGE_URL_PATTERN = re.compile(r"^https?://[^/]+\.slack\.com/archives/([A-Z0-9]+)/p(\d+)(?:\?.*)?$") # Bracketed URL in message text; length-bounded to keep the scan linear on @@ -325,6 +326,10 @@ def _with_inherited(message: dict[str, Any], **fallbacks: Any) -> dict[str, Any] _UNFURL_WAIT_MS = 2000 _UNFURL_POLL_MS = 150 _UNFURL_CACHE_TTL_MS = 60 * 60 * 1000 # 1 hour +# How long delivered event IDs are remembered for retry deduplication. Slack +# retries at ~1 min and ~5 min, and the opt-in Delayed Events feature +# redelivers hourly for up to 24 hours (upstream ``EVENT_DEDUPE_TTL_MS``). +_EVENT_DEDUPE_TTL_MS = 24 * 60 * 60 * 1000 # 24 hours # --------------------------------------------------------------------------- @@ -343,6 +348,35 @@ class _InstallationInfo: installation_id: str is_enterprise_install: bool enterprise_id: str | None = None + team_id: str | None = None + + +def _is_true_flag(value: Any) -> bool: + """Normalize a Slack ``is_enterprise_install`` flag. + + Form-encoded webhooks deliver ``"true"``, JSON and Socket Mode deliver + ``True``, and the field may be absent (upstream + ``v === true || v === "true"``). + """ + return value is True or value == "true" + + +def _parse_retry_num(value: Any) -> int: + """Parse a Slack redelivery count (``x-slack-retry-num`` / ``retryNum``). + + Missing or malformed values count as a first delivery (``0``), as + upstream's ``Number(...)`` yields ``NaN`` and ``NaN > 0`` is false. + """ + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value.strip()) + except ValueError: + return 0 + return 0 def _json_stringify(value: Any) -> str: @@ -1848,6 +1882,39 @@ def _get_client(self, token: str | None = None) -> Any: self._client_cache.popitem(last=False) return client + def _with_token_kwargs(self, **kwargs: Any) -> dict[str, Any]: + """Add Enterprise Grid context to a Web API call's kwargs. + + Port of upstream ``withToken`` (vercel/chat#724); the token itself is + already bound to the client from :meth:`_get_client`. Org-wide tokens + span every workspace in the org, so workspace-scoped methods need an + explicit ``team_id``: it is added from the event's workspace when the + request runs under an org-wide install. When the incoming event + carried a ``context_team_id`` (a shared channel hosted on an "away" + workspace), it is echoed as ``client_context_team_id`` on calls to the + originating channel. Keys the caller set (not ``None``) are never + overridden. + + Applied at the call sites where upstream uses ``withToken``. As + upstream, ``chat_stream`` (which already sends the #95 ``team_id``), + ``chat_scheduleMessage``/``chat_deleteScheduledMessage``, + ``files_upload_v2`` and ``oauth_v2_access`` are not routed through it. + """ + ctx = self._request_context.get() + if ctx is None: + return kwargs + if ctx.is_enterprise_install and ctx.team_id and kwargs.get("team_id") is None: + kwargs["team_id"] = ctx.team_id + channel = kwargs.get("channel") + if ( + ctx.context_team_id + and channel is not None + and channel == ctx.context_channel + and kwargs.get("client_context_team_id") is None + ): + kwargs["client_context_team_id"] = ctx.context_team_id + return kwargs + def _invalidate_client(self, token: str) -> None: """Remove a cached client (e.g., on token revocation). @@ -1904,7 +1971,7 @@ async def initialize(self, chat: ChatInstance) -> None: try: token = await self._resolve_default_token() client = self._get_client(token) - auth_result = await client.auth_test() + auth_result = await client.auth_test(**self._with_token_kwargs()) self._bot_user_id = auth_result.get("user_id") self._bot_id = auth_result.get("bot_id") or None user = auth_result.get("user") @@ -1968,6 +2035,12 @@ async def set_installation(self, team_id: str, installation: SlackInstallation) "botUserId": installation.bot_user_id, "teamName": installation.team_name, } + # Enterprise Grid fields are stored only when set, as upstream's + # ``{...installation}`` spread drops undefined keys (vercel/chat#724). + if installation.enterprise_id is not None: + data_to_store["enterpriseId"] = installation.enterprise_id + if installation.is_enterprise_install is not None: + data_to_store["isEnterpriseInstall"] = installation.is_enterprise_install await state.set(key, data_to_store) self._logger.info( @@ -1993,6 +2066,10 @@ async def get_installation(self, team_id: str) -> SlackInstallation | None: bot_token_raw = (stored.get("botToken") or stored.get("bot_token")) if isinstance(stored, dict) else None bot_user_id = (stored.get("botUserId") or stored.get("bot_user_id") or "") if isinstance(stored, dict) else "" team_name = (stored.get("teamName") or stored.get("team_name") or "") if isinstance(stored, dict) else "" + enterprise_id_raw = stored.get("enterpriseId") if isinstance(stored, dict) else None + enterprise_id = enterprise_id_raw if isinstance(enterprise_id_raw, str) else None + is_enterprise_raw = stored.get("isEnterpriseInstall") if isinstance(stored, dict) else None + is_enterprise_install = is_enterprise_raw if isinstance(is_enterprise_raw, bool) else None if self._encryption_key and is_encrypted_token_data(bot_token_raw): # `is_encrypted_token_data` is a runtime type guard but doesn't # carry TypeGuard narrowing, so pyrefly still sees `None`. Assert @@ -2010,12 +2087,16 @@ async def get_installation(self, team_id: str) -> SlackInstallation | None: bot_token=decrypted, bot_user_id=bot_user_id, team_name=team_name, + enterprise_id=enterprise_id, + is_enterprise_install=is_enterprise_install, ) return SlackInstallation( bot_token=bot_token_raw if isinstance(bot_token_raw, str) else "", bot_user_id=bot_user_id, team_name=team_name, + enterprise_id=enterprise_id, + is_enterprise_install=is_enterprise_install, ) async def handle_oauth_callback( @@ -2031,7 +2112,14 @@ async def handle_oauth_callback( during the code exchange. When provided it takes priority over any ``redirect_uri`` query parameter in the callback URL. - Returns ``{"team_id": ..., "installation": SlackInstallation}``. + For Enterprise Grid org-wide installs (``is_enterprise_install``), + Slack returns ``team: null`` and the installation is keyed by the + enterprise ID instead (vercel/chat#724). + + Returns ``{"team_id": ..., "enterprise_id": ..., "is_enterprise_install": + ..., "installation": SlackInstallation}``. ``team_id`` is always the + storage key (the enterprise ID for org-wide installs), so it can be + passed back to :meth:`get_installation` / :meth:`delete_installation`. """ if not (self._client_id and self._client_secret): raise ValidationError( @@ -2072,21 +2160,44 @@ async def handle_oauth_callback( kwargs["redirect_uri"] = redirect_uri result = await client.oauth_v2_access(**kwargs) - if not (result.get("ok") and result.get("access_token") and result.get("team", {}).get("id")): + # Org-wide installs return ``team: null`` and identify the org via + # ``enterprise.id``: key the installation by whichever ID webhook token + # resolution will later look up (see ``_resolve_event_request_context``). + team_raw = result.get("team") + team: dict[str, Any] = team_raw if isinstance(team_raw, dict) else {} + enterprise_raw = result.get("enterprise") + enterprise: dict[str, Any] = enterprise_raw if isinstance(enterprise_raw, dict) else {} + is_enterprise_install = bool(result.get("is_enterprise_install")) + enterprise_id: str | None = enterprise.get("id") or None + installation_id: str | None = enterprise_id if is_enterprise_install else (team.get("id") or None) + + if not (result.get("ok") and result.get("access_token") and installation_id): + missing = ( + "missing access_token or enterprise.id" if is_enterprise_install else "missing access_token or team.id" + ) raise AuthenticationError( "slack", - f"Slack OAuth failed: {result.get('error') or 'missing access_token or team.id'}", + f"Slack OAuth failed: {result.get('error') or missing}", ) - team_id = result["team"]["id"] + team_name = team.get("name") installation = SlackInstallation( bot_token=result["access_token"], bot_user_id=result.get("bot_user_id"), - team_name=result.get("team", {}).get("name"), + team_name=team_name if team_name is not None else enterprise.get("name"), + enterprise_id=enterprise_id, + # Recorded only for org-wide installs (upstream spreads the flag + # only when true), so workspace installs keep ``None``. + is_enterprise_install=True if is_enterprise_install else None, ) - await self.set_installation(team_id, installation) - return {"team_id": team_id, "installation": installation} + await self.set_installation(installation_id, installation) + return { + "team_id": installation_id, + "enterprise_id": enterprise_id, + "is_enterprise_install": is_enterprise_install, + "installation": installation, + } async def delete_installation(self, team_id: str) -> None: """Remove a workspace installation.""" @@ -2099,17 +2210,30 @@ async def delete_installation(self, team_id: str) -> None: await state.delete(self._installation_key(team_id)) self._logger.info("Slack installation deleted", {"teamId": team_id}) - def with_bot_token(self, token: str, fn: Callable[[], Any]) -> Any: - """Run *fn* with a specific bot token in context (for cron jobs, etc.).""" - tok = self._request_context.set(RequestContext(token=token)) + def with_bot_token(self, token: str, fn: Callable[[], Any], *, installation_id: str | None = None) -> Any: + """Run *fn* with a specific bot token in context (for cron jobs, etc.). + + In multi-workspace deployments pass ``installation_id`` (the + ``team_id``, or the ``enterprise_id`` for org-wide installs: the key + the installation was stored under) so installation-owned caches (user + profiles, display-name index, channel names) are scoped to it. + Without it they use the unscoped global keys (vercel/chat#724). + """ + tok = self._request_context.set(RequestContext(token=token, installation_id=installation_id)) try: return fn() finally: self._request_context.reset(tok) - async def with_bot_token_async(self, token: str, fn: Callable[[], Awaitable[Any]]) -> Any: - """Run an async function with a specific bot token in context.""" - tok = self._request_context.set(RequestContext(token=token)) + async def with_bot_token_async( + self, token: str, fn: Callable[[], Awaitable[Any]], *, installation_id: str | None = None + ) -> Any: + """Run an async function with a specific bot token in context. + + ``installation_id`` scopes installation-owned caches, as in + :meth:`with_bot_token`. + """ + tok = self._request_context.set(RequestContext(token=token, installation_id=installation_id)) try: return await fn() finally: @@ -2178,23 +2302,163 @@ def _extract_installation_from_interactive(self, body: str) -> _InstallationInfo if not payload_str: return None payload = json.loads(payload_str) - is_enterprise_install = bool(payload.get("is_enterprise_install")) - enterprise = payload.get("enterprise") or {} - enterprise_id = enterprise.get("id") or payload.get("enterprise_id") or None - team = payload.get("team") or {} - team_id = team.get("id") or payload.get("team_id") or None - installation_id = enterprise_id if is_enterprise_install else team_id - - if not installation_id: + if not isinstance(payload, dict): return None - return _InstallationInfo( - installation_id=installation_id, - is_enterprise_install=is_enterprise_install, - enterprise_id=enterprise_id, - ) + return self._extract_installation_from_interactive_payload(payload) except Exception: return None + def _extract_installation_from_interactive_payload(self, payload: dict[str, Any]) -> _InstallationInfo | None: + """Extract installation info from an already-parsed interactive payload. + + Socket Mode delivers these as objects rather than form-urlencoded, and + ``is_enterprise_install`` may arrive as ``True`` or ``"true"`` + (upstream ``extractInstallationFromInteractivePayload``). + """ + is_enterprise_install = _is_true_flag(payload.get("is_enterprise_install")) + enterprise = payload.get("enterprise") + team = payload.get("team") + enterprise_id = (enterprise.get("id") if isinstance(enterprise, dict) else None) or payload.get("enterprise_id") + team_id = (team.get("id") if isinstance(team, dict) else None) or payload.get("team_id") + enterprise_id = enterprise_id if isinstance(enterprise_id, str) and enterprise_id else None + team_id = team_id if isinstance(team_id, str) and team_id else None + installation_id = enterprise_id if is_enterprise_install else team_id + + if not installation_id: + return None + return _InstallationInfo( + installation_id=installation_id, + is_enterprise_install=is_enterprise_install, + enterprise_id=enterprise_id, + team_id=team_id, + ) + + async def _resolve_interactive_request_context(self, info: _InstallationInfo) -> RequestContext | None: + """Resolve the token for an interactive payload's installation. + + Returns the request context to dispatch under, or ``None`` when no + installation was found. Shared by the HTTP and Socket Mode paths. + """ + ctx = await self._resolve_token_for_team(info.installation_id, info.is_enterprise_install) + if ctx is None: + return None + return replace( + ctx, + enterprise_id=info.enterprise_id, + is_enterprise_install=info.is_enterprise_install, + installation_id=info.installation_id, + team_id=info.team_id, + ) + + async def _resolve_event_request_context( + self, payload: dict[str, Any] + ) -> RequestContext | Literal["not-applicable", "unresolved"]: + """Resolve the multi-workspace request context for an event payload. + + Port of upstream ``resolveEventRequestContext`` (vercel/chat#724), + shared by the HTTP webhook and Socket Mode paths. Returns: + + - the resolved context to run the event under, + - ``"not-applicable"`` when no resolution is needed (single-workspace + mode, a non-``event_callback`` payload, or no installation ID): + process without a request context, + - ``"unresolved"`` when an installation ID was present but no + installation was found: drop the event. + + ``authorizations[0]`` is preferred: Slack documents it as the + authoritative installation identity, while the top-level fields can + name a different (Slack Connect) workspace. Top-level fields are the + fallback for payloads that omit it. + """ + if self._is_single_workspace or payload.get("type") != "event_callback": + return "not-applicable" + + authorizations = payload.get("authorizations") + auth: dict[str, Any] = {} + if isinstance(authorizations, list) and authorizations and isinstance(authorizations[0], dict): + auth = authorizations[0] + # ``auth?.is_enterprise_install ?? payload.is_enterprise_install`` + raw_flag = auth.get("is_enterprise_install") + is_enterprise_install = _is_true_flag( + raw_flag if raw_flag is not None else payload.get("is_enterprise_install") + ) + # ``auth?.x || payload.x``: an empty or null authorization field falls + # back to the top-level one. + enterprise_id = auth.get("enterprise_id") or payload.get("enterprise_id") or None + team_id = auth.get("team_id") or payload.get("team_id") or None + installation_id = enterprise_id if is_enterprise_install else team_id + if not installation_id: + return "not-applicable" + + ctx = await self._resolve_token_for_team(installation_id, is_enterprise_install) + if ctx is None: + self._logger.warn( + "Could not resolve token for installation", + {"installationId": installation_id, "isEnterpriseInstall": is_enterprise_install}, + ) + return "unresolved" + + event = payload.get("event") + return replace( + ctx, + enterprise_id=enterprise_id, + is_enterprise_install=is_enterprise_install, + installation_id=installation_id, + team_id=team_id, + # context_team_id is an envelope top-level field, not inside `event`. + context_team_id=payload.get("context_team_id") or None, + context_channel=(event.get("channel") or None) if isinstance(event, dict) else None, + ) + + def _mark_event_delivered(self, payload: dict[str, Any]) -> None: + """Record that an event delivery was dispatched (fire-and-forget). + + Port of upstream ``markEventDelivered``: lets a redelivery + (``x-slack-retry-num`` / socket ``retry_attempt``) be dropped. A failed + write only means a retry gets reprocessed, which message dedupe + already tolerates, so errors are logged and never fail the webhook. + """ + event_id = payload.get("event_id") + if not (event_id and self._chat): + return + try: + loop = asyncio.get_running_loop() + task = loop.create_task( + self._chat.get_state().set(f"slack:event-delivered:{event_id}", True, _EVENT_DEDUPE_TTL_MS) + ) + except Exception as exc: + self._logger.debug("Failed to record event delivery", {"eventId": event_id, "error": exc}) + return + + def _log_failure(t: asyncio.Task[Any]) -> None: + exc = None if t.cancelled() else t.exception() + if exc is not None: + self._logger.debug("Failed to record event delivery", {"eventId": event_id, "error": exc}) + + task.add_done_callback(_log_failure) + _pin_task(task) + + async def _is_duplicate_event_delivery(self, payload: dict[str, Any], retry_num: int | None = None) -> bool: + """Return ``True`` when a retried delivery was already dispatched. + + Port of upstream ``isDuplicateEventDelivery``. Only consulted on + retries (``retry_num > 0``), so first deliveries pay no state read. + Events whose first delivery never arrived have no marker and are + still recovered via the retry. A state error means "process". + """ + event_id = payload.get("event_id") + if not (event_id and self._chat and retry_num and retry_num > 0): + return False + try: + seen = await self._chat.get_state().get(f"slack:event-delivered:{event_id}") + except Exception: + # State unavailable: process rather than drop. + return False + if seen: + self._logger.info("Skipping duplicate event delivery", {"eventId": event_id, "retryNum": retry_num}) + return True + return False + # ================================================================== # User / Channel lookup with caching # ================================================================== @@ -2253,7 +2517,7 @@ async def _lookup_user(self, user_id: str) -> SlackUserCacheEntry: try: client = self._get_client() - result = await client.users_info(user=user_id) + result = await client.users_info(**self._with_token_kwargs(user=user_id)) user = result.get("user") or {} # Slack can return `{"ok": True, "user": {}}` in some edge cases # (rare, but observed when scopes are partial or the workspace @@ -2336,7 +2600,7 @@ async def _lookup_channel(self, channel_id: str) -> str: try: client = self._get_client() - result = await client.conversations_info(channel=channel_id) + result = await client.conversations_info(**self._with_token_kwargs(channel=channel_id)) channel = result.get("channel", {}) name = channel.get("name", channel_id) @@ -2547,53 +2811,14 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No # Slash command if "command" in params and "payload" not in params: - if not self._is_single_workspace: - # For Enterprise Grid org-wide installs, use enterprise_id; - # otherwise use team_id. - is_enterprise_install = (params.get("is_enterprise_install") or [None])[0] == "true" - enterprise_id = (params.get("enterprise_id") or [None])[0] - team_id = (params.get("team_id") or [None])[0] - installation_id = enterprise_id if is_enterprise_install else team_id - - if installation_id: - ctx = await self._resolve_token_for_team(installation_id, is_enterprise_install) - if ctx: - ctx = replace( - ctx, - enterprise_id=enterprise_id, - is_enterprise_install=is_enterprise_install, - installation_id=installation_id, - ) - tok = self._request_context.set(ctx) - try: - return await self._handle_slash_command(params, options) - finally: - self._request_context.reset(tok) - self._logger.warn( - "Could not resolve token for slash command", - {"installationId": installation_id, "isEnterpriseInstall": is_enterprise_install}, - ) - # Missing or unresolved installation: acknowledge without - # dispatching, so handlers never run with no token context - # (vercel/chat#877). - return {"body": "", "status": 200} - return await self._handle_slash_command(params, options) + return await self._run_slash_command(params, options) # Interactive payload if not self._is_single_workspace: installation_info = self._extract_installation_from_interactive(body) if installation_info: - ctx = await self._resolve_token_for_team( - installation_info.installation_id, - installation_info.is_enterprise_install, - ) + ctx = await self._resolve_interactive_request_context(installation_info) if ctx: - ctx = replace( - ctx, - enterprise_id=installation_info.enterprise_id, - is_enterprise_install=installation_info.is_enterprise_install, - installation_id=installation_info.installation_id, - ) tok = self._request_context.set(ctx) try: return await self._handle_interactive_payload(body, options) @@ -2619,41 +2844,73 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No "headers": {"Content-Type": "application/json"}, } + # Drop redeliveries of events that were already dispatched + retry_header = headers.get("x-slack-retry-num") or headers.get("X-Slack-Retry-Num") + if await self._is_duplicate_event_delivery(payload, _parse_retry_num(retry_header)): + return {"body": "ok", "status": 200} + # Multi-workspace: resolve token before processing events. # Use contextvars.copy_context() so the ContextVar value persists into # any async tasks spawned by _process_event_payload (e.g. process_message # creates a task via asyncio.create_task). The copied context is # isolated -- the ContextVar change does not leak back to the caller # and does not need an explicit reset. - if not self._is_single_workspace and payload.get("type") == "event_callback": - # For Enterprise Grid org-wide installs, use enterprise_id; - # otherwise use team_id. - is_enterprise_install = bool(payload.get("is_enterprise_install")) - installation_id = payload.get("enterprise_id") if is_enterprise_install else payload.get("team_id") - - if installation_id: - ctx = await self._resolve_token_for_team(installation_id, is_enterprise_install) - if ctx: - ctx = replace( - ctx, - enterprise_id=payload.get("enterprise_id"), - is_enterprise_install=is_enterprise_install, - installation_id=installation_id, - ) - isolated = contextvars.copy_context() - isolated.run(self._request_context.set, ctx) - isolated.run(self._process_event_payload, payload, options) - return {"body": "ok", "status": 200} - self._logger.warn( - "Could not resolve token for installation", - {"installationId": installation_id, "isEnterpriseInstall": is_enterprise_install}, - ) - return {"body": "ok", "status": 200} + resolved = await self._resolve_event_request_context(payload) + if resolved == "unresolved": + # Installation ID present but no installation found: drop the event + return {"body": "ok", "status": 200} + if isinstance(resolved, RequestContext): + isolated = contextvars.copy_context() + isolated.run(self._request_context.set, resolved) + isolated.run(self._process_event_payload, payload, options) + return {"body": "ok", "status": 200} # Single-workspace mode or fallback self._process_event_payload(payload, options) return {"body": "ok", "status": 200} + async def _run_slash_command( + self, + params: dict[str, list[str]], + options: WebhookOptions | None = None, + ) -> dict[str, Any]: + """Handle a slash command, resolving the installation token first. + + Port of upstream ``runSlashCommand`` (vercel/chat#724), shared by the + HTTP webhook and Socket Mode paths. A missing or unresolved + installation is acknowledged without dispatching, so handlers never + run with no token context (vercel/chat#877). + """ + if self._is_single_workspace: + return await self._handle_slash_command(params, options) + # For Enterprise Grid org-wide installs, use enterprise_id; otherwise + # use team_id. + is_enterprise_install = (params.get("is_enterprise_install") or [None])[0] == "true" + enterprise_id = (params.get("enterprise_id") or [None])[0] + team_id = (params.get("team_id") or [None])[0] + installation_id = enterprise_id if is_enterprise_install else team_id + + if installation_id: + ctx = await self._resolve_token_for_team(installation_id, is_enterprise_install) + if ctx: + ctx = replace( + ctx, + enterprise_id=enterprise_id, + is_enterprise_install=is_enterprise_install, + installation_id=installation_id, + team_id=team_id, + ) + tok = self._request_context.set(ctx) + try: + return await self._handle_slash_command(params, options) + finally: + self._request_context.reset(tok) + self._logger.warn( + "Could not resolve token for slash command", + {"installationId": installation_id, "isEnterpriseInstall": is_enterprise_install}, + ) + return {"body": "", "status": 200} + # ================================================================== # Event dispatch # ================================================================== @@ -2663,13 +2920,12 @@ def _process_event_payload(self, payload: dict[str, Any], options: WebhookOption if payload.get("type") != "event_callback" or not payload.get("event"): return + self._mark_event_delivered(payload) event: dict[str, Any] = payload["event"] - # Track external/shared channel status. Note: socket-mode payloads - # synthesized in ``_route_socket_event`` never carry this field, which - # mirrors upstream's ``routeSocketEvent`` shape. Socket-mode adapters - # therefore won't populate ``_external_channels`` from this path — - # documented as a known divergence in ``docs/UPSTREAM_SYNC.md``. + # Track external/shared channel status from the payload-level flag. + # The socket-mode payload synthesized in ``_route_socket_event`` + # carries it too (upstream ``routeSocketEvent``, vercel/chat#724). if payload.get("is_ext_shared_channel"): channel_id = event.get("channel") or (event.get("item", {}).get("channel") if "item" in event else None) if channel_id: @@ -3427,7 +3683,10 @@ async def ack(response_payload: dict[str, Any] | None = None) -> None: self._logger.debug("Skipping socket mode retry", {"retry_attempt": retry_attempt}) return - await self._route_socket_event(payload, event_type, ack) + # ``retry_attempt`` feeds the event_id retry marker. While the skip + # above stays (#283 replaces it with upstream's "process retries"), + # only forwarded socket events reach the marker with a retry count. + await self._route_socket_event(payload, event_type, ack, retry_num=_parse_retry_num(retry_attempt)) async def _route_socket_event( self, @@ -3435,6 +3694,7 @@ async def _route_socket_event( event_type: str, ack: Callable[..., Awaitable[None]], options: WebhookOptions | None = None, + retry_num: int | None = None, ) -> None: """Route a socket-mode event to the same handler the webhook path uses. @@ -3442,7 +3702,9 @@ async def _route_socket_event( the SocketModeResponse back to Slack — for events_api and slash_commands we ack immediately and let processing run in the background; for interactive payloads we may attach a response body - (e.g. modal ``view_submission`` errors) onto the ack. + (e.g. modal ``view_submission`` errors) onto the ack. ``retry_num`` is + Slack's redelivery count for the envelope, used to drop retries of + events that were already dispatched. """ def wrap_async(coro: Awaitable[Any]) -> None: @@ -3475,35 +3737,35 @@ def _log_exc(t: asyncio.Task[Any]) -> None: {"body_type": type(body).__name__}, ) return - # Match the webhook path's synthesized payload exactly. Upstream - # doesn't include ``is_ext_shared_channel`` here, and the webhook - # JSON we pass into ``_process_event_payload`` doesn't either — - # adding it on the socket path is a quiet socket-vs-webhook - # divergence (hazard #7). Keep the keys that flow into - # downstream handlers, drop the rest. + # Rebuild the envelope with the fields upstream's + # ``routeSocketEvent`` keeps (vercel/chat#724): the installation + # identity (``authorizations``, enterprise fields), the Slack + # Connect context and the shared-channel flag. payload: dict[str, Any] = { "type": "event_callback", "event": event, + "authorizations": body.get("authorizations"), "team_id": body.get("team_id"), + "context_team_id": body.get("context_team_id"), + "enterprise_id": body.get("enterprise_id"), + "is_enterprise_install": _is_true_flag(body.get("is_enterprise_install")), + "is_ext_shared_channel": body.get("is_ext_shared_channel"), "event_id": body.get("event_id"), "event_time": body.get("event_time"), } - # Multi-workspace: resolve token before dispatch (mirrors webhook - # path). copy_context() keeps the ContextVar set on tasks spawned - # by handlers (hazard #6). - team_id_event = payload.get("team_id") try: - if not self._is_single_workspace and team_id_event: - ctx = await self._resolve_token_for_team(team_id_event) - if ctx is None: - self._logger.warn( - "Could not resolve token for team", - {"teamId": team_id_event}, - ) - return - ctx = replace(ctx, installation_id=team_id_event) + # Drop redeliveries of events that were already dispatched + if await self._is_duplicate_event_delivery(payload, retry_num): + return + # Resolve the per-installation token exactly like the HTTP + # path. copy_context() keeps the ContextVar set on tasks + # spawned by handlers (hazard #6). + resolved = await self._resolve_event_request_context(payload) + if resolved == "unresolved": + return + if isinstance(resolved, RequestContext): isolated = contextvars.copy_context() - isolated.run(self._request_context.set, ctx) + isolated.run(self._request_context.set, resolved) isolated.run(self._process_event_payload, payload, options) else: self._process_event_payload(payload, options) @@ -3524,47 +3786,31 @@ def _log_exc(t: asyncio.Task[Any]) -> None: # slash_commands payload is a flat dict mirroring the # form-urlencoded fields; convert to the parse_qs shape that # _handle_slash_command expects (each value wrapped in a list). - params: dict[str, list[str]] = {k: [v] for k, v in body.items() if isinstance(v, str)} - - async def run_slash() -> None: - if self._is_single_workspace: - await self._handle_slash_command(params, options) - return - team_id_slash = (params.get("team_id") or [None])[0] - ctx = await self._resolve_token_for_team(team_id_slash) if team_id_slash else None - if ctx is None: - # Missing or unresolved installation: already acked, do - # not dispatch without a token context (vercel/chat#877). - self._logger.warn("Could not resolve token for slash command") - return - ctx = replace(ctx, installation_id=team_id_slash) - tok = self._request_context.set(ctx) - try: - await self._handle_slash_command(params, options) - finally: - self._request_context.reset(tok) - - wrap_async(run_slash()) + # Socket Mode delivers the fields as JSON, so flags such as + # ``is_enterprise_install`` arrive as booleans: render them as the + # form strings (JS ``String(true)``). + params: dict[str, list[str]] = {} + for key, value in body.items(): + if isinstance(value, bool): + params[key] = ["true" if value else "false"] + elif isinstance(value, str): + params[key] = [value] + + wrap_async(self._run_slash_command(params, options)) return if event_type == "interactive": try: # Multi-workspace: scope token resolution to the dispatch. - team_ref = body.get("team") - # Upstream ``team?.id || payload.team_id``: a ``team`` object - # without an id still falls back to the top-level field. - team_id_interactive = (team_ref.get("id") if isinstance(team_ref, dict) else None) or body.get( - "team_id" - ) if not self._is_single_workspace: - ctx = await self._resolve_token_for_team(team_id_interactive) if team_id_interactive else None + info = self._extract_installation_from_interactive_payload(body) + ctx = await self._resolve_interactive_request_context(info) if info else None if ctx is None: # Missing or unresolved installation: ack without # dispatching (vercel/chat#877). self._logger.warn("Could not resolve token for socket interactive payload") await ack() return - ctx = replace(ctx, installation_id=team_id_interactive) tok = self._request_context.set(ctx) try: result = await self._dispatch_interactive_payload(body, options) @@ -3628,7 +3874,11 @@ async def noop_ack(_response: dict[str, Any] | None = None) -> None: {"event_type": type(event_type).__name__}, ) return - await self._route_socket_event(body, event_type, noop_ack, options) + # ``retryNum`` is the listener's redelivery count for the original + # socket envelope (upstream ``SlackForwardedSocketEvent.retryNum``). + await self._route_socket_event( + body, event_type, noop_ack, options, retry_num=_parse_retry_num(event.get("retryNum")) + ) # ================================================================== # Message events @@ -3823,10 +4073,7 @@ async def _resolve_and_process() -> None: try: client = self._get_client() result = await client.conversations_replies( - channel=channel, - ts=message_id, - limit=1, - inclusive=True, + **self._with_token_kwargs(channel=channel, ts=message_id, limit=1, inclusive=True), ) msgs = result.get("messages", []) if msgs: @@ -4081,7 +4328,7 @@ def _handle_user_change(self, event: dict[str, Any]) -> None: async def publish_home_view(self, user_id: str, view: dict[str, Any]) -> None: """Publish a Home tab view for a user.""" client = self._get_client() - await client.views_publish(user_id=user_id, view=view) + await client.views_publish(**self._with_token_kwargs(user_id=user_id, view=view)) async def set_suggested_prompts( self, @@ -4109,7 +4356,9 @@ async def set_suggested_prompts( # ``assistant_threads_setSuggestedPrompts`` sends) because that helper # requires ``thread_ts`` before slack-sdk 3.43.0, which is newer than # the ``slack-sdk>=3.40.0`` floor. - await client.api_call(api_method="assistant.threads.setSuggestedPrompts", json=payload) + await client.api_call( + api_method="assistant.threads.setSuggestedPrompts", json=self._with_token_kwargs(**payload) + ) def _schedule_configured_suggested_prompts( self, context: SlackSuggestedPromptsContext, options: WebhookOptions | None @@ -4192,12 +4441,14 @@ async def set_assistant_status( effective_loading_messages = loading_messages if loading_messages is not None else self._loading_messages if effective_loading_messages: kwargs["loading_messages"] = effective_loading_messages - await client.assistant_threads_setStatus(**kwargs) + await client.assistant_threads_setStatus(**self._with_token_kwargs(**kwargs)) async def set_assistant_title(self, channel_id: str, thread_ts: str, title: str) -> None: """Set title for an assistant thread (shown in History tab).""" client = self._get_client() - await client.assistant_threads_setTitle(channel_id=channel_id, thread_ts=thread_ts, title=title) + await client.assistant_threads_setTitle( + **self._with_token_kwargs(channel_id=channel_id, thread_ts=thread_ts, title=title) + ) # ================================================================== # Mention resolution @@ -4713,7 +4964,9 @@ def _create_link_preview(self, url: str) -> LinkPreview: async def fetch_message() -> Message: client = self._get_client() - result = await client.conversations_history(channel=channel, latest=ts, inclusive=True, limit=1) + result = await client.conversations_history( + **self._with_token_kwargs(channel=channel, latest=ts, inclusive=True, limit=1) + ) messages = result.get("messages", []) target = next((m for m in messages if m.get("ts") == ts), None) if not target: @@ -5454,12 +5707,14 @@ async def post_message(self, thread_id: str, message: AdapterPostableMessage) -> ) try: result = await client.chat_postMessage( - channel=channel, - thread_ts=thread_ts or None, - text=fallback_text, - blocks=blocks, - unfurl_links=False, - unfurl_media=False, + **self._with_token_kwargs( + channel=channel, + thread_ts=thread_ts or None, + text=fallback_text, + blocks=blocks, + unfurl_links=False, + unfurl_media=False, + ), ) except Exception as error: enriched = self._enrich_invalid_blocks_error(error, blocks) @@ -5485,11 +5740,9 @@ async def post_message(self, thread_id: str, message: AdapterPostableMessage) -> }, ) result = await client.chat_postMessage( - channel=channel, - thread_ts=thread_ts or None, - unfurl_links=False, - unfurl_media=False, - **payload, + **self._with_token_kwargs( + channel=channel, thread_ts=thread_ts or None, unfurl_links=False, unfurl_media=False, **payload + ), ) return RawMessage( id=result.get("ts", ""), @@ -5561,7 +5814,9 @@ async def edit_message( if card: blocks = card_to_block_kit(card) fallback_text = card_to_fallback_text(card) - result = await client.chat_update(channel=channel, ts=message_id, text=fallback_text, blocks=blocks) + result = await client.chat_update( + **self._with_token_kwargs(channel=channel, ts=message_id, text=fallback_text, blocks=blocks) + ) return RawMessage( id=result.get("ts", ""), thread_id=thread_id, @@ -5577,7 +5832,7 @@ async def edit_message( "payloadKey": "markdown_text" if "markdown_text" in payload else "text", }, ) - result = await client.chat_update(channel=channel, ts=message_id, **payload) + result = await client.chat_update(**self._with_token_kwargs(channel=channel, ts=message_id, **payload)) return RawMessage( id=result.get("ts", ""), thread_id=thread_id, @@ -5601,7 +5856,7 @@ async def delete_message(self, thread_id: str, message_id: str) -> None: try: client = self._get_client() self._logger.debug("Slack API: chat.delete", {"channel": channel, "messageId": message_id}) - await client.chat_delete(channel=channel, ts=message_id) + await client.chat_delete(**self._with_token_kwargs(channel=channel, ts=message_id)) except Exception as error: self._handle_slack_error(error) @@ -5622,7 +5877,7 @@ async def add_reaction(self, thread_id: str, message_id: str, emoji: EmojiValue "Slack API: reactions.add", {"channel": channel, "messageId": message_id, "emoji": name}, ) - await client.reactions_add(channel=channel, timestamp=message_id, name=name) + await client.reactions_add(**self._with_token_kwargs(channel=channel, timestamp=message_id, name=name)) except Exception as error: self._handle_slack_error(error) @@ -5639,7 +5894,7 @@ async def remove_reaction(self, thread_id: str, message_id: str, emoji: EmojiVal "Slack API: reactions.remove", {"channel": channel, "messageId": message_id, "emoji": name}, ) - await client.reactions_remove(channel=channel, timestamp=message_id, name=name) + await client.reactions_remove(**self._with_token_kwargs(channel=channel, timestamp=message_id, name=name)) except Exception as error: self._handle_slack_error(error) @@ -5677,10 +5932,9 @@ async def start_typing( try: client = self._get_client() await client.assistant_threads_setStatus( - channel_id=channel, - thread_ts=thread_ts, - status=status_text, - loading_messages=loading_messages, + **self._with_token_kwargs( + channel_id=channel, thread_ts=thread_ts, status=status_text, loading_messages=loading_messages + ), ) except Exception as exc: self._logger.warn( @@ -6326,11 +6580,9 @@ async def post_ephemeral( blocks = card_to_block_kit(card) fallback_text = card_to_fallback_text(card) result = await client.chat_postEphemeral( - channel=channel, - thread_ts=thread_ts or None, - user=user_id, - text=fallback_text, - blocks=blocks, + **self._with_token_kwargs( + channel=channel, thread_ts=thread_ts or None, user=user_id, text=fallback_text, blocks=blocks + ), ) return EphemeralMessage( id=result.get("message_ts", ""), @@ -6350,10 +6602,7 @@ async def post_ephemeral( }, ) result = await client.chat_postEphemeral( - channel=channel, - thread_ts=thread_ts or None, - user=user_id, - **payload, + **self._with_token_kwargs(channel=channel, thread_ts=thread_ts or None, user=user_id, **payload), ) return EphemeralMessage( id=result.get("message_ts", ""), @@ -6465,7 +6714,7 @@ async def open_dm(self, user_id: str) -> str: try: client = self._get_client() self._logger.debug("Slack API: conversations.open", {"userId": user_id}) - result = await client.conversations_open(users=user_id) + result = await client.conversations_open(**self._with_token_kwargs(users=user_id)) channel_info = result.get("channel", {}) channel_id = channel_info.get("id") if not channel_id: @@ -6496,7 +6745,7 @@ async def open_modal(self, trigger_id: str, modal: dict[str, Any], context_id: s try: client = self._get_client() - result = await client.views_open(trigger_id=trigger_id, view=view) + result = await client.views_open(**self._with_token_kwargs(trigger_id=trigger_id, view=view)) view_id = (result.get("view") or {}).get("id", "") return {"viewId": view_id} except Exception as error: @@ -6508,7 +6757,7 @@ async def update_modal(self, view_id: str, modal: dict[str, Any]) -> dict[str, s try: client = self._get_client() - result = await client.views_update(view_id=view_id, view=view) + result = await client.views_update(**self._with_token_kwargs(view_id=view_id, view=view)) new_view_id = (result.get("view") or {}).get("id", "") return {"viewId": new_view_id} except Exception as error: @@ -6597,7 +6846,9 @@ async def _fetch_messages_forward( cursor: str | None = None, ) -> FetchResult: client = self._get_client() - result = await client.conversations_replies(channel=channel, ts=thread_ts, limit=limit, cursor=cursor) + result = await client.conversations_replies( + **self._with_token_kwargs(channel=channel, ts=thread_ts, limit=limit, cursor=cursor) + ) slack_messages = result.get("messages", []) next_cursor = (result.get("response_metadata") or {}).get("next_cursor") @@ -6617,7 +6868,7 @@ async def _fetch_messages_backward( client = self._get_client() result = await client.conversations_replies( - channel=channel, ts=thread_ts, limit=fetch_limit, latest=latest, inclusive=False + **self._with_token_kwargs(channel=channel, ts=thread_ts, limit=fetch_limit, latest=latest, inclusive=False), ) slack_messages = result.get("messages", []) @@ -6647,10 +6898,14 @@ async def fetch_message(self, thread_id: str, message_id: str) -> Message | None # message. Fetch the single message from conversations.history # instead (mirrors the link-preview fetch_message at ~3293). if not thread_ts: - result = await client.conversations_history(channel=channel, latest=message_id, inclusive=True, limit=1) + result = await client.conversations_history( + **self._with_token_kwargs(channel=channel, latest=message_id, inclusive=True, limit=1) + ) else: result = await client.conversations_replies( - channel=channel, ts=thread_ts, oldest=message_id, inclusive=True, limit=1 + **self._with_token_kwargs( + channel=channel, ts=thread_ts, oldest=message_id, inclusive=True, limit=1 + ), ) messages = result.get("messages", []) target = next((m for m in messages if m.get("ts") == message_id), None) @@ -6668,7 +6923,7 @@ async def fetch_thread(self, thread_id: str) -> ThreadInfo: try: client = self._get_client() - result = await client.conversations_info(channel=channel) + result = await client.conversations_info(**self._with_token_kwargs(channel=channel)) channel_info = result.get("channel", {}) if channel_info.get("is_ext_shared"): @@ -6724,7 +6979,7 @@ async def _fetch_channel_messages_forward(self, channel: str, limit: int, cursor if cursor: kwargs["oldest"] = cursor kwargs["inclusive"] = False - result = await client.conversations_history(**kwargs) + result = await client.conversations_history(**self._with_token_kwargs(**kwargs)) slack_messages = list(reversed(result.get("messages", []))) messages = await asyncio.gather( @@ -6753,7 +7008,7 @@ async def _fetch_channel_messages_backward( if cursor: kwargs["latest"] = cursor kwargs["inclusive"] = False - result = await client.conversations_history(**kwargs) + result = await client.conversations_history(**self._with_token_kwargs(**kwargs)) slack_messages = result.get("messages", []) chronological = list(reversed(slack_messages)) @@ -6788,9 +7043,9 @@ async def list_threads(self, channel_id: str, options: ListThreadsOptions | None try: client = self._get_client() result = await client.conversations_history( - channel=channel, - limit=min(limit * 3, 200), - cursor=getattr(opts, "cursor", None), + **self._with_token_kwargs( + channel=channel, limit=min(limit * 3, 200), cursor=getattr(opts, "cursor", None) + ), ) slack_messages = result.get("messages", []) @@ -6832,7 +7087,7 @@ async def fetch_channel_info(self, channel_id: str) -> ChannelInfo: try: client = self._get_client() - result = await client.conversations_info(channel=channel) + result = await client.conversations_info(**self._with_token_kwargs(channel=channel)) info = result.get("channel", {}) if info.get("is_ext_shared"): diff --git a/src/chat_sdk/adapters/slack/types.py b/src/chat_sdk/adapters/slack/types.py index 906a72d..b733512 100644 --- a/src/chat_sdk/adapters/slack/types.py +++ b/src/chat_sdk/adapters/slack/types.py @@ -299,6 +299,13 @@ class SlackInstallation: bot_token: str bot_user_id: str | None = None team_name: str | None = None + # Enterprise Grid org ID, when the workspace belongs to (or the app is + # installed org-wide on) a Grid org + enterprise_id: str | None = None + # True when this is an Enterprise Grid org-wide installation (stored under + # the enterprise ID). ``None`` when not recorded, as upstream's optional + # ``isEnterpriseInstall``. + is_enterprise_install: bool | None = None class SlackInstallationProvider(Protocol): @@ -736,3 +743,13 @@ class RequestContext: # cache keys (user profiles, display-name index, channel names, unfurl # metadata) so one workspace's data is never served to another. installation_id: str | None = None + # Workspace the current event occurred in. Sent as an explicit ``team_id`` + # on Web API calls made with an org-wide token (``is_enterprise_install``). + team_id: str | None = None + # ``context_team_id`` from the incoming event envelope, echoed back as + # ``client_context_team_id`` on calls addressed to ``context_channel`` + # (shared channels hosted on an "away" workspace). + context_team_id: str | None = None + # Channel the ``context_team_id`` came from; the echo only applies to + # calls targeting that channel. + context_channel: str | None = None diff --git a/tests/test_slack_api.py b/tests/test_slack_api.py index ff316f7..519b988 100644 --- a/tests/test_slack_api.py +++ b/tests/test_slack_api.py @@ -2169,6 +2169,55 @@ async def text_gen() -> AsyncIterator[str]: assert captured["team_id"] == "T_GRID_WS" assert captured["recipient_team_id"] == "T_GRID_WS" + @pytest.mark.asyncio + async def test_stream_keeps_the_recipient_team_id_under_an_org_wide_context(self): + """#95 under an Enterprise Grid org-wide request context (#268). + + Org-wide contexts inject the event's ``team_id`` into Web API calls + (``_with_token_kwargs``), but ``chat_stream`` is not routed through + it (as upstream's ``chatStream``), so the #95 ``team_id`` stays the + ``recipient_team_id`` and no ``client_context_team_id`` is added. + """ + from chat_sdk.adapters.slack.types import RequestContext + + adapter, client, _ = await _init_adapter() + captured: dict[str, Any] = {} + streamer = MagicMock() + streamer.append = AsyncMock(return_value={"ok": True}) + streamer.stop = AsyncMock(return_value={"ok": True, "ts": "1234567890.951951"}) + + async def chat_stream(**kwargs: Any) -> Any: + captured.update(kwargs) + return streamer + + client.chat_stream = AsyncMock(side_effect=chat_stream) + + async def text_gen() -> AsyncIterator[str]: + yield "streamed hello on grid" + + tok = adapter._request_context.set( + RequestContext( + token="xoxb-org", + is_enterprise_install=True, + installation_id="E_ORG", + team_id="T_EVENT_WS", + context_team_id="T_AWAY", + context_channel="C_GRID", + ) + ) + try: + result = await adapter.stream( + "slack:C_GRID:1234567890.000000", + text_gen(), + StreamOptions(recipient_user_id="U_GRID", recipient_team_id="T_GRID_WS"), + ) + finally: + adapter._request_context.reset(tok) + + assert result.id == "1234567890.951951" + assert captured["team_id"] == "T_GRID_WS" + assert "client_context_team_id" not in captured + @pytest.mark.asyncio async def test_stream_without_team_id_on_grid_falls_back_to_post_and_edit(self): """Mutation guard for issue #95: prove the Grid simulation actually diff --git a/tests/test_slack_enterprise_grid.py b/tests/test_slack_enterprise_grid.py new file mode 100644 index 0000000..bdd6642 --- /dev/null +++ b/tests/test_slack_enterprise_grid.py @@ -0,0 +1,995 @@ +"""Slack Enterprise Grid support (#268). + +Ports the non-cache half of vercel/chat ``907450d7`` (#724, chat@4.35.0) +from ``packages/adapter-slack/src/index.test.ts``: org-wide OAuth installs, +``authorizations[0]`` event routing, Socket Mode per-installation token +resolution, ``team_id`` / ``client_context_team_id`` injection, the +``event_id`` retry marker and ``W``-prefixed user IDs. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import hmac +import json +import time +from collections.abc import Awaitable, Callable +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +try: + from chat_sdk.adapters.slack.adapter import SlackAdapter + from chat_sdk.adapters.slack.types import RequestContext, SlackAdapterConfig, SlackInstallation + from chat_sdk.cards import Card + from chat_sdk.shared.errors import AuthenticationError + from chat_sdk.state.memory import MemoryStateAdapter + from chat_sdk.types import FetchOptions + + _SLACK_AVAILABLE = True +except ImportError: + _SLACK_AVAILABLE = False + +pytestmark = [ + pytest.mark.skipif(not _SLACK_AVAILABLE, reason="Slack adapter import failed"), + pytest.mark.asyncio, +] + +_SECRET = "test-signing-secret" + +_MESSAGE_EVENT = { + "type": "message", + "user": "U_USER", + "channel": "C123", + "text": "hello from socket", + "ts": "1234567890.123456", +} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class _FakeRequest: + def __init__(self, body: str, headers: dict[str, str] | None = None, url: str = "") -> None: + self.body = body.encode("utf-8") + self.headers = headers or {} + self.url = url + + async def text(self) -> str: + return self.body.decode("utf-8") + + +def _signed_request( + body: str, content_type: str = "application/json", extra_headers: dict[str, str] | None = None +) -> _FakeRequest: + ts = str(int(time.time())) + sig = "v0=" + hmac.new(_SECRET.encode(), f"v0:{ts}:{body}".encode(), hashlib.sha256).hexdigest() + headers = {"x-slack-request-timestamp": ts, "x-slack-signature": sig, "content-type": content_type} + headers.update(extra_headers or {}) + return _FakeRequest(body, headers) + + +def _make_chat(state: Any) -> MagicMock: + chat = MagicMock() + chat.process_message = MagicMock() + chat.process_action = MagicMock() + chat.process_slash_command = MagicMock() + chat.get_state = MagicMock(return_value=state) + chat.get_user_name = MagicMock(return_value="test-bot") + chat.get_logger = MagicMock(return_value=MagicMock()) + return chat + + +async def _memory_state() -> MemoryStateAdapter: + state = MemoryStateAdapter() + await state.connect() + return state + + +async def _multi_workspace_adapter(**config: Any) -> tuple[SlackAdapter, MagicMock, MemoryStateAdapter]: + """Multi-workspace adapter (no default bot token) over a real memory state.""" + state = await _memory_state() + chat = _make_chat(state) + adapter = SlackAdapter(SlackAdapterConfig(signing_secret=_SECRET, **config)) + await adapter.initialize(chat) # type: ignore[arg-type] + return adapter, chat, state + + +async def _single_workspace_adapter() -> tuple[SlackAdapter, MagicMock, MemoryStateAdapter]: + state = await _memory_state() + chat = _make_chat(state) + adapter = SlackAdapter(SlackAdapterConfig(signing_secret=_SECRET, bot_token="xoxb-test-token", bot_user_id="U_BOT")) + await adapter.initialize(chat) # type: ignore[arg-type] + return adapter, chat, state + + +def _spy_resolver(adapter: SlackAdapter) -> AsyncMock: + """Wrap ``_resolve_token_for_team`` so calls are recorded but still resolve.""" + spy = AsyncMock(side_effect=adapter._resolve_token_for_team) + adapter._resolve_token_for_team = spy # type: ignore[method-assign] + return spy + + +async def _settle() -> None: + """Let fire-and-forget tasks (marker writes, slash dispatch) run.""" + for _ in range(10): + await asyncio.sleep(0) + + +def _oauth_adapter(access_result: dict[str, Any]) -> SlackAdapter: + adapter = SlackAdapter( + SlackAdapterConfig(signing_secret=_SECRET, client_id="client-id", client_secret="client-secret") + ) + client = MagicMock() + client.oauth_v2_access = AsyncMock(return_value=access_result) + adapter._client_cache[""] = client + return adapter + + +def _oauth_request(code: str = "oauth-code") -> _FakeRequest: + return _FakeRequest("", url=f"https://example.com/auth/callback/slack?code={code}") + + +_ORG_WIDE_ACCESS = { + "ok": True, + "access_token": "xoxb-org-bot-token", + "bot_user_id": "U_BOT_ORG", + "team": None, + "enterprise": {"id": "E_ORG_1", "name": "Acme Org"}, + "is_enterprise_install": True, +} + + +# --------------------------------------------------------------------------- +# handleOAuthCallback +# --------------------------------------------------------------------------- + + +class TestHandleOAuthCallbackEnterprise: + async def test_keys_org_wide_installs_by_enterprise_id_team_is_null(self): + adapter = _oauth_adapter(dict(_ORG_WIDE_ACCESS)) + await adapter.initialize(_make_chat(await _memory_state())) # type: ignore[arg-type] + + result = await adapter.handle_oauth_callback(_oauth_request("oauth-code-org")) + + assert result["team_id"] == "E_ORG_1" + assert result["enterprise_id"] == "E_ORG_1" + assert result["is_enterprise_install"] is True + assert result["installation"].team_name == "Acme Org" + + # Stored under the enterprise ID: the same key org-wide webhooks + # (is_enterprise_install: true) resolve tokens by. + stored = await adapter.get_installation("E_ORG_1") + assert stored is not None + assert stored.bot_token == "xoxb-org-bot-token" + assert stored.enterprise_id == "E_ORG_1" + assert stored.is_enterprise_install is True + + async def test_records_the_enterprise_id_on_workspace_installs_within_a_grid_org(self): + adapter = _oauth_adapter( + { + "ok": True, + "access_token": "xoxb-grid-workspace-token", + "bot_user_id": "U_BOT_GRID", + "team": {"id": "T_GRID_1", "name": "Grid Workspace"}, + "enterprise": {"id": "E_ORG_1", "name": "Acme Org"}, + "is_enterprise_install": False, + } + ) + await adapter.initialize(_make_chat(await _memory_state())) # type: ignore[arg-type] + + result = await adapter.handle_oauth_callback(_oauth_request("oauth-code-grid")) + + assert result["team_id"] == "T_GRID_1" + assert result["enterprise_id"] == "E_ORG_1" + assert result["is_enterprise_install"] is False + + stored = await adapter.get_installation("T_GRID_1") + assert stored is not None + assert stored.bot_token == "xoxb-grid-workspace-token" + assert stored.team_name == "Grid Workspace" + assert stored.enterprise_id == "E_ORG_1" + assert stored.is_enterprise_install is None + + async def test_throws_when_an_org_wide_install_response_is_missing_enterprise_id(self): + adapter = _oauth_adapter( + { + "ok": True, + "access_token": "xoxb-org-bot-token", + "team": None, + "enterprise": None, + "is_enterprise_install": True, + } + ) + state = await _memory_state() + await adapter.initialize(_make_chat(state)) # type: ignore[arg-type] + + with pytest.raises(AuthenticationError, match="missing access_token or enterprise.id"): + await adapter.handle_oauth_callback(_oauth_request("oauth-code-org")) + + async def test_org_wide_oauth_install_round_trips_with_org_wide_event_webhooks(self): + adapter = _oauth_adapter(dict(_ORG_WIDE_ACCESS)) + chat = _make_chat(await _memory_state()) + await adapter.initialize(chat) # type: ignore[arg-type] + seen: list[RequestContext | None] = [] + chat.process_message.side_effect = lambda *a, **kw: seen.append(adapter._request_context.get()) + + await adapter.handle_oauth_callback(_oauth_request()) + + body = json.dumps( + { + "type": "event_callback", + "team_id": "T_GRID_1", + "enterprise_id": "E_ORG_1", + "is_enterprise_install": True, + "event": {**_MESSAGE_EVENT, "text": "Hello org", "channel": "C456"}, + } + ) + response = await adapter.handle_webhook(_signed_request(body)) + + assert response["status"] == 200 + assert chat.process_message.call_count == 1 + ctx = seen[0] + assert ctx is not None + assert (ctx.token, ctx.installation_id, ctx.team_id, ctx.is_enterprise_install) == ( + "xoxb-org-bot-token", + "E_ORG_1", + "T_GRID_1", + True, + ) + + async def test_workspace_install_keeps_team_name_and_omits_enterprise_fields(self): + """Python-specific: the stored shape of a plain (non-Grid) install is + unchanged, so installations written before #268 read back the same.""" + adapter = _oauth_adapter( + { + "ok": True, + "access_token": "xoxb-plain", + "team": {"id": "T_PLAIN", "name": "Plain"}, + } + ) + state = await _memory_state() + await adapter.initialize(_make_chat(state)) # type: ignore[arg-type] + + result = await adapter.handle_oauth_callback(_oauth_request()) + + assert (result["team_id"], result["enterprise_id"], result["is_enterprise_install"]) == ("T_PLAIN", None, False) + assert await state.get("slack:installation:T_PLAIN") == { + "botToken": "xoxb-plain", + "botUserId": None, + "teamName": "Plain", + } + + +class TestEncryptedEnterpriseInstallation: + async def test_enterprise_fields_round_trip_with_encryption(self): + import base64 + import os + + key = base64.b64encode(os.urandom(32)).decode() + adapter, _, _ = await _multi_workspace_adapter(encryption_key=key) + + await adapter.set_installation( + "E_ORG_1", + SlackInstallation(bot_token="xoxb-org", enterprise_id="E_ORG_1", is_enterprise_install=True), + ) + stored = await adapter.get_installation("E_ORG_1") + + assert stored is not None + assert (stored.bot_token, stored.enterprise_id, stored.is_enterprise_install) == ("xoxb-org", "E_ORG_1", True) + + +# --------------------------------------------------------------------------- +# socket mode - multi-workspace token resolution +# --------------------------------------------------------------------------- + + +class TestSocketModeMultiWorkspaceTokenResolution: + async def test_resolves_the_per_workspace_token_for_events_api(self): + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation("T_SOCK_1", SlackInstallation(bot_token="xoxb-sock-token", bot_user_id="U_BOT")) + resolve = _spy_resolver(adapter) + + await adapter._route_socket_event( + {"team_id": "T_SOCK_1", "event": dict(_MESSAGE_EVENT)}, "events_api", AsyncMock() + ) + + resolve.assert_awaited_once_with("T_SOCK_1", False) + assert chat.process_message.call_count == 1 + + async def test_resolves_org_wide_installs_by_enterprise_id_for_events_api(self): + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation( + "E_ORG_1", + SlackInstallation(bot_token="xoxb-org-token", bot_user_id="U_BOT_ORG", is_enterprise_install=True), + ) + resolve = _spy_resolver(adapter) + + await adapter._route_socket_event( + { + "team_id": "T_ANY", + "enterprise_id": "E_ORG_1", + "is_enterprise_install": True, + "event": dict(_MESSAGE_EVENT), + }, + "events_api", + AsyncMock(), + ) + + resolve.assert_awaited_once_with("E_ORG_1", True) + assert chat.process_message.call_count == 1 + + async def test_drops_events_api_events_when_no_installation_is_found(self): + adapter, chat, _ = await _multi_workspace_adapter() + + await adapter._route_socket_event( + {"team_id": "T_UNKNOWN", "event": dict(_MESSAGE_EVENT)}, "events_api", AsyncMock() + ) + + chat.process_message.assert_not_called() + + async def test_resolves_tokens_for_slash_commands_with_boolean_is_enterprise_install(self): + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation( + "E_ORG_1", SlackInstallation(bot_token="xoxb-org-token", is_enterprise_install=True) + ) + resolve = _spy_resolver(adapter) + seen: list[RequestContext | None] = [] + chat.process_slash_command.side_effect = lambda *a, **kw: seen.append(adapter._request_context.get()) + + await adapter._route_socket_event( + { + "command": "/test", + "text": "arg1", + "user_id": "U_USER", + "channel_id": "C123", + "team_id": "T_ANY", + "enterprise_id": "E_ORG_1", + # Socket mode delivers form fields as JSON, so this arrives boolean + "is_enterprise_install": True, + }, + "slash_commands", + AsyncMock(), + ) + await _settle() + + assert chat.process_slash_command.call_count == 1 + resolve.assert_awaited_once_with("E_ORG_1", True) + ctx = seen[0] + assert ctx is not None + assert (ctx.token, ctx.installation_id, ctx.team_id) == ("xoxb-org-token", "E_ORG_1", "T_ANY") + + async def test_resolves_tokens_for_interactive_payloads(self): + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation("T_SOCK_2", SlackInstallation(bot_token="xoxb-sock-token-2")) + resolve = _spy_resolver(adapter) + + await adapter._route_socket_event( + { + "type": "block_actions", + "team": {"id": "T_SOCK_2"}, + "actions": [{"type": "button", "action_id": "test_action", "value": "v"}], + "channel": {"id": "C123", "name": "test"}, + "container": {"type": "message", "message_ts": "1234567890.123456", "channel_id": "C123"}, + "message": {"ts": "1234567890.123456"}, + "trigger_id": "trigger123", + "user": {"id": "U_USER", "username": "testuser"}, + }, + "interactive", + AsyncMock(), + ) + + resolve.assert_awaited_once_with("T_SOCK_2", False) + assert chat.process_action.call_count == 1 + + async def test_resolves_org_wide_interactive_payloads_by_enterprise_id(self): + """Python-specific: the socket interactive path now honors the + enterprise identity (it used to resolve by ``team.id`` only).""" + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation("E_ORG_1", SlackInstallation(bot_token="xoxb-org", is_enterprise_install=True)) + resolve = _spy_resolver(adapter) + seen: list[RequestContext | None] = [] + chat.process_action.side_effect = lambda *a, **kw: seen.append(adapter._request_context.get()) + + await adapter._route_socket_event( + { + "type": "block_actions", + "team": {"id": "T_GRID_1"}, + "enterprise": {"id": "E_ORG_1"}, + "is_enterprise_install": True, + "actions": [{"type": "button", "action_id": "a", "value": "v"}], + "channel": {"id": "C123", "name": "test"}, + "message": {"ts": "1234567890.123456"}, + "user": {"id": "U_USER", "username": "testuser"}, + }, + "interactive", + AsyncMock(), + ) + + resolve.assert_awaited_once_with("E_ORG_1", True) + ctx = seen[0] + assert ctx is not None + assert (ctx.token, ctx.installation_id, ctx.team_id, ctx.enterprise_id) == ( + "xoxb-org", + "E_ORG_1", + "T_GRID_1", + "E_ORG_1", + ) + + +class TestHttpEnterpriseResolution: + async def test_slash_command_with_form_enterprise_flag_resolves_by_enterprise_id(self): + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation("E_ORG_1", SlackInstallation(bot_token="xoxb-org", is_enterprise_install=True)) + seen: list[RequestContext | None] = [] + chat.process_slash_command.side_effect = lambda *a, **kw: seen.append(adapter._request_context.get()) + body = ( + "command=%2Ftest&text=x&user_id=U1&channel_id=C1" + "&team_id=T_GRID_1&enterprise_id=E_ORG_1&is_enterprise_install=true" + ) + + response = await adapter.handle_webhook(_signed_request(body, "application/x-www-form-urlencoded")) + + assert response["status"] == 200 + ctx = seen[0] + assert ctx is not None + assert (ctx.token, ctx.installation_id, ctx.team_id, ctx.is_enterprise_install) == ( + "xoxb-org", + "E_ORG_1", + "T_GRID_1", + True, + ) + + async def test_interactive_payload_with_string_false_flag_resolves_by_team(self): + """``is_enterprise_install: "false"`` must not count as an org-wide + install (upstream ``=== true || === "true"``).""" + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation("T_GRID_1", SlackInstallation(bot_token="xoxb-team")) + resolve = _spy_resolver(adapter) + payload = { + "type": "block_actions", + "team": {"id": "T_GRID_1"}, + "enterprise": {"id": "E_ORG_1"}, + "is_enterprise_install": "false", + "actions": [{"type": "button", "action_id": "a", "value": "v"}], + "channel": {"id": "C123", "name": "test"}, + "message": {"ts": "1234567890.123456"}, + "user": {"id": "U_USER", "username": "testuser"}, + } + from urllib.parse import quote + + body = "payload=" + quote(json.dumps(payload)) + + response = await adapter.handle_webhook(_signed_request(body, "application/x-www-form-urlencoded")) + + assert response["status"] == 200 + resolve.assert_awaited_once_with("T_GRID_1", False) + assert chat.process_action.call_count == 1 + + +# --------------------------------------------------------------------------- +# withToken enterprise context injection +# --------------------------------------------------------------------------- + +_GENERIC_SLACK_RESPONSE: dict[str, Any] = { + "ok": True, + "ts": "2.2", + "message_ts": "2.2", + "messages": [], + "has_more": False, + "channel": {"id": "D1", "name": "x"}, + "user": {"id": "U1", "name": "u", "real_name": "U", "profile": {"display_name": "u"}}, + "view": {"id": "V1"}, +} + +_MODAL: dict[str, Any] = {"type": "modal", "callback_id": "cb", "title": "T", "children": []} + + +async def _react(adapter: SlackAdapter) -> None: + adapter._handle_reaction_event( + { + "type": "reaction_added", + "user": "U1", + "reaction": "thumbsup", + "item": {"type": "message", "channel": "C1", "ts": "2.2"}, + } + ) + await _settle() + + +async def _fetch_linked_message(adapter: SlackAdapter) -> None: + preview = adapter._create_link_preview("https://acme.slack.com/archives/C1/p1234567890123456") + assert preview.fetch_message is not None + with pytest.raises(RuntimeError, match="Message not found"): # the generic response holds no match + await preview.fetch_message() + + +# (client method, adapter call, whether the call targets the originating +# channel C1 and so must also echo ``client_context_team_id``). +_WITH_TOKEN_CALL_SITES: list[tuple[str, Callable[[SlackAdapter], Awaitable[Any]], bool]] = [ + ("chat_postMessage", lambda a: a.post_message("slack:C1:1.1", "hello"), True), + ("chat_postMessage", lambda a: a.post_message("slack:C1:1.1", Card(title="t")), True), + ("chat_update", lambda a: a.edit_message("slack:C1:1.1", "2.2", "edited"), True), + ("chat_update", lambda a: a.edit_message("slack:C1:1.1", "2.2", Card(title="t")), True), + ("chat_delete", lambda a: a.delete_message("slack:C1:1.1", "2.2"), True), + ("reactions_add", lambda a: a.add_reaction("slack:C1:1.1", "2.2", "thumbsup"), True), + ("reactions_remove", lambda a: a.remove_reaction("slack:C1:1.1", "2.2", "thumbsup"), True), + ("chat_postEphemeral", lambda a: a.post_ephemeral("slack:C1:1.1", "U1", "psst"), True), + ("chat_postEphemeral", lambda a: a.post_ephemeral("slack:C1:1.1", "U1", Card(title="t")), True), + ("conversations_replies", lambda a: a.fetch_messages("slack:C1:1.1"), True), + ("conversations_replies", lambda a: a.fetch_messages("slack:C1:1.1", FetchOptions(direction="forward")), True), + ("conversations_replies", lambda a: a.fetch_message("slack:C1:1.1", "2.2"), True), + ("conversations_history", lambda a: a.fetch_message("slack:C1:", "2.2"), True), + ("conversations_history", lambda a: a.fetch_channel_messages("slack:C1"), True), + ( + "conversations_history", + lambda a: a.fetch_channel_messages("slack:C1", FetchOptions(direction="forward")), + True, + ), + ("conversations_history", lambda a: a.list_threads("slack:C1"), True), + ("conversations_history", _fetch_linked_message, True), + ("conversations_replies", _react, True), + ("conversations_info", lambda a: a.fetch_thread("slack:C1:1.1"), True), + ("conversations_info", lambda a: a.fetch_channel_info("slack:C_OTHER"), False), + ("conversations_info", lambda a: a._lookup_channel("C_OTHER"), False), + ("users_info", lambda a: a._lookup_user("U1"), False), + ("conversations_open", lambda a: a.open_dm("U1"), False), + ("views_open", lambda a: a.open_modal("trigger-1", _MODAL), False), + ("views_update", lambda a: a.update_modal("V1", _MODAL), False), + ("views_publish", lambda a: a.publish_home_view("U1", {"type": "home", "blocks": []}), False), + # assistant.* methods take ``channel_id``, never ``channel`` (as upstream). + ("assistant_threads_setStatus", lambda a: a.start_typing("slack:C1:1.1"), False), + ("assistant_threads_setStatus", lambda a: a.set_assistant_status("C1", "1.1", "thinking"), False), + ("assistant_threads_setTitle", lambda a: a.set_assistant_title("C1", "1.1", "title"), False), + ("api_call", lambda a: a.set_suggested_prompts("C1", None, [{"title": "t", "message": "m"}]), False), +] + + +def _run_in_context(adapter: SlackAdapter, ctx: RequestContext, **kwargs: Any) -> dict[str, Any]: + tok = adapter._request_context.set(ctx) + try: + return adapter._with_token_kwargs(**kwargs) + finally: + adapter._request_context.reset(tok) + + +class TestWithTokenEnterpriseContextInjection: + @staticmethod + def _adapter() -> SlackAdapter: + return SlackAdapter(SlackAdapterConfig(signing_secret=_SECRET)) + + async def test_injects_team_id_on_org_wide_install_calls(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-org", is_enterprise_install=True, team_id="T_EVENT_1"), + channel="C1", + ) + + assert result == {"channel": "C1", "team_id": "T_EVENT_1"} + + async def test_does_not_inject_team_id_for_workspace_installs(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-team", is_enterprise_install=False, team_id="T_EVENT_1"), + channel="C1", + ) + + assert result == {"channel": "C1"} + + async def test_does_not_override_a_caller_specified_team_id(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-org", is_enterprise_install=True, team_id="T_EVENT_1"), + channel="C1", + team_id="T_EXPLICIT", + ) + + assert result["team_id"] == "T_EXPLICIT" + + async def test_echoes_context_team_id_as_client_context_team_id_on_calls_to_the_originating_channel(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-team", context_team_id="T_AWAY_HOST", context_channel="C1"), + channel="C1", + text="hi", + ) + + assert result["client_context_team_id"] == "T_AWAY_HOST" + + async def test_does_not_echo_client_context_team_id_to_a_different_channel(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-team", context_team_id="T_AWAY_HOST", context_channel="C1"), + channel="C_OTHER", + text="hi", + ) + + assert result == {"channel": "C_OTHER", "text": "hi"} + + async def test_does_not_add_client_context_team_id_to_non_channel_calls(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-team", context_team_id="T_AWAY_HOST", context_channel="C1"), + user="U1", + ) + + assert result == {"user": "U1"} + + async def test_captures_team_id_and_context_team_id_in_the_event_request_context(self): + adapter, _, _ = await _multi_workspace_adapter() + await adapter.set_installation("E_ORG_1", SlackInstallation(bot_token="xoxb-org", is_enterprise_install=True)) + + resolved = await adapter._resolve_event_request_context( + { + "type": "event_callback", + "team_id": "T_GRID_1", + "enterprise_id": "E_ORG_1", + "is_enterprise_install": True, + # context_team_id is a top-level envelope field, not inside `event`. + "context_team_id": "T_AWAY_HOST", + "event": {"type": "message", "channel": "C1", "ts": "1.1"}, + } + ) + + assert isinstance(resolved, RequestContext) + assert ( + resolved.installation_id, + resolved.is_enterprise_install, + resolved.team_id, + resolved.context_team_id, + resolved.context_channel, + ) == ("E_ORG_1", True, "T_GRID_1", "T_AWAY_HOST", "C1") + + async def test_no_request_context_leaves_kwargs_unchanged(self): + assert self._adapter()._with_token_kwargs(channel="C1", text="hi") == {"channel": "C1", "text": "hi"} + + async def test_a_caller_specified_client_context_team_id_is_kept(self): + result = _run_in_context( + self._adapter(), + RequestContext(token="xoxb-team", context_team_id="T_AWAY_HOST", context_channel="C1"), + channel="C1", + client_context_team_id="T_EXPLICIT", + ) + + assert result == {"channel": "C1", "client_context_team_id": "T_EXPLICIT"} + + @pytest.mark.parametrize( + ("client_method", "invoke", "to_originating_channel"), + _WITH_TOKEN_CALL_SITES, + ids=[case[0] + ":" + str(i) for i, case in enumerate(_WITH_TOKEN_CALL_SITES)], + ) + async def test_api_calls_carry_the_resolved_enterprise_context( + self, client_method: str, invoke: Callable[[SlackAdapter], Awaitable[Any]], to_originating_channel: bool + ): + """Python-specific wiring check, one case per ``_with_token_kwargs`` + call site (upstream ``withToken``). Upstream's ``withToken`` also + supplies the token, so a skipped site fails loudly there; here the + token is bound by ``_get_client`` and a skipped wrap would silently + drop ``team_id`` / ``client_context_team_id``.""" + adapter, _, _ = await _multi_workspace_adapter() + await adapter.set_installation("E_ORG_1", SlackInstallation(bot_token="xoxb-org", is_enterprise_install=True)) + resolved = await adapter._resolve_event_request_context( + { + "type": "event_callback", + "team_id": "T_GRID_1", + "enterprise_id": "E_ORG_1", + "is_enterprise_install": True, + "context_team_id": "T_AWAY_HOST", + "event": {"type": "message", "channel": "C1", "ts": "1.1"}, + } + ) + assert isinstance(resolved, RequestContext) + client = MagicMock() + mocked = AsyncMock(return_value=_GENERIC_SLACK_RESPONSE) + setattr(client, client_method, mocked) + adapter._get_client = lambda token=None: client # type: ignore[method-assign] + + tok = adapter._request_context.set(resolved) + try: + await invoke(adapter) + finally: + adapter._request_context.reset(tok) + + mocked.assert_awaited_once() + sent = mocked.await_args.kwargs + if client_method == "api_call": + # ``api_call`` path: the ``json`` body is what goes through withToken. + sent = sent["json"] + assert (sent.get("team_id"), sent.get("client_context_team_id")) == ( + "T_GRID_1", + "T_AWAY_HOST" if to_originating_channel else None, + ) + + +# --------------------------------------------------------------------------- +# event delivery deduplication +# --------------------------------------------------------------------------- + + +def _event_body(event_id: str) -> str: + return json.dumps( + { + "type": "event_callback", + "team_id": "T123", + "event_id": event_id, + "event": { + "type": "message", + "user": "U_USER", + "channel": "C123", + "text": "hello", + "ts": "1234567890.123456", + }, + } + ) + + +class TestEventDeliveryDeduplication: + async def test_drops_a_retried_delivery_of_an_already_dispatched_event(self): + adapter, chat, state = await _single_workspace_adapter() + + await adapter.handle_webhook(_signed_request(_event_body("Ev1"))) + # Fire-and-forget marker write + await _settle() + assert chat.process_message.call_count == 1 + assert await state.get("slack:event-delivered:Ev1") is True + + retry = _signed_request(_event_body("Ev1"), extra_headers={"x-slack-retry-num": "1"}) + response = await adapter.handle_webhook(retry) + + assert response["status"] == 200 + assert chat.process_message.call_count == 1 + + async def test_processes_a_retry_when_the_original_delivery_was_never_dispatched(self): + adapter, chat, _ = await _single_workspace_adapter() + + retry = _signed_request(_event_body("Ev_missed"), extra_headers={"x-slack-retry-num": "2"}) + response = await adapter.handle_webhook(retry) + + assert response["status"] == 200 + assert chat.process_message.call_count == 1 + + async def test_does_not_consult_state_on_first_deliveries(self): + adapter, chat, state = await _single_workspace_adapter() + get_spy = AsyncMock(side_effect=state.get) + state.get = get_spy # type: ignore[method-assign] + + await adapter.handle_webhook(_signed_request(_event_body("Ev2"))) + + assert chat.process_message.call_count == 1 + assert "slack:event-delivered:Ev2" not in [c.args[0] for c in get_spy.await_args_list] + + async def test_dedupes_retried_socket_deliveries_by_event_id(self): + adapter, chat, _ = await _single_workspace_adapter() + body = { + "team_id": "T123", + "event_id": "Ev_sock", + "event": dict(_MESSAGE_EVENT), + } + + await adapter._route_socket_event(body, "events_api", AsyncMock()) + await _settle() + assert chat.process_message.call_count == 1 + + await adapter._route_socket_event(body, "events_api", AsyncMock(), None, 1) + + assert chat.process_message.call_count == 1 + + async def test_marker_is_written_with_a_24_hour_ttl(self): + adapter, _, state = await _single_workspace_adapter() + set_spy = AsyncMock(side_effect=state.set) + state.set = set_spy # type: ignore[method-assign] + + await adapter.handle_webhook(_signed_request(_event_body("Ev_ttl"))) + await _settle() + + set_spy.assert_awaited_once_with("slack:event-delivered:Ev_ttl", True, 24 * 60 * 60 * 1000) + + async def test_retry_is_processed_when_the_state_read_fails(self): + adapter, chat, state = await _single_workspace_adapter() + await state.set("slack:event-delivered:Ev_err", True) + state.get = AsyncMock(side_effect=RuntimeError("state down")) # type: ignore[method-assign] + + retry = _signed_request(_event_body("Ev_err"), extra_headers={"x-slack-retry-num": "1"}) + response = await adapter.handle_webhook(retry) + + assert response["status"] == 200 + assert chat.process_message.call_count == 1 + + async def test_a_failed_marker_write_does_not_fail_the_webhook(self): + adapter, chat, state = await _single_workspace_adapter() + state.set = AsyncMock(side_effect=RuntimeError("state down")) # type: ignore[method-assign] + + response = await adapter.handle_webhook(_signed_request(_event_body("Ev_write"))) + await _settle() + + assert response["status"] == 200 + assert chat.process_message.call_count == 1 + + async def test_malformed_retry_header_counts_as_a_first_delivery(self): + adapter, chat, state = await _single_workspace_adapter() + await state.set("slack:event-delivered:Ev_bad", True) + + retry = _signed_request(_event_body("Ev_bad"), extra_headers={"x-slack-retry-num": "not-a-number"}) + await adapter.handle_webhook(retry) + + assert chat.process_message.call_count == 1 + + async def test_forwarded_socket_event_retry_num_drops_a_dispatched_event(self): + state = await _memory_state() + chat = _make_chat(state) + adapter = SlackAdapter( + SlackAdapterConfig( + mode="socket", + app_token="xapp-1-x", + bot_token="xoxb-test-token", + socket_forwarding_secret="fwd-secret", + ) + ) + adapter._chat = chat + await state.set("slack:event-delivered:Ev_fwd", True) + + forwarded = json.dumps( + { + "type": "socket_event", + "eventType": "events_api", + "body": {"team_id": "T123", "event_id": "Ev_fwd", "event": dict(_MESSAGE_EVENT)}, + "retryNum": 1, + "timestamp": int(time.time() * 1000), + } + ) + response = await adapter.handle_webhook(_FakeRequest(forwarded, {"x-slack-socket-token": "fwd-secret"})) + + assert response["status"] == 200 + chat.process_message.assert_not_called() + + +# --------------------------------------------------------------------------- +# W-prefixed enterprise user IDs +# --------------------------------------------------------------------------- + + +class TestWPrefixedEnterpriseUserIds: + async def test_treats_bare_at_w_mentions_as_raw_user_ids_not_display_names(self): + adapter, _, state = await _single_workspace_adapter() + get_list_spy = AsyncMock(side_effect=state.get_list) + state.get_list = get_list_spy # type: ignore[method-assign] + + result = await adapter._resolve_outgoing_mentions("Hey @W012345AB, ping", "slack:C1:1.1") + + # Left for the markdown layer to render as <@W012345AB>, with no + # reverse-index lookup attempted for "w012345ab" + assert result == "Hey @W012345AB, ping" + assert "slack:user-by-name:w012345ab" not in [c.args[0] for c in get_list_spy.await_args_list] + + async def test_resolves_incoming_at_w_mentions_like_u_prefixed_ones(self): + adapter, _, _ = await _single_workspace_adapter() + client = MagicMock() + client.users_info = AsyncMock( + return_value={ + "user": { + "name": "wanda", + "profile": {"display_name": "Wanda", "real_name": "Wanda Grid"}, + "real_name": "Wanda Grid", + } + } + ) + adapter._get_client = lambda token=None: client # type: ignore[method-assign] + + message = await adapter._parse_slack_message( + { + "type": "message", + "user": "W_SENDER_1", + "username": "sender", + "text": "hello <@W012345AB>", + "ts": "1234567890.123456", + "channel": "C123", + }, + "slack:C123:1234567890.123456", + ) + + assert "Wanda" in message.text + + +# --------------------------------------------------------------------------- +# event routing via authorizations[] +# --------------------------------------------------------------------------- + + +class TestEventRoutingViaAuthorizations: + async def test_prefers_authorizations_0_over_top_level_fields_for_org_installs(self): + adapter, _, _ = await _multi_workspace_adapter() + await adapter.set_installation("E_ORG_1", SlackInstallation(bot_token="xoxb-org", is_enterprise_install=True)) + + # Envelope where the org identity lives only in authorizations (the + # documented location); top-level omits is_enterprise_install. + resolved = await adapter._resolve_event_request_context( + { + "type": "event_callback", + "team_id": "T_GRID_1", + "event": dict(_MESSAGE_EVENT), + "authorizations": [{"enterprise_id": "E_ORG_1", "team_id": None, "is_enterprise_install": True}], + } + ) + + assert isinstance(resolved, RequestContext) + assert (resolved.installation_id, resolved.is_enterprise_install, resolved.token, resolved.team_id) == ( + "E_ORG_1", + True, + "xoxb-org", + "T_GRID_1", + ) + + async def test_uses_the_authorizations_team_over_a_slack_connect_top_level_team(self): + adapter, _, _ = await _multi_workspace_adapter() + await adapter.set_installation("T_RECIPIENT", SlackInstallation(bot_token="xoxb-recipient")) + + # Shared-channel envelope: top-level names the other org's workspace, + # authorizations[0] names the actual recipient installation + resolved = await adapter._resolve_event_request_context( + { + "type": "event_callback", + "team_id": "T_OTHER_ORG", + "enterprise_id": "E_OTHER_ORG", + "event": dict(_MESSAGE_EVENT), + "authorizations": [{"enterprise_id": None, "team_id": "T_RECIPIENT", "is_enterprise_install": False}], + } + ) + + assert isinstance(resolved, RequestContext) + assert (resolved.installation_id, resolved.is_enterprise_install, resolved.token) == ( + "T_RECIPIENT", + False, + "xoxb-recipient", + ) + + async def test_falls_back_to_top_level_fields_when_authorizations_is_absent(self): + adapter, _, _ = await _multi_workspace_adapter() + await adapter.set_installation("E_ORG_1", SlackInstallation(bot_token="xoxb-org", is_enterprise_install=True)) + + resolved = await adapter._resolve_event_request_context( + { + "type": "event_callback", + "team_id": "T_GRID_1", + "enterprise_id": "E_ORG_1", + "is_enterprise_install": True, + "event": dict(_MESSAGE_EVENT), + } + ) + + assert isinstance(resolved, RequestContext) + assert (resolved.installation_id, resolved.is_enterprise_install, resolved.token) == ( + "E_ORG_1", + True, + "xoxb-org", + ) + + async def test_resolution_states_for_single_workspace_missing_ids_and_unknown_installs(self): + single, _, _ = await _single_workspace_adapter() + multi, _, _ = await _multi_workspace_adapter() + event = {"type": "event_callback", "event": dict(_MESSAGE_EVENT)} + + assert await single._resolve_event_request_context({**event, "team_id": "T1"}) == "not-applicable" + assert await multi._resolve_event_request_context(event) == "not-applicable" + assert await multi._resolve_event_request_context({"type": "url_verification"}) == "not-applicable" + assert await multi._resolve_event_request_context({**event, "team_id": "T_UNKNOWN"}) == "unresolved" + + async def test_http_webhook_drops_events_for_unknown_authorizations(self): + adapter, chat, _ = await _multi_workspace_adapter() + await adapter.set_installation("T_OTHER_ORG", SlackInstallation(bot_token="xoxb-other")) + body = json.dumps( + { + "type": "event_callback", + "team_id": "T_OTHER_ORG", + "event": dict(_MESSAGE_EVENT), + "authorizations": [{"team_id": "T_NOT_INSTALLED", "is_enterprise_install": False}], + } + ) + + response = await adapter.handle_webhook(_signed_request(body)) + + assert response["status"] == 200 + chat.process_message.assert_not_called() diff --git a/tests/test_slack_socket_mode.py b/tests/test_slack_socket_mode.py index b88b3a0..89d10a4 100644 --- a/tests/test_slack_socket_mode.py +++ b/tests/test_slack_socket_mode.py @@ -872,7 +872,7 @@ async def capture_slash(*args: Any, **kwargs: Any) -> dict[str, Any]: for _ in range(5): await asyncio.sleep(0) - resolve.assert_awaited_once_with("T1") + resolve.assert_awaited_once_with("T1", False) assert captured == [("xoxb-team-1", "T1")] assert adapter._request_context.get() is None @@ -910,7 +910,7 @@ def capture_action(*args: Any, **kwargs: Any) -> None: ack, ) - resolve.assert_awaited_once_with("T1") + resolve.assert_awaited_once_with("T1", False) assert captured == [("xoxb-team-1", "T1")] assert adapter._request_context.get() is None @@ -940,7 +940,7 @@ async def test_concurrent_events_for_different_teams_do_not_cross_contaminate(se # Per-team token lookup. The first lookup awaits long enough for # the second to interleave; if isolation is broken the first will # observe the second's token. - async def fake_resolve(team_id: str) -> RequestContext: + async def fake_resolve(team_id: str, _is_enterprise_install: bool = False) -> RequestContext: if team_id == "T1": # Yield so the T2 dispatch can race in and set the # ContextVar before T1's process_message runs. @@ -1032,7 +1032,7 @@ async def test_drops_socket_interactive_payloads_with_no_installation(self): ack, ) - resolve.assert_awaited_once_with("T_UNKNOWN") + resolve.assert_awaited_once_with("T_UNKNOWN", False) chat.process_action.assert_not_called() ack.assert_awaited_once_with() @@ -1279,15 +1279,16 @@ async def test_dispatch_exception_acks_with_errors_response_action(self): class TestSocketEventsApiPayloadParity: - """Regression for review finding #7. + """The socket events_api envelope keeps upstream's fields. - What to fix if this fails: the synthesized ``event_callback`` payload - in the socket-mode events_api branch must match the webhook path. - Adding ``is_ext_shared_channel`` here is a quiet socket-vs-webhook - divergence — neither upstream nor the Python webhook path includes it. + What to fix if this fails: the ``event_callback`` payload rebuilt in the + socket-mode events_api branch must carry the fields upstream's + ``routeSocketEvent`` keeps since vercel/chat#724 (chat@4.35): the + installation identity, ``context_team_id`` and ``is_ext_shared_channel``, + with ``is_enterprise_install`` normalized to a bool. """ - async def test_synthesized_payload_does_not_include_is_ext_shared_channel(self): + async def test_synthesized_payload_keeps_enterprise_and_shared_channel_fields(self): adapter = _make_socket_adapter() adapter._chat = _make_mock_chat() @@ -1299,25 +1300,54 @@ def fake_process(payload: dict[str, Any], _options: Any = None) -> None: adapter._process_event_payload = fake_process # type: ignore[method-assign] ack = AsyncMock() + authorizations = [{"enterprise_id": "E1", "team_id": None, "is_enterprise_install": True}] + event = { + "type": "message", + "channel": "C1", + "ts": "1.0", + "user": "U1", + "text": "hi", + "team": "T1", + } body = { "team_id": "T1", + "enterprise_id": "E1", + "context_team_id": "T_AWAY", + "authorizations": authorizations, "event_id": "Ev1", "event_time": 1234, - "is_ext_shared_channel": True, # Should be dropped. - "event": { - "type": "message", - "channel": "C1", - "ts": "1.0", - "user": "U1", - "text": "hi", - "team": "T1", - }, + "is_ext_shared_channel": True, + "unrelated": "dropped", + "event": event, } await adapter._route_socket_event(body, "events_api", ack) - assert len(captured) == 1 - assert "is_ext_shared_channel" not in captured[0] - # Sanity: the keys we *do* synthesize are still present. - assert captured[0]["type"] == "event_callback" - assert captured[0]["team_id"] == "T1" - assert captured[0]["event_id"] == "Ev1" - assert captured[0]["event_time"] == 1234 + assert captured == [ + { + "type": "event_callback", + "event": event, + "authorizations": authorizations, + "team_id": "T1", + "context_team_id": "T_AWAY", + "enterprise_id": "E1", + "is_enterprise_install": False, + "is_ext_shared_channel": True, + "event_id": "Ev1", + "event_time": 1234, + } + ] + + async def test_socket_shared_channel_flag_marks_the_channel_external(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + + await adapter._route_socket_event( + { + "team_id": "T1", + "is_ext_shared_channel": True, + "event": {"type": "message", "channel": "C_SHARED", "ts": "1.0", "user": "U1", "text": "hi"}, + }, + "events_api", + AsyncMock(), + ) + + assert "C_SHARED" in adapter._external_channels diff --git a/tests/test_slack_webhook.py b/tests/test_slack_webhook.py index 865c883..b3d98e7 100644 --- a/tests/test_slack_webhook.py +++ b/tests/test_slack_webhook.py @@ -2517,8 +2517,8 @@ class TestInstallationScopedCaches: the resolved installation so data fetched with one workspace's token is never served to another. Single-workspace mode keeps unscoped keys. - The two ``withBotToken`` cases belong to #213 (they need - ``with_bot_token(..., installation_id=...)``). + The two ``withBotToken`` cases use ``with_bot_token_async(..., + installation_id=...)`` (#268). What to fix if this fails: ``_installation_cache_scope`` / ``_unfurl_cache_key`` in ``src/chat_sdk/adapters/slack/adapter.py`` @@ -2634,6 +2634,41 @@ async def test_empty_installation_id_uses_unscoped_keys(self): assert (await state.get("slack:user:U1"))["display_name"] == "Alice" assert await state.get("slack:user::U1") is None + @pytest.mark.asyncio + async def test_scopes_the_cache_under_with_bot_token_when_installation_id_is_passed(self): + # Upstream passes an async fn to ``withBotToken``; AsyncLocalStorage + # carries the store into the promise. A Python coroutine created in + # the sync ``with_bot_token`` would run after the context is reset, + # so the async variant is the faithful translation here. + adapter, state, client = await self._make_cache_adapter() + + await adapter.with_bot_token_async("xoxb-team-a", lambda: adapter._lookup_user("U1"), installation_id="T_A") + + assert client.users_info.await_count == 1 + assert (await state.get("slack:user:T_A:U1"))["display_name"] == "Alice" + assert await state.get("slack:user:U1") is None + + @pytest.mark.asyncio + async def test_uses_unscoped_keys_under_with_bot_token_without_installation_id(self): + adapter, state, _ = await self._make_cache_adapter() + + await adapter.with_bot_token_async("xoxb-token", lambda: adapter._lookup_user("U1")) + + assert (await state.get("slack:user:U1"))["display_name"] == "Alice" + + @pytest.mark.asyncio + async def test_sync_with_bot_token_scopes_by_installation_id(self): + adapter, _, _ = await self._make_cache_adapter() + + observed = adapter.with_bot_token( + "xoxb-team-a", + lambda: (adapter.current_token, adapter._installation_cache_scope()), + installation_id="T_A", + ) + + assert observed == ("xoxb-team-a", "T_A:") + assert adapter._request_context.get() is None + @pytest.mark.asyncio async def test_scopes_the_display_name_reverse_index_by_installation(self): adapter, state, _ = await self._make_cache_adapter()