From fce178ca330cab5c75e021a4cea68bc8b73c8e89 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 8 May 2026 03:00:41 +0000 Subject: [PATCH 1/9] feat(slack): Socket Mode transport (vercel/chat#162) Add an opt-in Socket Mode transport to ``SlackAdapter`` so bots can consume Slack events over a persistent WebSocket instead of webhooks. Mirrors the upstream JS port at vercel/chat#162. Highlights: * New ``SlackAdapterConfig`` fields: ``mode`` (``"webhook"`` default or ``"socket"``), ``app_token`` (xapp-* with prefix validation), and ``socket_forwarding_secret`` for the forwarded-event HTTP path. * ``signing_secret`` is now optional in socket mode (Slack does not sign socket events). ``_verify_signature`` refuses when no secret is configured so a stray webhook in socket mode can't HMAC against ``""``. * ``initialize()`` opens the WebSocket when ``mode == "socket"``. ``disconnect()`` cancels a tracked reconnect loop and closes the client; both ``start_socket_mode()`` and ``stop_socket_mode()`` are idempotent. * ``_route_socket_event`` dispatches ``events_api`` / ``slash_commands`` / ``interactive`` payloads into the same handlers the webhook path uses (no fork). Skips Slack retries (``retry_attempt > 0``) but still acks. Interactive view-submission errors round-trip through the ack payload. * ``handle_webhook`` now accepts forwarded socket events at any time via ``x-slack-socket-token`` (constant-time compared against the configured secret); refuses direct webhook POSTs in socket mode. * New ``ModalResponse(action="clear")`` produces ``response_action: clear`` for closing the entire modal view stack. * ``slack-socket`` extra in ``pyproject.toml`` (``slack-sdk`` + ``aiohttp``); the import inside ``start_socket_mode`` stays lazy. Hazard coverage: explicit ``asyncio.Task`` tracking and shutdown signaling for the WebSocket loop (#5); ``contextvars.copy_context()`` preserved through the socket dispatch path so per-event token resolution still inherits into spawned handlers (#6); single-process SocketModeClient lifecycle (#11); ``app_token`` format validated, never logged (#12). Tests: 24 new unit tests covering config validation, modal-clear emission, forwarded-event auth (accept / reject / no-secret-configured / invalid-token), the full ``_route_socket_event`` matrix (events_api / slash / interactive / retry / unknown), lifecycle (start, stop, idempotency, first-connect failure surfaced, transient-disconnect reconnect), and the ContextVar boundary on multi-workspace token resolution. The new file stubs ``slack_sdk.socket_mode.*`` in ``sys.modules`` to match the existing ``test_slack_client_cache.py`` pattern. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- docs/UPSTREAM_SYNC.md | 2 + pyproject.toml | 4 + src/chat_sdk/adapters/slack/adapter.py | 522 +++++++++++++++++++- src/chat_sdk/adapters/slack/types.py | 21 +- src/chat_sdk/types.py | 13 +- tests/test_slack_socket_mode.py | 648 +++++++++++++++++++++++++ 6 files changed, 1200 insertions(+), 10 deletions(-) create mode 100644 tests/test_slack_socket_mode.py diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index a9057930..5c316d36 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -494,6 +494,8 @@ stay explicit instead of being rediscovered in code review. | `StreamingPlan.is_supported()` / `get_fallback_text()` | Raise `RuntimeError` to fail loudly if a generic posting path (e.g. `ChannelImpl.post`, `post_postable_object`) tries to consume a `StreamingPlan` as a normal `PostableObject` | Silently return `True` / `""` — `ChannelImpl.post` would route through `postPostableObject` and post an empty-string fallback | Prevents `StreamingPlan` being silently routed through non-stream-aware posting paths where upstream would post a blank message or attempt a wrong-shape `adapter.post_object("stream", ...)` call. Internal dispatch is guarded by the `kind == "stream"` short-circuit in `post_postable_object` / `Thread.post`; this also protects third-party code that duck-types PostableObjects. | | `rehydrate_attachment` URL allowlist (Slack / Teams / Google Chat) | Validates the downloaded URL's scheme + host against a per-adapter allowlist inside the fetch closure; raises `ValidationError` on untrusted hosts before forwarding bearer tokens | No validation — `fetchData` blindly GETs `fetchMetadata.url` and forwards the workspace/bot token | SSRF + token-exfil risk upstream: after the 4.26 `rehydrateAttachment` hook lands, a crafted `fetchMetadata` in persisted state can redirect auth'd downloads to an arbitrary host. Python port enforces `CLAUDE.md`'s "Validate external URLs before requests (SSRF)" rule. Allowlist: Slack = `{files.slack.com, slack.com, *.slack.com, *.slack-edge.com}`; Teams = `{smba.trafficmanager.net, graph.microsoft.com, attachments.office.net, *.botframework.com, *.graph.microsoft.com, *.sharepoint.com, *.officeapps.live.com, *.office.com, *.office365.com, *.onedrive.com, *.microsoft.com}`; Google Chat = `{chat.googleapis.com, googleapis.com, *.googleapis.com, *.googleusercontent.com, *.google.com}`. | | `_rehydrate_message` with `Message` input | Falls through to the `rehydrate_attachment` pass even when the dequeued entry is already a `Message` instance | Early-returns on `raw instanceof Message` before rehydration | The Python port's Redis + Postgres `dequeue()` upgrade raw JSON to `Message.from_json(...)` before returning (upstream's dequeue returns the raw JSON.parse'd dict). Upstream's `instanceof Message` shortcut therefore only fires for in-memory state, but ours would fire for persistent backends too, leaving `fetch_data` stripped forever. The rehydrate pass still skips any attachment that already has `fetch_data`, so in-memory callers pay no cost. | +| Slack Socket Mode reconnect loop | Outer reconnect loop on top of `slack_sdk.socket_mode.aiohttp.SocketModeClient` (which itself has `auto_reconnect_enabled=True`). Exponential backoff (1s → 30s) with explicit shutdown signaling and a tracked `asyncio.Task` so `disconnect()` can cancel cleanly | Single `SocketModeClient` instance from `@slack/socket-mode`; relies entirely on the package's internal reconnect | Hazard #5 (async task lifecycle): a long-lived WebSocket needs an explicit shutdown path so `disconnect()` doesn't leak the loop, and a guarded outer reconnect path so the adapter survives `connect()` itself raising (which the inner client doesn't retry). Inner auto-reconnect still runs; the outer loop is belt-and-suspenders, not a divergence in observable behavior. | +| Slack Socket Mode listener serverless variant | Not ported | `startSocketModeListener()` / `runSocketModeListener()` open a transient socket for `durationMs` and forward events via HTTP POST | Vercel-specific pattern (cron-triggered ephemeral listener with `waitUntil`). The forwarded-event receiver (`x-slack-socket-token` handling in `handle_webhook`) is ported so a separate Python process can run the long-lived listener; the deployment glue itself isn't part of the SDK. | ### Platform-specific gaps diff --git a/pyproject.toml b/pyproject.toml index 97fb917d..d6b02a7a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,10 @@ Issues = "https://github.com/Chinchill-AI/chat-sdk-python/issues" [project.optional-dependencies] slack = ["slack-sdk>=3.27.0"] +# Slack Socket Mode (xapp-* WebSocket transport). slack_sdk's +# SocketModeClient ships with the slack-sdk wheel, but the aiohttp variant +# we use needs aiohttp at runtime. +slack-socket = ["slack-sdk>=3.27.0", "aiohttp>=3.9"] github = ["pyjwt[crypto]>=2.8"] redis = ["redis>=5.0"] postgres = ["asyncpg>=0.29"] diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 7fc74cf8..5afe05f6 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -10,6 +10,7 @@ import asyncio import base64 +import contextlib import contextvars import hashlib import hmac @@ -47,6 +48,7 @@ from chat_sdk.adapters.slack.types import ( RequestContext, SlackAdapterConfig, + SlackAdapterMode, SlackInstallation, SlackThreadId, ) @@ -190,21 +192,63 @@ def __init__(self, config: SlackAdapterConfig | None = None) -> None: if config is None: config = SlackAdapterConfig() + mode = config.mode or "webhook" signing_secret = config.signing_secret or os.environ.get("SLACK_SIGNING_SECRET") - if not signing_secret: + if not signing_secret and mode == "webhook": raise ValidationError( "slack", - "signingSecret is required. Set SLACK_SIGNING_SECRET or provide it in config.", + "signingSecret is required for webhook mode. Set SLACK_SIGNING_SECRET or provide it in config.", ) + app_token = config.app_token or os.environ.get("SLACK_APP_TOKEN") + if mode == "socket": + if not app_token: + raise ValidationError( + "slack", + "appToken is required for socket mode. Set SLACK_APP_TOKEN or provide it in config.", + ) + # Hazard #12: validate the long-lived secret format on init so a + # typo'd bot token (xoxb-) doesn't get silently used as an app + # token. Slack app-level tokens always start with ``xapp-``. + if not app_token.startswith("xapp-"): + raise ValidationError( + "slack", + "appToken must start with 'xapp-' (Slack app-level token). " + "Bot tokens (xoxb-) are not valid for socket mode.", + ) + # Auth fields: botToken presence selects single-workspace mode. - zero_config = not (config.signing_secret or config.bot_token or config.client_id or config.client_secret) + zero_config = not ( + config.signing_secret or config.bot_token or config.client_id or config.client_secret or config.app_token + ) bot_token = config.bot_token or (os.environ.get("SLACK_BOT_TOKEN") if zero_config else None) self._name = "slack" - self._signing_secret = signing_secret + self._signing_secret: str | None = signing_secret self._default_bot_token: str | None = bot_token + + # Socket mode state + self._mode: SlackAdapterMode = mode + self._app_token: str | None = app_token + self._socket_forwarding_secret: str | None = ( + config.socket_forwarding_secret or os.environ.get("SLACK_SOCKET_FORWARDING_SECRET") or app_token + ) + # The active SocketModeClient instance (when running in socket mode). + # Typed as ``Any`` because slack_sdk is an optional dependency. + self._socket_client: Any = None + # Background task that runs the connect/run/reconnect loop. Tracked so + # ``disconnect()`` can cancel it cleanly (hazard #5). + self._socket_task: asyncio.Task[None] | None = None + # Set when shutdown is requested so the reconnect loop knows to exit + # rather than retry on a clean disconnect. + self._socket_shutdown = False + # Default backoff schedule in seconds. Kept short so tests run fast, + # but capped low enough that a flapping Slack connection doesn't busy + # loop. Slack's recommended pattern is exponential backoff with jitter; + # our minimal schedule mirrors that behavior with explicit caps. + self._socket_initial_backoff_s = 1.0 + self._socket_max_backoff_s = 30.0 self._logger: Logger = config.logger or ConsoleLogger("info") self._user_name: str = config.user_name or "bot" self._bot_user_id: str | None = config.bot_user_id or None @@ -260,6 +304,16 @@ def lock_scope(self) -> LockScope: def persist_message_history(self) -> bool: return self._persist_message_history + @property + def mode(self) -> SlackAdapterMode: + """Connection mode (``"webhook"`` or ``"socket"``).""" + return self._mode + + @property + def is_socket_mode(self) -> bool: + """``True`` when the adapter is configured for Socket Mode.""" + return self._mode == "socket" + # ------------------------------------------------------------------ # Public request-context accessors # @@ -377,8 +431,18 @@ async def initialize(self, chat: ChatInstance) -> None: if not self._default_bot_token: self._logger.info("Slack adapter initialized in multi-workspace mode") + if self._mode == "socket": + await self.start_socket_mode() + async def disconnect(self) -> None: - """No persistent connections to close.""" + """Close any persistent connections held by the adapter. + + In webhook mode this is a no-op. In socket mode it cancels the + background reconnect loop, closes the active ``SocketModeClient``, + and waits for the loop to settle. Idempotent — calling it twice or + before ``initialize()`` is safe. + """ + await self.stop_socket_mode() # ================================================================== # Multi-workspace installation management @@ -722,6 +786,31 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No # Extract headers headers = getattr(request, "headers", {}) + + # Forwarded socket-mode events bypass Slack signature verification — + # they're authenticated by a shared bearer secret instead. This lets a + # separate process run the WebSocket and POST events back to the + # webhook endpoint over HTTP. Hazard #12: refuse if no secret is + # configured rather than treating an empty header match as success. + socket_token = headers.get("x-slack-socket-token") or headers.get("X-Slack-Socket-Token") + if socket_token: + if not self._socket_forwarding_secret or not hmac.compare_digest( + socket_token, self._socket_forwarding_secret + ): + self._logger.warn("Invalid socket forwarding token") + return {"body": "Invalid socket token", "status": 401} + try: + event = json.loads(body) + except (json.JSONDecodeError, ValueError): + return {"body": "Invalid JSON", "status": 400} + await self._handle_forwarded_socket_event(event, options) + return {"body": "ok", "status": 200} + + # In socket mode, refuse direct webhook POSTs — Slack delivers events + # over the WebSocket instead. We still allow forwarded events above. + if self._mode == "socket": + return {"body": "Webhooks are disabled in socket mode", "status": 405} + timestamp = headers.get("x-slack-request-timestamp") or headers.get("X-Slack-Request-Timestamp") signature = headers.get("x-slack-signature") or headers.get("X-Slack-Signature") @@ -802,7 +891,11 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No # ================================================================== def _verify_signature(self, body: str, timestamp: str | None, signature: str | None) -> bool: - if not (timestamp and signature): + # Refuse rather than HMAC against an empty key. This matters in socket + # mode where ``signing_secret`` is optional — without this guard a + # caller could call ``handle_webhook`` while in socket mode and + # silently pass verification with an empty secret. + if not (timestamp and signature and self._signing_secret): return False # Check timestamp is recent (within 5 minutes) @@ -881,6 +974,18 @@ async def _handle_interactive_payload(self, body: str, options: WebhookOptions | except (json.JSONDecodeError, ValueError): return {"body": "Invalid payload JSON", "status": 400} + return await self._dispatch_interactive_payload(payload, options) + + async def _dispatch_interactive_payload( + self, + payload: dict[str, Any], + options: WebhookOptions | None = None, + ) -> dict[str, Any]: + """Dispatch a pre-parsed interactive payload to the right handler. + + Used by both the webhook path (after form-decoding) and the socket + mode path (which receives the payload as a JSON object directly). + """ payload_type = payload.get("type") if payload_type == "block_actions": @@ -1198,6 +1303,9 @@ def _handle_view_closed(self, payload: dict[str, Any], options: WebhookOptions | def _modal_response_to_slack(self, response: ModalResponse, context_id: str | None = None) -> SlackModalResponse: if response.action == "close": return {} + if response.action == "clear": + # Close the entire modal view stack (Slack ``response_action: clear``). + return {"response_action": "clear"} if response.action == "errors": return {"response_action": "errors", "errors": response.errors or {}} if response.action in ("update", "push"): @@ -1213,6 +1321,395 @@ def _modal_response_to_slack(self, response: ModalResponse, context_id: str | No return {"response_action": response.action, "view": view} return {} + # ================================================================== + # Socket Mode + # ================================================================== + + async def start_socket_mode(self) -> None: + """Open a Slack Socket Mode WebSocket and dispatch events. + + Spawns a tracked background task that connects, runs the message + loop, and reconnects with exponential backoff on disconnect (per + Slack's recommendation). Returns once the initial connection has + been established. + + Raises :class:`ValidationError` if the adapter wasn't configured + with ``app_token`` (must start with ``xapp-``). + + Idempotent: a second call while connected is a no-op. + """ + if not self._app_token: + raise ValidationError( + "slack", + "appToken is required for socket mode. Set SLACK_APP_TOKEN or provide it in config.", + ) + + if self._socket_task is not None and not self._socket_task.done(): + # Already running. + return + + # Lazy import (hazard #10) — slack_sdk is an optional dependency. + try: + from slack_sdk.socket_mode.aiohttp import SocketModeClient # noqa: F401 + except ImportError as exc: # pragma: no cover - import-time failure + raise ValidationError( + "slack", + "slack_sdk is not installed. Install with `pip install chat-sdk[slack]`.", + ) from exc + + self._socket_shutdown = False + connected = asyncio.Event() + loop = asyncio.get_running_loop() + # Hazard #5: track the task explicitly so ``stop_socket_mode`` can + # cancel it cleanly. Don't use ``asyncio.ensure_future`` without + # tracking — a stray reference loss would orphan the WebSocket. + self._socket_task = loop.create_task(self._socket_mode_loop(connected)) + + # Wait for either the first successful connect or for the loop to + # exit (which means the very first connect raised). Re-raise so the + # caller learns about a hard config failure (bad app token, network + # offline) instead of silently spinning forever. + first_done, _ = await asyncio.wait( + {asyncio.create_task(connected.wait()), self._socket_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + if connected.is_set(): + return + # Socket loop exited before connecting — surface its exception. + for done in first_done: + if done is self._socket_task: + exc = done.exception() + if exc is not None: + raise exc + + async def stop_socket_mode(self) -> None: + """Close the Socket Mode connection and cancel the reconnect loop. + + Idempotent. Safe to call from any task; it disconnects the active + client and waits for the background task to finish. + """ + self._socket_shutdown = True + + client = self._socket_client + self._socket_client = None + if client is not None: + try: + await client.disconnect() + except Exception as exc: # pragma: no cover - best-effort cleanup + self._logger.warn("Error disconnecting Slack socket client", {"error": str(exc)}) + + task = self._socket_task + self._socket_task = None + if task is not None and not task.done(): + task.cancel() + # Cancellation is expected; any other exception was already + # logged inside the loop. Don't re-raise on shutdown. + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + if task is not None: + self._logger.info("Slack socket mode disconnected") + + async def _socket_mode_loop(self, connected: asyncio.Event) -> None: + """Connect/run/reconnect loop for Socket Mode. + + Slack's Socket Mode WebSocket is long-lived but can disconnect for + many reasons (refresh, network blip, restart). We retry with + exponential backoff (with jitter) and reset the backoff once a + connection holds for any non-trivial time. + """ + from slack_sdk.socket_mode.aiohttp import SocketModeClient + + backoff = self._socket_initial_backoff_s + try: + while not self._socket_shutdown: + client = SocketModeClient(app_token=cast(str, self._app_token)) + # Register our request handler. ``socket_mode_request_listeners`` + # is the documented public extension point on the slack_sdk + # client; each listener is ``async (client, request) -> None``. + client.socket_mode_request_listeners.append(self._on_socket_request) + self._socket_client = client + try: + await client.connect() + except Exception as exc: + self._logger.error( + "Slack socket mode connect failed", + {"error": str(exc)}, + ) + self._socket_client = None + if self._socket_shutdown: + return + if not connected.is_set(): + # First connect failed and nobody's listening yet — + # surface the error to the caller of start_socket_mode. + raise + await self._socket_sleep_with_backoff(backoff) + backoff = min(backoff * 2, self._socket_max_backoff_s) + continue + + # Connection established (or in progress) — let the caller of + # start_socket_mode resume. + self._logger.info("Slack socket mode connected") + connected.set() + backoff = self._socket_initial_backoff_s + + # Wait until the socket disconnects or shutdown is requested. + while not self._socket_shutdown: + if not client.is_connected(): + break + await asyncio.sleep(1.0) + + # Tear down the current client before reconnecting. + self._socket_client = None + try: + await client.disconnect() + except Exception as exc: # pragma: no cover - best-effort + self._logger.warn( + "Error disconnecting Slack socket client during reconnect", + {"error": str(exc)}, + ) + + if self._socket_shutdown: + return + self._logger.info("Slack socket mode disconnected, reconnecting") + await self._socket_sleep_with_backoff(backoff) + backoff = min(backoff * 2, self._socket_max_backoff_s) + except asyncio.CancelledError: + raise + except Exception as exc: + # Make sure first-connect failures propagate to the caller of + # start_socket_mode, but also log everything else loudly. + if not connected.is_set(): + raise + self._logger.error( + "Slack socket mode loop crashed", + {"error": str(exc)}, + ) + finally: + client = self._socket_client + self._socket_client = None + if client is not None: + with contextlib.suppress(Exception): # pragma: no cover - best-effort + await client.disconnect() + + async def _socket_sleep_with_backoff(self, seconds: float) -> None: + """Sleep for ``seconds`` but wake immediately on shutdown.""" + # Poll every 0.25s so a stop_socket_mode call doesn't have to wait + # the full backoff window. asyncio.Event would be slightly cleaner + # but we don't want to add an Event per loop iteration. + deadline = asyncio.get_event_loop().time() + seconds + while not self._socket_shutdown: + remaining = deadline - asyncio.get_event_loop().time() + if remaining <= 0: + return + await asyncio.sleep(min(0.25, remaining)) + + async def _on_socket_request(self, client: Any, request: Any) -> None: + """Listener invoked by ``SocketModeClient`` for each socket message. + + Signature matches slack_sdk's documented hook: + ``async (client: SocketModeClient, request: SocketModeRequest) -> None``. + """ + from slack_sdk.socket_mode.response import SocketModeResponse + + envelope_id = getattr(request, "envelope_id", "") or "" + event_type = getattr(request, "type", "") or "" + payload = getattr(request, "payload", None) or {} + retry_attempt = getattr(request, "retry_attempt", 0) or 0 + + async def ack(response_payload: dict[str, Any] | None = None) -> None: + try: + await client.send_socket_mode_response( + SocketModeResponse(envelope_id=envelope_id, payload=response_payload) + ) + except Exception as exc: # pragma: no cover - best-effort + self._logger.warn( + "Failed to send socket mode ack", + {"envelope_id": envelope_id, "error": str(exc)}, + ) + + # Slack re-delivers events that weren't acked in time. Skip retries + # so we don't double-process — but still ack so Slack stops resending. + if retry_attempt and retry_attempt > 0: + await ack() + self._logger.debug("Skipping socket mode retry", {"retry_attempt": retry_attempt}) + return + + await self._route_socket_event(payload, event_type, ack) + + async def _route_socket_event( + self, + body: dict[str, Any], + event_type: str, + ack: Callable[..., Awaitable[None]], + options: WebhookOptions | None = None, + ) -> None: + """Route a socket-mode event to the same handler the webhook path uses. + + Mirrors upstream's ``routeSocketEvent``. The ``ack`` callback delivers + 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. + """ + + def wrap_async(coro: Awaitable[Any]) -> None: + """Run ``coro`` either via ``waitUntil`` or as a tracked task.""" + if options is not None and options.wait_until is not None: + # ``wait_until`` semantics: caller takes ownership. + options.wait_until(cast(Any, coro)) + return + task = asyncio.get_event_loop().create_task(cast(Any, coro)) + + def _log_exc(t: asyncio.Task[Any]) -> None: + if t.cancelled(): + return + exc = t.exception() + if exc is not None: + self._logger.error( + "Error in socket mode async handler", + {"error": str(exc)}, + ) + + task.add_done_callback(_log_exc) + _pin_task(task) + + if event_type == "events_api": + await ack() + event = body.get("event") + if not isinstance(event, dict): + self._logger.warn( + "Socket mode events_api missing event field", + {"body_type": type(body).__name__}, + ) + return + payload: dict[str, Any] = { + "type": "event_callback", + "event": event, + "team_id": body.get("team_id"), + "event_id": body.get("event_id"), + "event_time": body.get("event_time"), + "is_ext_shared_channel": body.get("is_ext_shared_channel"), + } + # 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._default_bot_token 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 + isolated = contextvars.copy_context() + isolated.run(self._request_context.set, ctx) + isolated.run(self._process_event_payload, payload, options) + else: + self._process_event_payload(payload, options) + except Exception as exc: + self._logger.error( + "Error processing socket mode events_api", + {"error": str(exc)}, + ) + return + + if event_type == "slash_commands": + await ack() + # 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: + team_id_slash = (params.get("team_id") or [None])[0] + if not self._default_bot_token and team_id_slash: + ctx = await self._resolve_token_for_team(team_id_slash) + if ctx is None: + self._logger.warn("Could not resolve token for slash command") + return + tok = self._request_context.set(ctx) + try: + await self._handle_slash_command(params, options) + finally: + self._request_context.reset(tok) + else: + await self._handle_slash_command(params, options) + + wrap_async(run_slash()) + return + + if event_type == "interactive": + try: + # Multi-workspace: scope token resolution to the dispatch. + team_ref = body.get("team") + team_id_interactive = team_ref.get("id") if isinstance(team_ref, dict) else body.get("team_id") + if not self._default_bot_token and team_id_interactive: + ctx = await self._resolve_token_for_team(team_id_interactive) + if ctx is None: + self._logger.warn("Could not resolve token for interactive payload") + await ack() + return + tok = self._request_context.set(ctx) + try: + result = await self._dispatch_interactive_payload(body, options) + finally: + self._request_context.reset(tok) + else: + result = await self._dispatch_interactive_payload(body, options) + except Exception as exc: + self._logger.error( + "Error processing socket mode interactive", + {"error": str(exc)}, + ) + await ack() + return + + response_body: dict[str, Any] | None = None + body_str = result.get("body") if isinstance(result, dict) else None + if isinstance(body_str, str) and body_str: + content_type = result.get("headers", {}).get("Content-Type", "") if isinstance(result, dict) else "" + if "application/json" in content_type: + try: + parsed = json.loads(body_str) + if isinstance(parsed, dict): + response_body = parsed + except (json.JSONDecodeError, ValueError): + response_body = None + await ack(response_body) + return + + # Unknown event type — still ack so Slack doesn't redeliver. + await ack() + self._logger.debug("Unhandled socket mode event type", {"type": event_type}) + + async def _handle_forwarded_socket_event( + self, + event: dict[str, Any], + options: WebhookOptions | None = None, + ) -> None: + """Process a socket-mode event forwarded over HTTP. + + Companion to :meth:`_route_socket_event` for the serverless pattern + where a long-running listener runs in one process and posts events + to a webhook handler elsewhere. The ack already happened on the + listener side; we just route to the same handler dispatch. + """ + + async def noop_ack(_response: dict[str, Any] | None = None) -> None: + return None + + body = event.get("body") + event_type = event.get("eventType") or event.get("event_type") or "" + if not isinstance(body, dict) or not isinstance(event_type, str): + self._logger.warn( + "Forwarded socket event has invalid shape", + {"event_type": type(event_type).__name__}, + ) + return + await self._route_socket_event(body, event_type, noop_ack, options) + # ================================================================== # Message events # ================================================================== @@ -3167,5 +3664,16 @@ async def _send_to_response_url( def create_slack_adapter(config: SlackAdapterConfig | None = None) -> SlackAdapter: - """Create a new SlackAdapter instance.""" + """Create a new SlackAdapter instance. + + For socket mode, the factory rejects multi-workspace setups upfront — + Socket Mode is a single-workspace transport (the WebSocket carries one + app's events for one workspace) and silently mixing the two would mask + a config mistake. + """ + if config is not None and (config.mode or "webhook") == "socket" and (config.client_id or config.client_secret): + raise ValidationError( + "slack", + "Multi-workspace (clientId/clientSecret) is not supported in socket mode.", + ) return SlackAdapter(config) diff --git a/src/chat_sdk/adapters/slack/types.py b/src/chat_sdk/adapters/slack/types.py index a085b80b..13f33921 100644 --- a/src/chat_sdk/adapters/slack/types.py +++ b/src/chat_sdk/adapters/slack/types.py @@ -3,10 +3,15 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, TypedDict +from typing import Any, Literal, TypedDict from chat_sdk.logger import Logger +# Connection mode for the Slack adapter. ``"webhook"`` (default) consumes +# events via signed HTTP POSTs from Slack. ``"socket"`` opens a long-lived +# WebSocket via Slack's Socket Mode and ACKs each event over the socket. +SlackAdapterMode = Literal["webhook", "socket"] + # ============================================================================= # Configuration # ============================================================================= @@ -16,6 +21,8 @@ class SlackAdapterConfig: """Configuration for the Slack adapter.""" + # App-level token (xapp-...). Required when ``mode == "socket"``. + app_token: str | None = None # Bot token (xoxb-...). Required for single-workspace mode. Omit for multi-workspace. bot_token: str | None = None # Bot user ID (will be fetched if not provided) @@ -32,8 +39,20 @@ class SlackAdapterConfig: installation_key_prefix: str = "slack:installation" # Logger instance for error reporting. Defaults to ConsoleLogger. logger: Logger | None = None + # Connection mode: ``"webhook"`` (default) or ``"socket"``. When set to + # ``"socket"`` the adapter opens a Slack Socket Mode WebSocket on + # ``initialize()`` and dispatches events over it. ``signing_secret`` is + # not required in socket mode (Slack does not sign socket events). + mode: SlackAdapterMode = "webhook" # Signing secret for webhook verification. Defaults to SLACK_SIGNING_SECRET env var. + # Required in webhook mode; optional in socket mode. signing_secret: str | None = None + # Shared secret for authenticating events forwarded from a separate + # socket-mode listener via HTTP POST. Auto-detected from + # SLACK_SOCKET_FORWARDING_SECRET. Falls back to ``app_token`` if not set + # (matches upstream behavior; prefer setting this explicitly so the + # long-lived xapp- token isn't used as a bearer credential). + socket_forwarding_secret: str | None = None # Maximum number of cached AsyncWebClient instances (LRU-bounded). # Defaults to 100. Increase for large multi-workspace deployments. client_cache_max: int | None = None diff --git a/src/chat_sdk/types.py b/src/chat_sdk/types.py index 6cdfe297..89763d86 100644 --- a/src/chat_sdk/types.py +++ b/src/chat_sdk/types.py @@ -1074,9 +1074,18 @@ class ModalCloseEvent: @dataclass class ModalResponse: - """Response to a modal submit event.""" + """Response to a modal submit event. - action: Literal["close", "update", "push", "errors"] + The ``action`` field selects which Slack ``response_action`` is sent: + + * ``"close"`` — close the current view (no body) + * ``"clear"`` — close the entire view stack + * ``"update"`` — replace the current view with ``modal`` + * ``"push"`` — push ``modal`` onto the view stack + * ``"errors"`` — show field-level errors (``errors`` dict) + """ + + action: Literal["close", "clear", "update", "push", "errors"] modal: Any = None errors: dict[str, str] | None = None diff --git a/tests/test_slack_socket_mode.py b/tests/test_slack_socket_mode.py new file mode 100644 index 00000000..85d4c4d4 --- /dev/null +++ b/tests/test_slack_socket_mode.py @@ -0,0 +1,648 @@ +"""Tests for Slack adapter Socket Mode support (vercel/chat#162 port). + +Covers: + +* Configuration validation (app_token required, xapp- prefix, signing + optional in socket mode, multi-workspace rejected by factory). +* ``ModalResponse(action="clear")`` produces ``response_action: clear``. +* ``handle_webhook`` accepts forwarded socket events with a valid + ``x-slack-socket-token`` header and rejects mismatches. +* ``handle_webhook`` returns 405 for direct POSTs in socket mode. +* ``_route_socket_event`` dispatches events_api / slash_commands / + interactive payloads to the same handlers the webhook path uses, calls + ``ack`` exactly once, and skips Slack retries. +* ``start_socket_mode`` / ``stop_socket_mode`` are idempotent and the + reconnect loop reconnects after a transient disconnect. +* ContextVar boundaries: events received over the socket inherit the + per-instance request-context ContextVar. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +from types import ModuleType +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +# --------------------------------------------------------------------------- +# Stub slack_sdk.socket_mode.* before importing the adapter. Other Slack +# test modules already stub ``slack_sdk.web`` via ``setdefault``; we extend +# the same pattern to the socket mode submodules so the adapter's lazy +# ``from slack_sdk.socket_mode.aiohttp import SocketModeClient`` resolves +# regardless of whether the real ``slack-sdk`` extra is installed. +# --------------------------------------------------------------------------- + + +def _ensure_socket_mode_stub() -> None: + if "slack_sdk.socket_mode.aiohttp" in sys.modules and hasattr( + sys.modules["slack_sdk.socket_mode.aiohttp"], "SocketModeClient" + ): + return + + sys.modules.setdefault("slack_sdk", ModuleType("slack_sdk")) + sm_root = sys.modules.setdefault("slack_sdk.socket_mode", ModuleType("slack_sdk.socket_mode")) + sm_aio = sys.modules.setdefault( + "slack_sdk.socket_mode.aiohttp", ModuleType("slack_sdk.socket_mode.aiohttp") + ) + sm_resp = sys.modules.setdefault( + "slack_sdk.socket_mode.response", ModuleType("slack_sdk.socket_mode.response") + ) + + if not hasattr(sm_aio, "SocketModeClient"): + + class _StubSocketModeClient: + """Replaced per-test by tests that exercise the lifecycle.""" + + def __init__(self, *args: Any, app_token: str | None = None, **kwargs: Any): + self.app_token = app_token + self.socket_mode_request_listeners: list[Any] = [] + self._connected = False + + async def connect(self) -> None: + self._connected = True + + async def disconnect(self) -> None: + self._connected = False + + def is_connected(self) -> bool: + return self._connected + + async def send_socket_mode_response(self, _response: Any) -> None: + return None + + sm_aio.SocketModeClient = _StubSocketModeClient # type: ignore[attr-defined] + + if not hasattr(sm_resp, "SocketModeResponse"): + + class _StubSocketModeResponse: + def __init__(self, envelope_id: str = "", payload: Any = None): + self.envelope_id = envelope_id + self.payload = payload + + sm_resp.SocketModeResponse = _StubSocketModeResponse # type: ignore[attr-defined] + + sm_root.aiohttp = sm_aio # type: ignore[attr-defined] + sm_root.response = sm_resp # type: ignore[attr-defined] + setattr(sys.modules["slack_sdk"], "socket_mode", sm_root) + + +_ensure_socket_mode_stub() + + +try: + from chat_sdk.adapters.slack.adapter import SlackAdapter, create_slack_adapter # noqa: E402 + from chat_sdk.adapters.slack.types import SlackAdapterConfig # noqa: E402 + from chat_sdk.shared.errors import ValidationError # noqa: E402 + from chat_sdk.types import ModalResponse # noqa: E402 + + _SLACK_AVAILABLE = True +except ImportError: + _SLACK_AVAILABLE = False + +pytestmark = pytest.mark.skipif(not _SLACK_AVAILABLE, reason="Slack adapter import failed") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_socket_adapter(**overrides: Any) -> SlackAdapter: + config = SlackAdapterConfig( + mode="socket", + app_token=overrides.pop("app_token", "xapp-1-test"), + bot_token=overrides.pop("bot_token", "xoxb-test-token"), + signing_secret=overrides.pop("signing_secret", None), + socket_forwarding_secret=overrides.pop("socket_forwarding_secret", "fwd-secret"), + **overrides, + ) + return SlackAdapter(config) + + +def _make_mock_state() -> MagicMock: + cache: dict[str, Any] = {} + state = MagicMock() + state.get = AsyncMock(side_effect=lambda k: cache.get(k)) + state.set = AsyncMock(side_effect=lambda k, v, *a, **kw: cache.__setitem__(k, v)) + state.delete = AsyncMock(side_effect=lambda k: cache.pop(k, None)) + state.append_to_list = AsyncMock() + state.get_list = AsyncMock(return_value=[]) + state._cache = cache + return state + + +def _make_mock_chat() -> MagicMock: + state = _make_mock_state() + chat = MagicMock() + chat.process_message = MagicMock() + chat.handle_incoming_message = AsyncMock() + chat.process_reaction = MagicMock() + chat.process_action = MagicMock() + chat.process_modal_submit = AsyncMock() + chat.process_modal_close = MagicMock() + chat.process_slash_command = MagicMock() + chat.process_member_joined_channel = 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 + + +class _FakeRequest: + def __init__(self, body: str, headers: dict[str, str] | None = None): + self.body = body.encode("utf-8") + self.headers = headers or {} + self.url = "" + + async def text(self) -> str: + return self.body.decode("utf-8") + + +# --------------------------------------------------------------------------- +# Construction / config validation +# --------------------------------------------------------------------------- + + +class TestSocketModeConfig: + def test_socket_mode_requires_app_token(self): + with pytest.raises(ValidationError, match="appToken is required"): + SlackAdapter(SlackAdapterConfig(mode="socket", bot_token="xoxb-test")) + + def test_socket_mode_app_token_must_start_with_xapp(self): + with pytest.raises(ValidationError, match="must start with 'xapp-'"): + SlackAdapter( + SlackAdapterConfig( + mode="socket", + app_token="xoxb-not-an-app-token", + bot_token="xoxb-test", + ) + ) + + def test_socket_mode_signing_secret_optional(self): + # Should not raise even though signing_secret is None. + adapter = SlackAdapter( + SlackAdapterConfig( + mode="socket", + app_token="xapp-1-foo", + bot_token="xoxb-test", + ) + ) + assert adapter.is_socket_mode is True + assert adapter.mode == "socket" + + def test_webhook_mode_still_requires_signing_secret(self): + # Make sure socket-mode allowance didn't accidentally relax the + # webhook-mode check. Clear env so the test isn't accidentally + # satisfied by SLACK_SIGNING_SECRET in the dev shell. + prev = os.environ.pop("SLACK_SIGNING_SECRET", None) + try: + with pytest.raises(ValidationError, match="signingSecret is required"): + SlackAdapter(SlackAdapterConfig(bot_token="xoxb-test")) + finally: + if prev is not None: + os.environ["SLACK_SIGNING_SECRET"] = prev + + def test_app_token_picked_up_from_env(self): + prev = os.environ.get("SLACK_APP_TOKEN") + os.environ["SLACK_APP_TOKEN"] = "xapp-1-from-env" + try: + adapter = SlackAdapter( + SlackAdapterConfig(mode="socket", bot_token="xoxb-test"), + ) + assert adapter._app_token == "xapp-1-from-env" + finally: + if prev is not None: + os.environ["SLACK_APP_TOKEN"] = prev + else: + os.environ.pop("SLACK_APP_TOKEN", None) + + def test_socket_forwarding_secret_falls_back_to_app_token(self): + adapter = SlackAdapter( + SlackAdapterConfig( + mode="socket", + app_token="xapp-1-foo", + bot_token="xoxb-test", + ) + ) + # Falls back to app_token only when neither config nor env is set. + assert adapter._socket_forwarding_secret == "xapp-1-foo" + + def test_create_slack_adapter_rejects_multi_workspace_in_socket_mode(self): + with pytest.raises(ValidationError, match="Multi-workspace"): + create_slack_adapter( + SlackAdapterConfig( + mode="socket", + app_token="xapp-1-foo", + client_id="cid", + client_secret="csec", + ) + ) + + +# --------------------------------------------------------------------------- +# Modal "clear" response action +# --------------------------------------------------------------------------- + + +class TestModalClearResponse: + def test_clear_action_emits_response_action_clear(self): + adapter = SlackAdapter( + SlackAdapterConfig(signing_secret="s", bot_token="xoxb-x"), + ) + result = adapter._modal_response_to_slack(ModalResponse(action="clear")) + assert result == {"response_action": "clear"} + + +# --------------------------------------------------------------------------- +# Forwarded socket events via handle_webhook +# --------------------------------------------------------------------------- + + +class TestForwardedSocketEvents: + async def test_webhook_in_socket_mode_returns_405_without_token(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + request = _FakeRequest( + json.dumps({"type": "event_callback", "event": {"type": "message"}}), + headers={"content-type": "application/json"}, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 405 + + async def test_webhook_accepts_valid_socket_token(self): + adapter = _make_socket_adapter() + chat = _make_mock_chat() + adapter._chat = chat + # Drive the route through events_api so we can verify that the + # underlying process_message handler is invoked. + forwarded = { + "type": "socket_event", + "eventType": "events_api", + "body": { + "type": "event_callback", + "event": { + "type": "message", + "channel": "C123", + "ts": "1.0", + "user": "U1", + "text": "hi", + "team": "T1", + }, + "team_id": "T1", + }, + "timestamp": 0, + } + request = _FakeRequest( + json.dumps(forwarded), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "fwd-secret", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 200 + assert result["body"] == "ok" + # process_message gets called with the parsed event. + assert chat.process_message.called + + async def test_webhook_rejects_invalid_socket_token(self): + adapter = _make_socket_adapter(socket_forwarding_secret="real-secret") + adapter._chat = _make_mock_chat() + request = _FakeRequest( + json.dumps({"type": "socket_event", "eventType": "events_api", "body": {}}), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "wrong-secret", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 401 + + async def test_webhook_socket_token_rejected_when_no_secret_configured(self): + # Adapter in webhook mode with no forwarding secret + no app token. + adapter = SlackAdapter( + SlackAdapterConfig(signing_secret="s", bot_token="xoxb-x"), + ) + # Belt and suspenders: ensure the secret is unset. + adapter._socket_forwarding_secret = None + request = _FakeRequest( + json.dumps({"type": "socket_event"}), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "anything", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 401 + + +# --------------------------------------------------------------------------- +# _route_socket_event dispatch +# --------------------------------------------------------------------------- + + +class TestRouteSocketEvent: + async def test_events_api_acks_then_dispatches(self): + adapter = _make_socket_adapter() + chat = _make_mock_chat() + adapter._chat = chat + ack = AsyncMock() + body = { + "team_id": "T1", + "event": { + "type": "message", + "channel": "C1", + "ts": "1.0", + "user": "U1", + "text": "hello", + "team": "T1", + }, + } + await adapter._route_socket_event(body, "events_api", ack) + ack.assert_awaited_once_with() + assert chat.process_message.called + + async def test_events_api_missing_event_field_does_not_crash(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + ack = AsyncMock() + await adapter._route_socket_event({"team_id": "T1"}, "events_api", ack) + ack.assert_awaited_once_with() + + async def test_slash_command_acks_immediately_and_dispatches(self): + adapter = _make_socket_adapter() + chat = _make_mock_chat() + adapter._chat = chat + # Slash dispatch calls _lookup_user → Slack API. Stub it. + adapter._lookup_user = AsyncMock( # type: ignore[method-assign] + return_value={"display_name": "u1", "real_name": "u1"} + ) + ack = AsyncMock() + body = { + "command": "/foo", + "text": "bar", + "user_id": "U1", + "channel_id": "C1", + "team_id": "T1", + } + await adapter._route_socket_event(body, "slash_commands", ack) + # Ack is sent immediately with no payload. + ack.assert_awaited_once_with() + # Slash dispatch is fire-and-forget — give the spawned task a turn + # of the event loop to land. + for _ in range(50): + if chat.process_slash_command.called: + break + await asyncio.sleep(0.01) + assert chat.process_slash_command.called + + async def test_interactive_acks_with_response_body_for_view_submission_errors(self): + adapter = _make_socket_adapter() + chat = _make_mock_chat() + adapter._chat = chat + + # Have the modal-submit handler return errors so the dispatcher + # builds an `errors` response body that should round-trip through + # the ack. + async def fake_modal_submit(*args: Any, **kwargs: Any) -> ModalResponse: + return ModalResponse(action="errors", errors={"field": "bad"}) + + chat.process_modal_submit = AsyncMock(side_effect=fake_modal_submit) + ack = AsyncMock() + payload = { + "type": "view_submission", + "team": {"id": "T1"}, + "user": {"id": "U1", "name": "x"}, + "view": { + "id": "V1", + "callback_id": "cb", + "private_metadata": "", + "state": {"values": {}}, + }, + "trigger_id": "trig", + } + await adapter._route_socket_event(payload, "interactive", ack) + # Ack was awaited exactly once with the errors response body. + assert ack.await_count == 1 + ack_args = ack.call_args + assert ack_args.args, "ack should be called with the response body" + body_arg = ack_args.args[0] + assert isinstance(body_arg, dict) + assert body_arg.get("response_action") == "errors" + + async def test_retry_attempt_is_skipped_but_acked(self): + adapter = _make_socket_adapter() + chat = _make_mock_chat() + adapter._chat = chat + request = MagicMock() + request.envelope_id = "env-1" + request.type = "events_api" + request.payload = {"event": {"type": "message"}} + request.retry_attempt = 2 # Slack retry — should be skipped. + + client = MagicMock() + client.send_socket_mode_response = AsyncMock() + + await adapter._on_socket_request(client, request) + + # Ack went out (so Slack stops resending), but no dispatch. + assert client.send_socket_mode_response.await_count == 1 + assert chat.process_message.called is False + + async def test_unknown_event_type_acks_and_does_nothing(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + ack = AsyncMock() + await adapter._route_socket_event({}, "weird", ack) + ack.assert_awaited_once_with() + + +# --------------------------------------------------------------------------- +# Lifecycle: start_socket_mode / stop_socket_mode +# --------------------------------------------------------------------------- + + +class _FakeSocketModeClient: + """In-process stand-in for slack_sdk's SocketModeClient.""" + + instances: list[_FakeSocketModeClient] = [] + + def __init__(self, *args: Any, app_token: str | None = None, **kwargs: Any): + self.app_token = app_token + self.socket_mode_request_listeners: list[Any] = [] + self._connected = False + self.connect_calls = 0 + self.disconnect_calls = 0 + # Test hooks + self.fail_first_connect = False + self.disconnect_after_s: float | None = None + type(self).instances.append(self) + + async def connect(self) -> None: + self.connect_calls += 1 + if self.fail_first_connect: + raise RuntimeError("simulated connect failure") + self._connected = True + if self.disconnect_after_s is not None: + asyncio.get_event_loop().call_later(self.disconnect_after_s, lambda: setattr(self, "_connected", False)) + + async def disconnect(self) -> None: + self.disconnect_calls += 1 + self._connected = False + + def is_connected(self) -> bool: + return self._connected + + +@pytest.fixture +def patched_socket_client(monkeypatch): + """Patch slack_sdk.socket_mode.aiohttp.SocketModeClient with a fake.""" + _FakeSocketModeClient.instances.clear() + import slack_sdk.socket_mode.aiohttp as sm + + monkeypatch.setattr(sm, "SocketModeClient", _FakeSocketModeClient) + yield _FakeSocketModeClient + _FakeSocketModeClient.instances.clear() + + +class TestSocketModeLifecycle: + async def test_start_then_stop(self, patched_socket_client): + adapter = _make_socket_adapter() + # Make backoff fast so the loop responds quickly to disconnects. + adapter._socket_initial_backoff_s = 0.05 + adapter._socket_max_backoff_s = 0.1 + await adapter.start_socket_mode() + assert len(patched_socket_client.instances) == 1 + client = patched_socket_client.instances[0] + assert client.connect_calls == 1 + # Listener registered + assert len(client.socket_mode_request_listeners) == 1 + + await adapter.stop_socket_mode() + # Disconnect was called at least once on the client. + assert client.disconnect_calls >= 1 + # Task is cleared. + assert adapter._socket_task is None + + async def test_start_is_idempotent(self, patched_socket_client): + adapter = _make_socket_adapter() + adapter._socket_initial_backoff_s = 0.05 + await adapter.start_socket_mode() + await adapter.start_socket_mode() # second call no-ops + assert len(patched_socket_client.instances) == 1 + await adapter.stop_socket_mode() + + async def test_stop_is_idempotent(self, patched_socket_client): + adapter = _make_socket_adapter() + adapter._socket_initial_backoff_s = 0.05 + await adapter.start_socket_mode() + await adapter.stop_socket_mode() + await adapter.stop_socket_mode() # safe second call + + async def test_first_connect_failure_propagates(self, patched_socket_client): + # Patch the fake to fail every connect. + original_connect = _FakeSocketModeClient.connect + + async def always_fail(self): + self.connect_calls += 1 + raise RuntimeError("boom") + + _FakeSocketModeClient.connect = always_fail # type: ignore[assignment] + try: + adapter = _make_socket_adapter() + adapter._socket_initial_backoff_s = 0.01 + with pytest.raises(RuntimeError, match="boom"): + await adapter.start_socket_mode() + finally: + _FakeSocketModeClient.connect = original_connect # type: ignore[assignment] + + async def test_reconnects_after_transient_disconnect(self, patched_socket_client): + # First client: connect succeeds, then immediately reports + # disconnected so the loop has to reconnect. + original_connect = _FakeSocketModeClient.connect + attempt_count = {"n": 0} + + async def staged_connect(self): + attempt_count["n"] += 1 + self.connect_calls += 1 + self._connected = True + # First-stage client: drop connection after a tick to force + # the reconnect path. Second-stage client stays up. + if attempt_count["n"] == 1: + # Schedule a drop after the loop's polling interval. + async def drop(): + await asyncio.sleep(0.01) + self._connected = False + + asyncio.create_task(drop()) + + _FakeSocketModeClient.connect = staged_connect # type: ignore[assignment] + try: + adapter = _make_socket_adapter() + adapter._socket_initial_backoff_s = 0.01 + adapter._socket_max_backoff_s = 0.05 + await adapter.start_socket_mode() + # Give the loop time to detect the drop and reconnect at least once. + for _ in range(50): + if attempt_count["n"] >= 2: + break + await asyncio.sleep(0.05) + await adapter.stop_socket_mode() + assert attempt_count["n"] >= 2, f"expected reconnect, got {attempt_count['n']}" + finally: + _FakeSocketModeClient.connect = original_connect # type: ignore[assignment] + + +# --------------------------------------------------------------------------- +# ContextVar boundary: per-event request context survives into spawned tasks +# --------------------------------------------------------------------------- + + +class TestSocketContextVar: + async def test_request_context_isolated_per_event(self): + # Multi-workspace adapter so that events_api goes through token + # resolution + ContextVar setup. + adapter = SlackAdapter( + SlackAdapterConfig( + mode="socket", + app_token="xapp-1-x", + client_id="cid", + client_secret="csec", + ) + ) + chat = _make_mock_chat() + adapter._chat = chat + from chat_sdk.adapters.slack.types import RequestContext + + # Stub the per-team token lookup so the route doesn't hit storage. + adapter._resolve_token_for_team = AsyncMock( # type: ignore[method-assign] + return_value=RequestContext(token="xoxb-team-1") + ) + + captured_tokens: list[str | None] = [] + + def capture_message(*args: Any, **kwargs: Any) -> None: + ctx = adapter._request_context.get() + captured_tokens.append(ctx.token if ctx else None) + + chat.process_message = MagicMock(side_effect=capture_message) + ack = AsyncMock() + body = { + "team_id": "T1", + "event": { + "type": "message", + "channel": "C1", + "ts": "1.0", + "user": "U1", + "text": "hi", + }, + } + await adapter._route_socket_event(body, "events_api", ack) + assert captured_tokens == ["xoxb-team-1"] + # And the outer context wasn't polluted. + assert adapter._request_context.get() is None From b8a6e076361d79ca9eb2167c653afcb2ac9ea3f2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 9 May 2026 18:22:02 +0000 Subject: [PATCH 2/9] fix(slack): asyncio.Event for socket shutdown + cancel orphan wait_task Address gemini-code-assist review on PR #86: 1. Line 1375: ``connected.wait()`` was wrapped in an untracked task that stayed pending if the loop task finished first (e.g., first connect failed). Track it explicitly and cancel when it lands in ``pending``. 2. Line 1504: replace the ``not self._socket_shutdown`` polling loop with a per-adapter ``asyncio.Event`` waited via ``asyncio.wait_for``. One event per adapter (not per backoff iteration); wakeup latency is now bounded by the event-loop scheduler rather than the previous 250ms poll cadence. Updates ``start_socket_mode`` (clear), ``stop_socket_mode`` (set), and the four loop checks. 24 socket-mode tests still pass. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- src/chat_sdk/adapters/slack/adapter.py | 47 +++++++++++++++----------- 1 file changed, 27 insertions(+), 20 deletions(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 5afe05f6..fe879396 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -241,8 +241,10 @@ def __init__(self, config: SlackAdapterConfig | None = None) -> None: # ``disconnect()`` can cancel it cleanly (hazard #5). self._socket_task: asyncio.Task[None] | None = None # Set when shutdown is requested so the reconnect loop knows to exit - # rather than retry on a clean disconnect. - self._socket_shutdown = False + # rather than retry on a clean disconnect. The Event also wakes up + # ``_socket_sleep_with_backoff`` immediately so ``stop_socket_mode`` + # doesn't have to wait the full backoff window. + self._socket_shutdown_event: asyncio.Event = asyncio.Event() # Default backoff schedule in seconds. Kept short so tests run fast, # but capped low enough that a flapping Slack connection doesn't busy # loop. Slack's recommended pattern is exponential backoff with jitter; @@ -1357,7 +1359,7 @@ async def start_socket_mode(self) -> None: "slack_sdk is not installed. Install with `pip install chat-sdk[slack]`.", ) from exc - self._socket_shutdown = False + self._socket_shutdown_event.clear() connected = asyncio.Event() loop = asyncio.get_running_loop() # Hazard #5: track the task explicitly so ``stop_socket_mode`` can @@ -1369,10 +1371,15 @@ async def start_socket_mode(self) -> None: # exit (which means the very first connect raised). Re-raise so the # caller learns about a hard config failure (bad app token, network # offline) instead of silently spinning forever. - first_done, _ = await asyncio.wait( - {asyncio.create_task(connected.wait()), self._socket_task}, + wait_task = asyncio.create_task(connected.wait()) + first_done, pending = await asyncio.wait( + {wait_task, self._socket_task}, return_when=asyncio.FIRST_COMPLETED, ) + # Hazard #5: if the loop task finished first, cancel the wait task + # explicitly so the orphan ``connected.wait()`` doesn't sit forever. + if wait_task in pending: + wait_task.cancel() if connected.is_set(): return # Socket loop exited before connecting — surface its exception. @@ -1388,7 +1395,7 @@ async def stop_socket_mode(self) -> None: Idempotent. Safe to call from any task; it disconnects the active client and waits for the background task to finish. """ - self._socket_shutdown = True + self._socket_shutdown_event.set() client = self._socket_client self._socket_client = None @@ -1421,7 +1428,7 @@ async def _socket_mode_loop(self, connected: asyncio.Event) -> None: backoff = self._socket_initial_backoff_s try: - while not self._socket_shutdown: + while not self._socket_shutdown_event.is_set(): client = SocketModeClient(app_token=cast(str, self._app_token)) # Register our request handler. ``socket_mode_request_listeners`` # is the documented public extension point on the slack_sdk @@ -1436,7 +1443,7 @@ async def _socket_mode_loop(self, connected: asyncio.Event) -> None: {"error": str(exc)}, ) self._socket_client = None - if self._socket_shutdown: + if self._socket_shutdown_event.is_set(): return if not connected.is_set(): # First connect failed and nobody's listening yet — @@ -1453,7 +1460,7 @@ async def _socket_mode_loop(self, connected: asyncio.Event) -> None: backoff = self._socket_initial_backoff_s # Wait until the socket disconnects or shutdown is requested. - while not self._socket_shutdown: + while not self._socket_shutdown_event.is_set(): if not client.is_connected(): break await asyncio.sleep(1.0) @@ -1468,7 +1475,7 @@ async def _socket_mode_loop(self, connected: asyncio.Event) -> None: {"error": str(exc)}, ) - if self._socket_shutdown: + if self._socket_shutdown_event.is_set(): return self._logger.info("Slack socket mode disconnected, reconnecting") await self._socket_sleep_with_backoff(backoff) @@ -1492,16 +1499,16 @@ async def _socket_mode_loop(self, connected: asyncio.Event) -> None: await client.disconnect() async def _socket_sleep_with_backoff(self, seconds: float) -> None: - """Sleep for ``seconds`` but wake immediately on shutdown.""" - # Poll every 0.25s so a stop_socket_mode call doesn't have to wait - # the full backoff window. asyncio.Event would be slightly cleaner - # but we don't want to add an Event per loop iteration. - deadline = asyncio.get_event_loop().time() + seconds - while not self._socket_shutdown: - remaining = deadline - asyncio.get_event_loop().time() - if remaining <= 0: - return - await asyncio.sleep(min(0.25, remaining)) + """Sleep for ``seconds`` but wake immediately on shutdown. + + Uses the per-adapter ``_socket_shutdown_event`` so ``stop_socket_mode`` + can interrupt the backoff window without polling — wakeup latency is + bounded by event-loop scheduling, not the previous 0.25s poll. + """ + try: + await asyncio.wait_for(self._socket_shutdown_event.wait(), timeout=seconds) + except asyncio.TimeoutError: + return async def _on_socket_request(self, client: Any, request: Any) -> None: """Listener invoked by ``SocketModeClient`` for each socket message. From e1c8047186cb6292b1a0667582009e102dab6c0d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 9 May 2026 20:16:28 +0000 Subject: [PATCH 3/9] fix(slack): address PR #86 review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hardens the socket-mode port against the seven review findings on PR #86: * Add ``connect_timeout_s`` config (default 30s) and wrap the initial socket-mode handshake in ``asyncio.wait_for`` so a hung ``SocketModeClient.connect()`` can't make ``initialize()`` block forever (hazard #11). On timeout we tear the loop down before raising. * Reject forwarded socket events whose ``timestamp`` field is outside the same 5-minute window ``_verify_signature`` enforces, so a captured forwarded payload can't be replayed indefinitely (hazard #12). * Narrow the ``contextlib.suppress`` in ``stop_socket_mode`` to ``CancelledError`` only — surprising loop crashes are no longer silently swallowed during shutdown. * Replace ``asyncio.get_event_loop().create_task`` in the socket-mode ``wrap_async`` helper with ``get_running_loop`` (Python 3.12+ compatibility, hazard #5). * Have the socket-mode interactive branch ack with ``response_action: errors`` instead of an empty ack when dispatch raises — an empty ack on ``view_submission`` silently closes the modal so the user sees no signal anything went wrong. * Drop ``is_ext_shared_channel`` from the synthesized ``event_callback`` payload in the socket-mode events_api branch so socket and webhook paths feed identical shapes into ``_process_event_payload`` (hazard #7). * Add a regression test firing two concurrent ``_route_socket_event`` events_api dispatches for different teams via ``asyncio.gather`` to pin down the existing ``copy_context()`` isolation against future drift (hazard #6). Also fixes a pre-existing B010 lint warning in the test stub. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- src/chat_sdk/adapters/slack/adapter.py | 70 ++++++- src/chat_sdk/adapters/slack/types.py | 5 + tests/test_slack_socket_mode.py | 268 ++++++++++++++++++++++++- 3 files changed, 324 insertions(+), 19 deletions(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index fe879396..8eacfbab 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -251,6 +251,11 @@ def __init__(self, config: SlackAdapterConfig | None = None) -> None: # our minimal schedule mirrors that behavior with explicit caps. self._socket_initial_backoff_s = 1.0 self._socket_max_backoff_s = 30.0 + # Bound the initial Socket Mode handshake so ``initialize()`` doesn't + # block forever if slack_sdk's ``connect()`` hangs (hazard #11). + self._socket_connect_timeout_s: float = ( + config.connect_timeout_s if config.connect_timeout_s is not None else 30.0 + ) self._logger: Logger = config.logger or ConsoleLogger("info") self._user_name: str = config.user_name or "bot" self._bot_user_id: str | None = config.bot_user_id or None @@ -805,6 +810,21 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No event = json.loads(body) except (json.JSONDecodeError, ValueError): return {"body": "Invalid JSON", "status": 400} + # Hazard #12 (replay): the shared bearer alone is not enough — + # without a freshness check, an old captured forwarded event + # could be replayed indefinitely. Mirror the 5-minute window + # ``_verify_signature`` enforces on signed webhook traffic. + ts_raw = event.get("timestamp") if isinstance(event, dict) else None + try: + ts_int = int(ts_raw) if ts_raw is not None else None + except (TypeError, ValueError): + ts_int = None + if ts_int is None or abs(int(time.time()) - ts_int) > 300: + self._logger.warn( + "Forwarded socket event outside freshness window", + {"timestamp": ts_raw}, + ) + return {"body": "Stale socket event", "status": 401} await self._handle_forwarded_socket_event(event, options) return {"body": "ok", "status": 200} @@ -1370,12 +1390,29 @@ async def start_socket_mode(self) -> None: # Wait for either the first successful connect or for the loop to # exit (which means the very first connect raised). Re-raise so the # caller learns about a hard config failure (bad app token, network - # offline) instead of silently spinning forever. + # offline) instead of silently spinning forever. Bound the wait with + # ``connect_timeout_s`` so a hung handshake (slack_sdk's ``connect()`` + # never returning) doesn't make ``initialize()`` block indefinitely + # (hazard #11). wait_task = asyncio.create_task(connected.wait()) - first_done, pending = await asyncio.wait( - {wait_task, self._socket_task}, - return_when=asyncio.FIRST_COMPLETED, - ) + try: + first_done, pending = await asyncio.wait_for( + asyncio.shield( + asyncio.wait( + {wait_task, self._socket_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + ), + timeout=self._socket_connect_timeout_s, + ) + except asyncio.TimeoutError: + # Don't leak the wait task or the still-running socket loop — + # tear them down before surfacing the failure. + wait_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await wait_task + await self.stop_socket_mode() + raise TimeoutError(f"Slack Socket Mode connect timed out after {self._socket_connect_timeout_s}s") from None # Hazard #5: if the loop task finished first, cancel the wait task # explicitly so the orphan ``connected.wait()`` doesn't sit forever. if wait_task in pending: @@ -1409,9 +1446,9 @@ async def stop_socket_mode(self) -> None: self._socket_task = None if task is not None and not task.done(): task.cancel() - # Cancellation is expected; any other exception was already - # logged inside the loop. Don't re-raise on shutdown. - with contextlib.suppress(asyncio.CancelledError, Exception): + # Cancellation is expected on shutdown; surface anything else so + # surprising loop crashes aren't silently swallowed. + with contextlib.suppress(asyncio.CancelledError): await task if task is not None: self._logger.info("Slack socket mode disconnected") @@ -1565,7 +1602,7 @@ def wrap_async(coro: Awaitable[Any]) -> None: # ``wait_until`` semantics: caller takes ownership. options.wait_until(cast(Any, coro)) return - task = asyncio.get_event_loop().create_task(cast(Any, coro)) + task = asyncio.get_running_loop().create_task(cast(Any, coro)) def _log_exc(t: asyncio.Task[Any]) -> None: if t.cancelled(): @@ -1589,13 +1626,18 @@ 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. payload: dict[str, Any] = { "type": "event_callback", "event": event, "team_id": body.get("team_id"), "event_id": body.get("event_id"), "event_time": body.get("event_time"), - "is_ext_shared_channel": body.get("is_ext_shared_channel"), } # Multi-workspace: resolve token before dispatch (mirrors webhook # path). copy_context() keeps the ContextVar set on tasks spawned @@ -1670,7 +1712,13 @@ async def run_slash() -> None: "Error processing socket mode interactive", {"error": str(exc)}, ) - await ack() + # Hazard #15 (UX): an empty ack on view_submission silently + # closes the modal, so the user has no signal anything went + # wrong. Return ``response_action=errors`` so Slack keeps the + # modal open with a visible message. Safe for non-modal + # interactive types too — Slack ignores the field when the + # payload type doesn't expect it. + await ack({"response_action": "errors", "errors": {"_": "internal error"}}) return response_body: dict[str, Any] | None = None diff --git a/src/chat_sdk/adapters/slack/types.py b/src/chat_sdk/adapters/slack/types.py index 13f33921..2f3fd21b 100644 --- a/src/chat_sdk/adapters/slack/types.py +++ b/src/chat_sdk/adapters/slack/types.py @@ -56,6 +56,11 @@ class SlackAdapterConfig: # Maximum number of cached AsyncWebClient instances (LRU-bounded). # Defaults to 100. Increase for large multi-workspace deployments. client_cache_max: int | None = None + # Maximum number of seconds to wait for the initial Socket Mode WebSocket + # handshake. If the slack_sdk ``connect()`` call hangs (e.g. Slack edge + # is degraded), ``start_socket_mode`` raises after this many seconds so + # ``initialize()`` doesn't block forever (hazard #11). + connect_timeout_s: float = 30.0 # Override bot username (optional) user_name: str | None = None diff --git a/tests/test_slack_socket_mode.py b/tests/test_slack_socket_mode.py index 85d4c4d4..67a3a924 100644 --- a/tests/test_slack_socket_mode.py +++ b/tests/test_slack_socket_mode.py @@ -46,12 +46,8 @@ def _ensure_socket_mode_stub() -> None: sys.modules.setdefault("slack_sdk", ModuleType("slack_sdk")) sm_root = sys.modules.setdefault("slack_sdk.socket_mode", ModuleType("slack_sdk.socket_mode")) - sm_aio = sys.modules.setdefault( - "slack_sdk.socket_mode.aiohttp", ModuleType("slack_sdk.socket_mode.aiohttp") - ) - sm_resp = sys.modules.setdefault( - "slack_sdk.socket_mode.response", ModuleType("slack_sdk.socket_mode.response") - ) + sm_aio = sys.modules.setdefault("slack_sdk.socket_mode.aiohttp", ModuleType("slack_sdk.socket_mode.aiohttp")) + sm_resp = sys.modules.setdefault("slack_sdk.socket_mode.response", ModuleType("slack_sdk.socket_mode.response")) if not hasattr(sm_aio, "SocketModeClient"): @@ -88,7 +84,7 @@ def __init__(self, envelope_id: str = "", payload: Any = None): sm_root.aiohttp = sm_aio # type: ignore[attr-defined] sm_root.response = sm_resp # type: ignore[attr-defined] - setattr(sys.modules["slack_sdk"], "socket_mode", sm_root) + sys.modules["slack_sdk"].socket_mode = sm_root # type: ignore[attr-defined] _ensure_socket_mode_stub() @@ -275,6 +271,8 @@ async def test_webhook_in_socket_mode_returns_405_without_token(self): assert result["status"] == 405 async def test_webhook_accepts_valid_socket_token(self): + import time as _time + adapter = _make_socket_adapter() chat = _make_mock_chat() adapter._chat = chat @@ -295,7 +293,7 @@ async def test_webhook_accepts_valid_socket_token(self): }, "team_id": "T1", }, - "timestamp": 0, + "timestamp": int(_time.time()), } request = _FakeRequest( json.dumps(forwarded), @@ -646,3 +644,257 @@ def capture_message(*args: Any, **kwargs: Any) -> None: assert captured_tokens == ["xoxb-team-1"] # And the outer context wasn't polluted. assert adapter._request_context.get() is None + + async def test_concurrent_events_for_different_teams_do_not_cross_contaminate(self): + """Two concurrent ``_route_socket_event(events_api)`` calls for + different teams must not see each other's tokens (hazard #6). + + What to fix if this fails: the events_api branch in + ``_route_socket_event`` must use ``contextvars.copy_context()`` (or + equivalent isolation) so that a slow handler for team T1 doesn't + observe the ContextVar that another concurrent dispatch set for + team T2. Direct ``ContextVar.set()`` without isolation will leak + across ``asyncio.gather`` task boundaries. + """ + adapter = SlackAdapter( + SlackAdapterConfig( + mode="socket", + app_token="xapp-1-x", + client_id="cid", + client_secret="csec", + ) + ) + chat = _make_mock_chat() + adapter._chat = chat + from chat_sdk.adapters.slack.types import RequestContext + + # 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: + if team_id == "T1": + # Yield so the T2 dispatch can race in and set the + # ContextVar before T1's process_message runs. + await asyncio.sleep(0.02) + return RequestContext(token=f"xoxb-{team_id}") + + adapter._resolve_token_for_team = AsyncMock( # type: ignore[method-assign] + side_effect=fake_resolve + ) + + observed: dict[str, list[str | None]] = {"T1": [], "T2": []} + + def capture(*args: Any, **kwargs: Any) -> None: + ctx = adapter._request_context.get() + tok = ctx.token if ctx else None + # Map back to the team via the token suffix we constructed. + if tok == "xoxb-T1": + observed["T1"].append(tok) + elif tok == "xoxb-T2": + observed["T2"].append(tok) + else: + observed.setdefault("other", []).append(tok) + + chat.process_message = MagicMock(side_effect=capture) + ack = AsyncMock() + + def body_for(team: str) -> dict[str, Any]: + return { + "team_id": team, + "event": { + "type": "message", + "channel": f"C-{team}", + "ts": "1.0", + "user": "U1", + "text": "hi", + "team": team, + }, + } + + # Fire both concurrently. + await asyncio.gather( + adapter._route_socket_event(body_for("T1"), "events_api", ack), + adapter._route_socket_event(body_for("T2"), "events_api", ack), + ) + + assert observed["T1"] == ["xoxb-T1"], f"T1 saw wrong token(s): {observed}" + assert observed["T2"] == ["xoxb-T2"], f"T2 saw wrong token(s): {observed}" + # Outer context wasn't polluted by either dispatch. + assert adapter._request_context.get() is None + + +# --------------------------------------------------------------------------- +# Review-finding regression tests (PR #86) +# --------------------------------------------------------------------------- + + +class TestSocketConnectTimeout: + """Regression for review finding #1. + + What to fix if this fails: ``start_socket_mode`` must wrap the wait on + the initial connect with ``asyncio.wait_for(..., timeout=...)`` so a + hung ``SocketModeClient.connect()`` cannot block ``initialize()`` + forever (hazard #11). On timeout the loop must be torn down. + """ + + async def test_hung_connect_raises_timeout_and_cleans_up(self, patched_socket_client): + original_connect = _FakeSocketModeClient.connect + + async def hang_forever(self): + self.connect_calls += 1 + # Sleep longer than any reasonable test timeout. The fix should + # cancel us via the outer ``asyncio.wait_for``. + await asyncio.sleep(60) + + _FakeSocketModeClient.connect = hang_forever # type: ignore[assignment] + try: + adapter = _make_socket_adapter() + adapter._socket_connect_timeout_s = 0.1 + adapter._socket_initial_backoff_s = 0.01 + with pytest.raises(TimeoutError, match="timed out"): + await adapter.start_socket_mode() + # The teardown path must clear the background task. + assert adapter._socket_task is None + assert adapter._socket_client is None + finally: + _FakeSocketModeClient.connect = original_connect # type: ignore[assignment] + + def test_connect_timeout_default_is_30s(self): + """Default surfaces in adapter state. + + What to fix if this fails: ``SlackAdapterConfig.connect_timeout_s`` + must default to ~30s and propagate into the adapter, otherwise the + timeout fix would silently regress to ``None`` / very small values. + """ + adapter = _make_socket_adapter() + assert adapter._socket_connect_timeout_s == 30.0 + + +class TestForwardedSocketFreshness: + """Regression for review finding #2. + + What to fix if this fails: ``handle_webhook`` must reject forwarded + socket events whose ``timestamp`` field is outside the 5-minute window + (mirroring ``_verify_signature``). Without this an attacker who + captures one forwarded payload can replay it indefinitely (hazard #12). + """ + + async def test_replay_old_event_rejected(self): + import time as _time + + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + forwarded = { + "type": "socket_event", + "eventType": "events_api", + "body": {"type": "event_callback", "event": {}}, + "timestamp": int(_time.time()) - 6 * 60, # 6 minutes old + } + request = _FakeRequest( + json.dumps(forwarded), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "fwd-secret", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 401, "stale forwarded event must be rejected" + + async def test_missing_timestamp_rejected(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + forwarded = { + "type": "socket_event", + "eventType": "events_api", + "body": {"type": "event_callback", "event": {}}, + # No "timestamp" — must not pass freshness check. + } + request = _FakeRequest( + json.dumps(forwarded), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "fwd-secret", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 401 + + +class TestInteractiveDispatchErrorAck: + """Regression for review finding #5. + + What to fix if this fails: when ``_dispatch_interactive_payload`` raises + in the socket-mode interactive branch, the ack must include + ``response_action: errors`` instead of being empty. An empty ack on a + ``view_submission`` silently closes the modal — the user gets no signal + anything went wrong. + """ + + async def test_dispatch_exception_acks_with_errors_response_action(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + # Force the dispatcher to blow up. + adapter._dispatch_interactive_payload = AsyncMock( # type: ignore[method-assign] + side_effect=RuntimeError("kaboom") + ) + ack = AsyncMock() + payload = { + "type": "view_submission", + "team": {"id": "T1"}, + "user": {"id": "U1", "name": "x"}, + "view": {"id": "V1", "callback_id": "cb", "private_metadata": "", "state": {"values": {}}}, + "trigger_id": "trig", + } + await adapter._route_socket_event(payload, "interactive", ack) + assert ack.await_count == 1 + ack_args = ack.call_args + assert ack_args.args, "ack must be called with a response payload, not empty" + body_arg = ack_args.args[0] + assert isinstance(body_arg, dict) + assert body_arg.get("response_action") == "errors" + assert "errors" in body_arg + + +class TestSocketEventsApiPayloadParity: + """Regression for review finding #7. + + 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. + """ + + async def test_synthesized_payload_does_not_include_is_ext_shared_channel(self): + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + + # Capture the payload handed to _process_event_payload. + captured: list[dict[str, Any]] = [] + + def fake_process(payload: dict[str, Any], _options: Any = None) -> None: + captured.append(payload) + + adapter._process_event_payload = fake_process # type: ignore[method-assign] + ack = AsyncMock() + body = { + "team_id": "T1", + "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", + }, + } + 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 From e6e337fb259478f01740ccb33588ac0ce57aacb3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 9 May 2026 20:32:22 +0000 Subject: [PATCH 4/9] fix(slack): forwarded-event freshness accepts JS Date.now() (ms) timestamps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR #86 re-review M1 (critical interop break). Upstream's forwardSocketEvent always emits ``timestamp: Date.now()`` — milliseconds since epoch. Python's ``time.time()`` returns seconds. The previous freshness check compared the two directly, treating a real JS-emitted ms timestamp (~1.78e12) as 56,000 years skewed and rejecting every real forwarded event with 401. Since startSocketModeListener was intentionally not ported (UPSTREAM_SYNC.md non-parity), the only expected producer is a JS process — making this a 100% interop break. Auto-detect the unit by magnitude: anything > 10**11 is ms (that magnitude crossed in 2001), normalize to seconds before comparing to ``time.time()``. Accepts both wire formats so a future Python-emitted listener also works. Adds two tests: - test_js_emitted_milliseconds_timestamp_accepted: a real Date.now()-shaped ms timestamp now passes freshness. - test_js_emitted_milliseconds_replay_rejected: a 6-minute-old ms timestamp is still rejected. All 33 socket-mode tests pass. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- src/chat_sdk/adapters/slack/adapter.py | 14 +++++- tests/test_slack_socket_mode.py | 61 ++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 8eacfbab..4cef243c 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -814,12 +814,24 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No # without a freshness check, an old captured forwarded event # could be replayed indefinitely. Mirror the 5-minute window # ``_verify_signature`` enforces on signed webhook traffic. + # + # Wire format: upstream's ``forwardSocketEvent`` always emits + # ``timestamp: Date.now()`` — milliseconds since the Unix epoch + # (~1.78e12 today). Python's ``time.time()`` returns seconds. + # Auto-detect the unit by magnitude (anything > 10**11 is + # certainly milliseconds — that crossed in 2001) so we accept + # both the JS-emitted ms shape AND a Python-emitted seconds + # shape if a future ``forward_socket_event`` listener lands. ts_raw = event.get("timestamp") if isinstance(event, dict) else None try: ts_int = int(ts_raw) if ts_raw is not None else None except (TypeError, ValueError): ts_int = None - if ts_int is None or abs(int(time.time()) - ts_int) > 300: + if ts_int is not None and ts_int > 10**11: + ts_seconds = ts_int // 1000 + else: + ts_seconds = ts_int + if ts_seconds is None or abs(int(time.time()) - ts_seconds) > 300: self._logger.warn( "Forwarded socket event outside freshness window", {"timestamp": ts_raw}, diff --git a/tests/test_slack_socket_mode.py b/tests/test_slack_socket_mode.py index 67a3a924..348169a7 100644 --- a/tests/test_slack_socket_mode.py +++ b/tests/test_slack_socket_mode.py @@ -819,6 +819,67 @@ async def test_missing_timestamp_rejected(self): result = await adapter.handle_webhook(request) assert result["status"] == 401 + async def test_js_emitted_milliseconds_timestamp_accepted(self): + """Upstream's ``forwardSocketEvent`` always emits ``Date.now()`` + (milliseconds since epoch). The Python receiver must accept the JS + wire format, not only the Python ``int(time.time())`` (seconds) shape. + + What to fix if this fails: in + ``src/chat_sdk/adapters/slack/adapter.py`` ``handle_webhook``, the + forwarded-event freshness check must auto-detect millisecond-shaped + timestamps (anything > 10**11 — that magnitude crossed in 2001) and + normalize to seconds before comparing to ``time.time()``. A naive + seconds-only check rejects every real upstream-emitted forward with + a ~56,000-year skew. + """ + import time as _time + + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + # JS-shaped: ``Date.now()`` returns milliseconds. + forwarded = { + "type": "socket_event", + "eventType": "events_api", + "body": {"type": "event_callback", "event": {}}, + "timestamp": int(_time.time() * 1000), # ms, like Date.now() + } + request = _FakeRequest( + json.dumps(forwarded), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "fwd-secret", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 200, ( + "Forwarded event with JS-shaped Date.now() timestamp (ms) was " + "rejected by the freshness check; the receiver must auto-detect " + "ms vs s by magnitude" + ) + + async def test_js_emitted_milliseconds_replay_rejected(self): + """A 6-minute-old JS-shaped (ms) timestamp must still be rejected.""" + import time as _time + + adapter = _make_socket_adapter() + adapter._chat = _make_mock_chat() + forwarded = { + "type": "socket_event", + "eventType": "events_api", + "body": {"type": "event_callback", "event": {}}, + # 6 minutes old, in milliseconds + "timestamp": int((_time.time() - 6 * 60) * 1000), + } + request = _FakeRequest( + json.dumps(forwarded), + headers={ + "content-type": "application/json", + "x-slack-socket-token": "fwd-secret", + }, + ) + result = await adapter.handle_webhook(request) + assert result["status"] == 401 + class TestInteractiveDispatchErrorAck: """Regression for review finding #5. From a8cc706646914cfe918066a3d5acfa1e5ac8574e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 9 May 2026 20:32:45 +0000 Subject: [PATCH 5/9] style(slack): collapse ms-detect to ternary (ruff SIM108) --- src/chat_sdk/adapters/slack/adapter.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 4cef243c..294d0a60 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -827,10 +827,7 @@ async def handle_webhook(self, request: Any, options: WebhookOptions | None = No ts_int = int(ts_raw) if ts_raw is not None else None except (TypeError, ValueError): ts_int = None - if ts_int is not None and ts_int > 10**11: - ts_seconds = ts_int // 1000 - else: - ts_seconds = ts_int + ts_seconds = ts_int // 1000 if ts_int is not None and ts_int > 10**11 else ts_int if ts_seconds is None or abs(int(time.time()) - ts_seconds) > 300: self._logger.warn( "Forwarded socket event outside freshness window", From b312f36fed18a547e7615d36f3b7c26d57c4f365 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 15 May 2026 21:38:39 +0000 Subject: [PATCH 6/9] docs(slack): address PR #86 re-review nits (N1, N3-N5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * N1: comment near ``_process_event_payload`` is_ext_shared_channel branch noting socket-mode payloads never carry this field — mirrors upstream's ``routeSocketEvent`` shape, documented as a known divergence so it isn't quietly re-introduced on a future port. * N3: drop the dead ``is not None else 30.0`` ternary; the config field is typed ``float`` with a 30s default so the fallback is unreachable. * N4: comment on ``asyncio.wait_for(asyncio.shield(asyncio.wait(...)))`` explaining why the shielded inner wait doesn't leak — both inner tasks get cancelled in the timeout branch. * N5: comment near the ``slash_commands`` ack noting slash responses are out-of-band via ``response_url``, not the WebSocket ack. Comment-only / dead-code-only changes; no behavior change. All 33 socket-mode tests pass; full suite 3701 passed (1 pre-existing failure). https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- src/chat_sdk/adapters/slack/adapter.py | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 294d0a60..9a7de1f7 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -252,10 +252,10 @@ def __init__(self, config: SlackAdapterConfig | None = None) -> None: self._socket_initial_backoff_s = 1.0 self._socket_max_backoff_s = 30.0 # Bound the initial Socket Mode handshake so ``initialize()`` doesn't - # block forever if slack_sdk's ``connect()`` hangs (hazard #11). - self._socket_connect_timeout_s: float = ( - config.connect_timeout_s if config.connect_timeout_s is not None else 30.0 - ) + # block forever if slack_sdk's ``connect()`` hangs (hazard #11). The + # config field is typed ``float`` with a 30s default, so this is just + # a read. + self._socket_connect_timeout_s: float = config.connect_timeout_s self._logger: Logger = config.logger or ConsoleLogger("info") self._user_name: str = config.user_name or "bot" self._bot_user_id: str | None = config.bot_user_id or None @@ -964,7 +964,11 @@ def _process_event_payload(self, payload: dict[str, Any], options: WebhookOption event: dict[str, Any] = payload["event"] - # Track external/shared channel status + # 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``. 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: @@ -1405,6 +1409,12 @@ async def start_socket_mode(self) -> None: # (hazard #11). wait_task = asyncio.create_task(connected.wait()) try: + # ``shield`` keeps the inner ``asyncio.wait`` alive when + # ``wait_for`` times out, so we can deterministically tear + # ``wait_task`` and ``_socket_task`` down ourselves in the + # except branch. Both tasks are cancelled there (``wait_task`` + # directly, ``_socket_task`` via ``stop_socket_mode``), so the + # shielded inner wait always resolves shortly after — no orphan. first_done, pending = await asyncio.wait_for( asyncio.shield( asyncio.wait( @@ -1674,6 +1684,11 @@ def _log_exc(t: asyncio.Task[Any]) -> None: return if event_type == "slash_commands": + # Slash responses are out-of-band via ``response_url`` (Slack's + # delayed-response pattern), not the WebSocket ack. The empty ack + # here just tells Slack we received the command; the body of the + # reply flows through ``_handle_slash_command`` posting to the + # response_url. Matches upstream's `routeSocketEvent`. await ack() # slash_commands payload is a flat dict mirroring the # form-urlencoded fields; convert to the parse_qs shape that From d4a0cdd5a4af7fa9943f3b9fafeccf630af60e16 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 16 May 2026 01:25:40 +0000 Subject: [PATCH 7/9] docs(slack): explain load-bearing await inside contextlib.suppress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ``github-code-quality`` and other static analyzers periodically flag the ``await wait_task`` / ``await task`` inside ``with contextlib.suppress(asyncio.CancelledError):`` blocks in ``_socket_connect_with_timeout`` (~L1432) and ``stop_socket_mode`` (~L1471) as "Statement has no effect" — they model ``await`` syntactically rather than as a side-effecting suspension that drains a cancelled task. The pattern is intentional and load-bearing: - After ``task.cancel()`` we MUST ``await`` the task to deterministically drain it. Without the await, the cancellation request is queued but the function returns before the task actually tears down. - For ``stop_socket_mode`` this breaks ``test_stop_idempotent`` (a second call can race the first's cleanup) and risks "Task was destroyed but it is pending!" warnings from the asyncio loop. - ``contextlib.suppress(asyncio.CancelledError)`` absorbs the expected ``CancelledError`` raised by the awaited task on cancel. The same false-positive pattern was flagged and explained on earlier commits — the inline comments now make the intent visible at the source so future readers (human or bot) don't have to re-discover it via review history. Tests: 33 socket-mode tests pass, ruff clean. --- src/chat_sdk/adapters/slack/adapter.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 9a7de1f7..73b94254 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -1429,6 +1429,18 @@ async def start_socket_mode(self) -> None: # tear them down before surfacing the failure. wait_task.cancel() with contextlib.suppress(asyncio.CancelledError): + # ``await`` is load-bearing here: it drains the cancelled + # task before this function returns so the asyncio loop + # doesn't emit "Task was destroyed but it is pending!" + # warnings at shutdown, and so callers can rely on a + # synchronous "task fully gone" contract after timeout. + # ``contextlib.suppress`` absorbs the expected + # ``CancelledError`` that wait_task raises on cancel. + # Static analyzers (github-code-quality, etc.) sometimes + # flag this as "statement has no effect" because they + # model ``await`` syntactically rather than as a + # side-effecting suspension — that's a false positive; + # do not remove this line. await wait_task await self.stop_socket_mode() raise TimeoutError(f"Slack Socket Mode connect timed out after {self._socket_connect_timeout_s}s") from None @@ -1468,6 +1480,19 @@ async def stop_socket_mode(self) -> None: # Cancellation is expected on shutdown; surface anything else so # surprising loop crashes aren't silently swallowed. with contextlib.suppress(asyncio.CancelledError): + # ``await`` is load-bearing: deterministically drains the + # cancelled loop task before ``stop_socket_mode()`` + # returns. Without it, ``stop_socket_mode`` can return + # while the loop task is still tearing down, which: + # - breaks ``test_stop_idempotent`` (the second call + # can race the first's cleanup) + # - risks "Task was destroyed but it is pending!" + # warnings if the loop holds GC-only references + # ``contextlib.suppress`` absorbs the expected + # ``CancelledError``. Static analyzers sometimes flag + # this as "statement has no effect" because they model + # ``await`` syntactically — that's a false positive; do + # not remove this line. await task if task is not None: self._logger.info("Slack socket mode disconnected") From ab8cd26d5b7909c708c49764d17df738031ad789 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 22 May 2026 03:46:49 +0000 Subject: [PATCH 8/9] fix(slack): point Socket Mode import-error at chat-sdk[slack-socket] extra The aiohttp transport lives behind chat-sdk[slack-socket] (which adds aiohttp on top of slack-sdk), not chat-sdk[slack]. Following the old hint left users stuck because aiohttp was still missing after reinstall. Adds a regression test covering the import-error pathway so the hint can't drift back to the wrong extras name. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- src/chat_sdk/adapters/slack/adapter.py | 3 ++- tests/test_slack_socket_mode.py | 28 ++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 73b94254..c68b0c5f 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -1389,7 +1389,8 @@ async def start_socket_mode(self) -> None: except ImportError as exc: # pragma: no cover - import-time failure raise ValidationError( "slack", - "slack_sdk is not installed. Install with `pip install chat-sdk[slack]`.", + "slack_sdk Socket Mode dependencies are not installed. " + "Install with `pip install chat-sdk[slack-socket]`.", ) from exc self._socket_shutdown_event.clear() diff --git a/tests/test_slack_socket_mode.py b/tests/test_slack_socket_mode.py index 348169a7..10d5e3a6 100644 --- a/tests/test_slack_socket_mode.py +++ b/tests/test_slack_socket_mode.py @@ -465,6 +465,34 @@ async def test_unknown_event_type_acks_and_does_nothing(self): # --------------------------------------------------------------------------- +class TestSocketModeImportError: + """Regression: the import-error hint must point at the correct extra. + + The aiohttp transport lives behind the ``chat-sdk[slack-socket]`` extra + (which pulls in both ``slack-sdk`` and ``aiohttp``), NOT the plain + ``chat-sdk[slack]`` extra. If the message ever drifts back to ``[slack]`` + users following the hint end up in a broken loop because ``aiohttp`` + is still missing after reinstall. + """ + + async def test_missing_aiohttp_transport_message_points_at_slack_socket_extra(self, monkeypatch): + import slack_sdk.socket_mode.aiohttp as sm + + # Simulate ``aiohttp`` (and therefore the transport submodule) being + # absent at import time. + monkeypatch.delattr(sm, "SocketModeClient", raising=False) + monkeypatch.delitem(sys.modules, "slack_sdk.socket_mode.aiohttp", raising=False) + + adapter = _make_socket_adapter() + with pytest.raises(ValidationError) as exc_info: + await adapter.start_socket_mode() + + msg = str(exc_info.value) + assert "chat-sdk[slack-socket]" in msg, msg + # Guard against regressing back to the wrong extra name. + assert "chat-sdk[slack]`" not in msg, msg + + class _FakeSocketModeClient: """In-process stand-in for slack_sdk's SocketModeClient.""" From cf67818232ba09bd8c1a40e83ab511bae7f6b02f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 22 May 2026 05:25:36 +0000 Subject: [PATCH 9/9] fix(slack): drop stale `pragma: no cover` on Socket Mode import-error path PR #86 commit ab8cd26 added ``test_missing_aiohttp_transport_message_points_at_slack_socket_extra`` which exercises this branch directly, so the ``# pragma: no cover - import-time failure`` marker is stale. Removing it keeps coverage honest and prevents future regressions in the hint-message wording from silently slipping past coverage gating. Self-review (PR #86): no behavior change, just the inline marker. https://claude.ai/code/session_01FyMxQn2BEAzmwKS1GZczKj --- src/chat_sdk/adapters/slack/adapter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index c68b0c5f..310cda0e 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -1386,7 +1386,7 @@ async def start_socket_mode(self) -> None: # Lazy import (hazard #10) — slack_sdk is an optional dependency. try: from slack_sdk.socket_mode.aiohttp import SocketModeClient # noqa: F401 - except ImportError as exc: # pragma: no cover - import-time failure + except ImportError as exc: raise ValidationError( "slack", "slack_sdk Socket Mode dependencies are not installed. "