feat(slack): support standard events and interactive callbacks

src/langbot/libs/slack_api/api.py

@@ -1,6 +1,8 @@
 import asyncio
 import json
 import traceback
+from urllib.parse import parse_qs
+from slack_sdk.signature import SignatureVerifier
 from quart import Quart, jsonify, request
 from slack_sdk.web.async_client import AsyncWebClient
 from .slackevent import SlackEvent
@@ -64,13 +66,20 @@ class SlackClient:
             body = await req.get_data()
             if len(body) > _MAX_CALLBACK_BODY_BYTES:
                 raise ValueError('Slack callback body exceeds the size limit')
-            data = await asyncio.to_thread(json.loads, body)
+            if not SignatureVerifier(self.signing_secret).is_valid_request(body, dict(req.headers)):
+                return jsonify({'error': 'Invalid Slack signature'}), 401
+            if req.mimetype == 'application/x-www-form-urlencoded':
+                data = json.loads(parse_qs(body.decode()).get('payload', ['{}'])[0])
+            else:
+                data = json.loads(body)
             if 'type' in data:
                 if data['type'] == 'url_verification':
                     return data['challenge']
 
-            await self.handle_payload(data)
-            return jsonify({'status': 'ok'})
+            self._start_worker()
+            if not self._enqueue(data, data.get('event_id')):
+                return jsonify({'error': 'Slack event queue is full'}), 503
+            return '', 200
 
         except Exception as e:
             await self.logger.error(f'Error in handle_callback_request: {traceback.format_exc()}')
@@ -79,29 +88,64 @@ class SlackClient:
     async def handle_payload(self, data):
         """Dispatch the same event representation for HTTP and Socket Mode."""
         raw = data.get('event', {})
+        for authorization in data.get('authorizations', []):
+            if not self.bot_user_id and authorization.get('is_bot') and authorization.get('user_id'):
+                self.bot_user_id = authorization['user_id']
+                break
+        if data.get('type') in {'block_actions', 'view_submission', 'view_closed', 'shortcut', 'message_action'}:
+            await self._handle_message(SlackEvent(data), 'event')
+            return
         if raw.get('bot_id') or raw.get('subtype') == 'bot_message':
             return
-        if raw.get('channel_type') == 'im':
+        kind, subtype = raw.get('type'), raw.get('subtype')
+        if kind == 'message' and subtype not in (None, 'file_share', 'thread_broadcast'):
+            await self._handle_message(SlackEvent(data), 'event')
+        elif raw.get('channel_type') == 'im' and kind == 'message':
             await self._handle_message(SlackEvent.from_payload(data))
-        elif raw.get('type') == 'app_mention':
+        elif kind == 'app_mention':
             raw['channel_type'] = 'channel'
             await self._handle_message(SlackEvent.from_payload(data))
+        elif kind != 'message':
+            await self._handle_message(SlackEvent(data), 'event')
+
+    def _enqueue(self, payload, key=None):
+        if key and key in self.socket_seen:
+            return True
+        try:
+            self.socket_queue.put_nowait(payload)
+        except asyncio.QueueFull:
+            return False
+        if key:
+            self.socket_seen[key] = True
+            while len(self.socket_seen) > 4096:
+                self.socket_seen.popitem(last=False)
+        return True
+
+    def _start_worker(self):
+        if self.socket_worker is not None and not self.socket_worker.done():
+            return
+
+        async def worker():
+            while True:
+                payload = await self.socket_queue.get()
+                try:
+                    await self.handle_payload(payload)
+                except Exception:
+                    await self.logger.error('Slack event processing failed')
+                finally:
+                    self.socket_queue.task_done()
+
+        self.socket_worker = asyncio.create_task(worker())
 
     async def _socket_request(self, client, req):
         from slack_sdk.socket_mode.response import SocketModeResponse
 
-        if req.type == 'events_api':
+        if req.type in {'events_api', 'interactive'}:
             key = req.payload.get('event_id') or req.envelope_id
-            if key not in self.socket_seen:
-                try:
-                    self.socket_queue.put_nowait(req.payload)
-                except asyncio.QueueFull:
-                    # Leave unacknowledged so Slack can retry; do not grow tasks without bounds.
-                    await self.logger.warning('Slack Socket Mode event queue is full')
-                    return
-                self.socket_seen[key] = True
-                while len(self.socket_seen) > 4096:
-                    self.socket_seen.popitem(last=False)
+            if not self._enqueue(req.payload, key):
+                # Leave unacknowledged so Slack can retry; do not grow tasks without bounds.
+                await self.logger.warning('Slack Socket Mode event queue is full')
+                return
         await client.send_socket_mode_response(SocketModeResponse(envelope_id=req.envelope_id))
 
     async def run_socket(self, app_token):
@@ -111,17 +155,7 @@ class SlackClient:
         self.socket_client = SocketModeClient(app_token=app_token, web_client=self.client)
         self.socket_client.socket_mode_request_listeners.append(self._socket_request)
 
-        async def worker():
-            while True:
-                payload = await self.socket_queue.get()
-                try:
-                    await self.handle_payload(payload)
-                except Exception:
-                    await self.logger.error('Slack Socket Mode event processing failed')
-                finally:
-                    self.socket_queue.task_done()
-
-        self.socket_worker = asyncio.create_task(worker())
+        self._start_worker()
         try:
             await self.socket_client.connect()
             await self.logger.info('Slack Socket Mode connected')
@@ -143,11 +177,11 @@ class SlackClient:
             self.socket_queue.get_nowait()
             self.socket_queue.task_done()
 
-    async def _handle_message(self, event: SlackEvent):
+    async def _handle_message(self, event: SlackEvent, kind=None):
         """
         处理消息事件。
         """
-        msg_type = event.type
+        msg_type = kind or event.type
         if msg_type in self._message_handlers:
             for handler in self._message_handlers[msg_type]:
                 await handler(event)
@@ -167,7 +201,7 @@ class SlackClient:
         try:
             response = await self.client.chat_postMessage(channel=channel_id, text=text)
             if self.bot_user_id is None and response.get('ok'):
-                self.bot_user_id = response.get('message', {}).get('bot_id')
+                self.bot_user_id = response.get('message', {}).get('user')
             return response.data if hasattr(response, 'data') else response
         except Exception as e:
             await self.logger.error(f'Error in send_message: {e}')
@@ -177,7 +211,7 @@ class SlackClient:
         try:
             response = await self.client.chat_postMessage(channel='@' + user_id, text=text)
             if self.bot_user_id is None and response.get('ok'):
-                self.bot_user_id = response.get('message', {}).get('bot_id')
+                self.bot_user_id = response.get('message', {}).get('user')
 
             return response.data if hasattr(response, 'data') else response
         except Exception as e:

src/langbot/libs/slack_api/slackevent.py

@@ -65,7 +65,8 @@ class SlackEvent(dict):
 
     @property
     def message_id(self) -> str:
-        return self.get('event_id', '')
+        raw = self.get('event', {})
+        return raw.get('deleted_ts') or raw.get('message', {}).get('ts') or raw.get('ts', '')
 
     @property
     def pic_url(self) -> str:

src/langbot/pkg/api/http/service/slack_setup.py

@@ -47,7 +47,30 @@ def manifest(name: str, webhook_url: str, redirect_url: str, *, events: bool, so
     if events:
         settings['event_subscriptions'] = {
             **({} if socket_mode else {'request_url': webhook_url}),
-            'bot_events': ['app_mention', 'message.im'],
+            'bot_events': [
+                'app_mention',
+                'message.im',
+                'message.channels',
+                'message.groups',
+                'message.mpim',
+                'reaction_added',
+                'reaction_removed',
+                'member_joined_channel',
+                'member_left_channel',
+                'channel_rename',
+                'group_rename',
+                'channel_archive',
+                'channel_unarchive',
+                'group_archive',
+                'group_unarchive',
+                'app_home_opened',
+                'app_uninstalled',
+                'tokens_revoked',
+            ],
+        }
+        settings['interactivity'] = {
+            'is_enabled': True,
+            **({} if socket_mode else {'request_url': webhook_url}),
         }
     return {
         'display_information': {'name': name},
@@ -60,6 +83,7 @@ def manifest(name: str, webhook_url: str, redirect_url: str, *, events: bool, so
             'scopes': {
                 'bot': [
                     'app_mentions:read',
+                    'reactions:read',
                     'chat:write',
                     'channels:history',
                     'channels:read',

src/langbot/pkg/platform/adapters/slack/adapter.py

@@ -33,7 +33,7 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap
         typing.Type[platform_events.Event],
         typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None],
     ] = {}
-    _message_cache: dict[str, platform_events.MessageReceivedEvent] = {}
+    _message_cache: dict[tuple[str, str], platform_events.MessageReceivedEvent] = {}
     _user_cache: dict[str, platform_entities.User] = {}
     _group_cache: dict[str, platform_entities.UserGroup] = {}
     _member_cache: dict[tuple[str, str], platform_entities.UserGroupMember] = {}
@@ -74,6 +74,14 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap
     def get_supported_events(self) -> list[str]:
         return [
             'message.received',
+            'message.edited',
+            'message.deleted',
+            'message.reaction',
+            'group.member_joined',
+            'group.member_left',
+            'group.info_updated',
+            'bot.invited_to_group',
+            'bot.removed_from_group',
             'platform.specific',
         ]
 
@@ -111,8 +119,15 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap
             raise ValueError('Slack reply_message requires a SlackEvent source object')
         target_type = 'channel' if source.type == 'channel' else 'person'
         target_id = source.channel_id if source.type == 'channel' else source.user_id
-        raw = await send_chain(self.bot, target_type, target_id, message)
-        return platform_events.MessageResult(message_id=source.message_id, raw=raw)
+        thread_ts = source.get('event', {}).get('thread_ts') or (source.message_id if quote_origin else None)
+        if thread_ts:
+            target_type, target_id = 'channel', source.channel_id
+        raw = await send_chain(self.bot, target_type, target_id, message, thread_ts)
+        responses = raw.get('responses', [])
+        sent = responses[-1] if responses else {}
+        return platform_events.MessageResult(
+            message_id=sent.get('ts') or sent.get('message', {}).get('ts', ''), raw=raw
+        )
 
     async def call_platform_api(self, action: str, params: dict = {}) -> dict:
         handler = PLATFORM_API_MAP.get(action)
@@ -144,6 +159,9 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap
         return await self.bot.handle_unified_webhook(request)
 
     async def run_async(self):
+        identity = await self.bot.client.auth_test()
+        self.bot_account_id = identity.get('user_id', '')
+        self.bot.bot_user_id = self.bot_account_id
         if self.config.get('socket_mode', False):
             await self.bot.run_socket(self.config['app_token'])
             return
@@ -163,11 +181,30 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap
         return False
 
     def _register_native_handlers(self):
-        for msg_type in ('im', 'channel'):
+        for msg_type in ('im', 'channel', 'event'):
             self.bot.on_message(msg_type)(self._handle_native_event)
 
     async def _handle_native_event(self, event: SlackEvent):
         try:
+            self.event_converter.bot_user_id = getattr(self.bot, 'bot_user_id', None) or self.bot_account_id
+            raw = event.get('event', {})
+            channel = raw.get('channel') or raw.get('item', {}).get('channel', '')
+            if (
+                isinstance(channel, str)
+                and channel.startswith('D')
+                and channel not in self.event_converter.private_chats
+                and (
+                    raw.get('subtype') in {'message_changed', 'message_deleted'}
+                    or raw.get('type') in {'reaction_added', 'reaction_removed'}
+                )
+            ):
+                try:
+                    conversation = await self.bot.client.conversations_info(channel=channel)
+                    self.event_converter.remember_private_chat(channel, conversation.get('channel', {}).get('user', ''))
+                except Exception:
+                    await self.logger.warning('Slack DM participant lookup failed; using channel ID for this event')
+            if event.get('event', {}).get('type') in {'app_uninstalled', 'tokens_revoked'}:
+                await self.logger.warning(f'Slack application lifecycle event: {event["event"]["type"]}')
             if platform_events.FriendMessage in self.listeners or platform_events.GroupMessage in self.listeners:
                 legacy_event = await self.event_converter.target2legacy(event)
                 if legacy_event and type(legacy_event) in self.listeners:
@@ -190,7 +227,7 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap
     def _cache_event(self, event: platform_events.Event):
         if not isinstance(event, platform_events.MessageReceivedEvent):
             return
-        self._message_cache[str(event.message_id)] = event
+        self._message_cache[(str(event.chat_id), str(event.message_id))] = event
         self._user_cache[str(event.sender.id)] = event.sender
         if event.group:
             self._group_cache[str(event.group.id)] = event.group

src/langbot/pkg/platform/adapters/slack/api_impl.py

@@ -9,7 +9,7 @@ from langbot_plugin.api.entities.builtin.platform import message as platform_mes
 
 
 class SlackAPIMixin:
-    _message_cache: dict[str, platform_events.MessageReceivedEvent]
+    _message_cache: dict[tuple[str, str], platform_events.MessageReceivedEvent]
     _user_cache: dict[str, platform_entities.User]
     _group_cache: dict[str, platform_entities.UserGroup]
     _member_cache: dict[tuple[str, str], platform_entities.UserGroupMember]
@@ -20,7 +20,7 @@ class SlackAPIMixin:
         chat_id: typing.Union[int, str],
         message_id: typing.Union[int, str],
     ) -> platform_events.MessageReceivedEvent:
-        event = self._message_cache.get(str(message_id))
+        event = self._message_cache.get((str(chat_id), str(message_id)))
         if event is None:
             raise NotSupportedError('get_message:message_not_cached')
         return event

src/langbot/pkg/platform/adapters/slack/event_converter.py

@@ -12,8 +12,16 @@ from langbot_plugin.api.entities.builtin.platform import events as platform_even
 
 
 class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter):
-    def __init__(self, bot_token: str = ''):
+    def __init__(self, bot_token: str = '', bot_user_id: str = ''):
         self.bot_token = bot_token
+        self.bot_user_id = bot_user_id
+        self.private_chats: dict[str, str] = {}
+
+    def remember_private_chat(self, channel_id: str, user_id: str):
+        if channel_id and user_id:
+            self.private_chats[channel_id] = user_id
+            while len(self.private_chats) > 4096:
+                self.private_chats.pop(next(iter(self.private_chats)))
 
     @staticmethod
     async def yiri2target(event: platform_events.Event) -> typing.Any:
@@ -54,9 +62,93 @@ class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter):
         )
 
     async def target2yiri(self, event: SlackEvent) -> platform_events.Event:
-        if event.type in {'im', 'channel'}:
+        raw = event.get('event', {})
+        kind = raw.get('type', '')
+        subtype = raw.get('subtype', '')
+        common = dict(
+            adapter_name=ADAPTER_NAME,
+            timestamp=_timestamp_value(event),
+            source_platform_object=SlackEvent(_public_payload(dict(event))),
+        )
+        channel = raw.get('channel', '') or raw.get('item', {}).get('channel', '')
+        channel_info = channel if isinstance(channel, dict) else {'id': channel}
+        channel_id = channel_info.get('id', '')
+        group = platform_entities.UserGroup(id=channel_id, name=channel_info.get('name', channel_id))
+
+        def user(uid):
+            return platform_entities.User(id=uid, nickname=uid) if uid else None
+
+        private = raw.get('channel_type') == 'im' or str(channel_id).startswith('D')
+        chat = dict(
+            chat_type=platform_entities.ChatType.PRIVATE if private else platform_entities.ChatType.GROUP,
+            chat_id=self.private_chats.get(channel_id, channel_id) if private else channel_id,
+            group=None if private else group,
+        )
+        if kind == 'message' and subtype == 'message_changed':
+            message = raw.get('message', {})
+            updated = SlackEvent({**event, 'event': {**raw, **message}})
+            return platform_events.MessageEditedEvent(
+                **common,
+                **chat,
+                message_id=message.get('ts', ''),
+                editor=user(message.get('edited', {}).get('user') or message.get('user')) or user('unknown'),
+                new_content=await SlackMessageConverter.target2yiri(updated, self.bot_token),
+            )
+        if kind == 'message' and subtype == 'message_deleted':
+            return platform_events.MessageDeletedEvent(**common, **chat, message_id=raw.get('deleted_ts', ''))
+        if kind in {'reaction_added', 'reaction_removed'} and raw.get('item', {}).get('type') == 'message':
+            return platform_events.MessageReactionEvent(
+                **common,
+                **chat,
+                message_id=raw['item'].get('ts', ''),
+                user=user(raw.get('user')) or user('unknown'),
+                reaction=raw.get('reaction', ''),
+                is_add=kind == 'reaction_added',
+            )
+        if kind in {'member_joined_channel', 'member_left_channel'}:
+            member = user(raw.get('user')) or user('unknown')
+            inviter = user(raw.get('inviter'))
+            if self.bot_user_id and raw.get('user') == self.bot_user_id:
+                if kind == 'member_joined_channel' and inviter:
+                    return platform_events.BotInvitedToGroupEvent(**common, group=group, inviter=inviter)
+                if kind == 'member_left_channel':
+                    return platform_events.BotRemovedFromGroupEvent(**common, group=group)
+            if kind == 'member_joined_channel':
+                return platform_events.MemberJoinedEvent(
+                    **common,
+                    group=group,
+                    member=member,
+                    inviter=inviter,
+                    join_type='invite' if inviter else 'direct',
+                )
+            return platform_events.MemberLeftEvent(**common, group=group, member=member)
+        fields = {
+            'channel_rename': 'name',
+            'group_rename': 'name',
+            'channel_archive': 'archived',
+            'channel_unarchive': 'archived',
+            'group_archive': 'archived',
+            'group_unarchive': 'archived',
+            'channel_topic': 'topic',
+            'group_topic': 'topic',
+            'channel_purpose': 'purpose',
+            'group_purpose': 'purpose',
+        }
+        change = subtype if kind == 'message' else kind
+        if change in fields:
+            return platform_events.GroupInfoUpdatedEvent(
+                **common,
+                group=group,
+                operator=user(raw.get('user')),
+                changed_fields=[fields[change]],
+            )
+        if (
+            kind in {'message', 'app_mention'}
+            and subtype in {'', 'file_share', 'thread_broadcast'}
+            and event.type in {'im', 'channel'}
+        ):
             return await self.message_to_eba(event)
-        return self.platform_specific(event, f'slack.{event.type or "unknown"}')
+        return self.platform_specific(event, f'slack.{subtype or kind or event.get("type") or event.type or "unknown"}')
 
     async def message_to_eba(self, event: SlackEvent) -> platform_events.MessageReceivedEvent:
         sender_id = event.user_id or ''
@@ -67,6 +159,8 @@ class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter):
         chat_type = platform_entities.ChatType.PRIVATE
         chat_id = sender_id
         group = None
+        if event.type == 'im':
+            self.remember_private_chat(event.channel_id, sender_id)
         if event.type == 'channel':
             chat_type = platform_entities.ChatType.GROUP
             chat_id = event.channel_id or ''
@@ -82,7 +176,7 @@ class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter):
             chat_id=chat_id or '',
             group=group,
             timestamp=_timestamp_value(event),
-            source_platform_object=event,
+            source_platform_object=SlackEvent(_public_payload(dict(event))),
         )
 
     @staticmethod
@@ -91,15 +185,28 @@ class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter):
             type='platform.specific',
             adapter_name=ADAPTER_NAME,
             action=action,
-            data=dict(event),
+            data=_public_payload(dict(event)),
             timestamp=_timestamp_value(event),
-            source_platform_object=event,
+            source_platform_object=SlackEvent(_public_payload(dict(event))),
         )
 
 
 def _timestamp_value(event: SlackEvent) -> float:
-    raw_ts = event.get('event', {}).get('ts') or event.get('event', {}).get('event_ts')
+    raw_ts = event.get('event', {}).get('event_ts') or event.get('event', {}).get('ts') or event.get('event_time')
     try:
         return float(raw_ts)
     except (TypeError, ValueError):
         return time.time()
+
+
+def _public_payload(value):
+    """Keep callback credentials out of event data and monitoring records."""
+    if isinstance(value, dict):
+        return {
+            key: _public_payload(item)
+            for key, item in value.items()
+            if key not in {'token', 'response_url', 'response_urls'}
+        }
+    if isinstance(value, list):
+        return [_public_payload(item) for item in value]
+    return value

src/langbot/pkg/platform/adapters/slack/manifest.yaml

@@ -73,6 +73,14 @@ spec:
 
   supported_events:
     - message.received
+    - message.edited
+    - message.deleted
+    - message.reaction
+    - group.member_joined
+    - group.member_left
+    - group.info_updated
+    - bot.invited_to_group
+    - bot.removed_from_group
     - platform.specific
 
   supported_apis:

src/langbot/pkg/platform/adapters/slack/platform_api.py

@@ -5,7 +5,8 @@ import typing
 
 async def get_mode(adapter, params: dict) -> dict:
     return {
-        'webhook': True,
+        'webhook': not adapter.config.get('socket_mode', False),
+        'socket_mode': adapter.config.get('socket_mode', False),
         'bot_account_id': adapter.bot_account_id,
     }