fix(discord): complete streaming media and event delivery

2026-10-22 16:35:20 +08:00 · 2026-10-17 15:11:29 +08:00
parent bbf517edbc
commit c05aac538f
14 changed files with 407 additions and 105 deletions
@@ -2,6 +2,8 @@

 import hashlib
 import json
+import datetime
+import uuid

 import sqlalchemy as sa

@@ -14,6 +16,32 @@ def reset_scope(workspace, bot, conversation):
    return 'host-reset:' + hashlib.sha256(json.dumps([workspace, bot, conversation]).encode()).hexdigest()


+async def reset_runner_context(persistence, workspace, bot, conversation):
+    """Reset scoped Runner state and history visibility within the caller's transaction."""
+    await persistence.execute_async(
+        sa.delete(RunnerState).where(
+            RunnerState.workspace_id == workspace,
+            RunnerState.bot_id == bot,
+            RunnerState.conversation_id == conversation,
+            RunnerState.scope == 'conversation',
+        )
+    )
+    await persistence.execute_async(
+        sa.insert(RunnerState).values(
+            runner_id=HOST_RESET_RUNNER,
+            binding_identity=HOST_RESET_RUNNER,
+            scope='conversation',
+            scope_key=reset_scope(workspace, bot, conversation),
+            state_key='generation',
+            value_json=json.dumps(str(uuid.uuid4())),
+            workspace_id=workspace,
+            bot_id=bot,
+            conversation_id=conversation,
+            created_at=datetime.datetime.utcnow(),
+        )
+    )
+
+
 async def get_reset_generation(engine, workspace, bot, conversation):
    if not all((workspace, bot, conversation)):
        return None
@@ -60,7 +60,7 @@ PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = (
            'Call this tool to send a message; Agent output text is not sent automatically.',
        },
        _object_schema({'text': {**_TEXT, 'description': 'Reply text'}}, ['text']),
-        ('message.*', 'friend.*', 'group.*', 'feedback.*'),
+        ('message.*', 'friend.*', 'group.*', 'feedback.*', 'interaction.submitted'),
        'reply_target',
    ),
    PlatformToolDefinition(
@@ -2247,8 +2247,7 @@ class MonitoringService(ExecutionDetailsMixin):
    @_workspace_transaction
    async def reset_session_context(self, context: TenantContext, bot_id: str, session_id: str) -> dict:
        """Reset model context without deleting monitoring history or other sessions."""
-        from ....agent.runner.context_reset import HOST_RESET_RUNNER, reset_scope
-        from ....entity.persistence.runner_state import RunnerState
+        from ....agent.runner.context_reset import reset_runner_context

        if not isinstance(bot_id, str) or not bot_id or not isinstance(session_id, str) or not session_id:
            raise ValueError('bot_id and session_id are required')
@@ -2325,29 +2324,7 @@ class MonitoringService(ExecutionDetailsMixin):
        ).first()
        if busy or pending or waiting:
            raise RuntimeError('Session is processing a task; wait until it finishes')
-        await execute(
-            sqlalchemy.delete(RunnerState).where(
-                RunnerState.workspace_id == workspace,
-                RunnerState.bot_id == bot_id,
-                RunnerState.conversation_id == conversation,
-                RunnerState.scope == 'conversation',
-            )
-        )
-        generation = str(uuid.uuid4())
-        await execute(
-            sqlalchemy.insert(RunnerState).values(
-                runner_id=HOST_RESET_RUNNER,
-                binding_identity=HOST_RESET_RUNNER,
-                scope='conversation',
-                scope_key=reset_scope(workspace, bot_id, conversation),
-                state_key='generation',
-                value_json=json.dumps(generation),
-                workspace_id=workspace,
-                bot_id=bot_id,
-                conversation_id=conversation,
-                created_at=datetime.datetime.utcnow(),
-            )
-        )
+        await reset_runner_context(self.ap.persistence_mgr, workspace, bot_id, conversation)
        return {'reset': True, 'session_id': session_id}

    async def get_sessions(
@@ -349,8 +349,10 @@ class PreProcessor(stage.PipelineStage):
                plain_text += me.text
            elif isinstance(me, platform_message.Image):
                if keep_image_inputs:
-                    if me.base64 is not None:
+                    if me.base64:
                        content_list.append(provider_message.ContentElement.from_image_base64(me.base64))
+                    elif me.url:
+                        content_list.append(provider_message.ContentElement.from_image_url(me.url))
                else:
                    content_list.append(provider_message.ContentElement.from_text('[Image]'))
                    plain_text += '[Image]'
@@ -371,8 +373,10 @@ class PreProcessor(stage.PipelineStage):
                        content_list.append(provider_message.ContentElement.from_text(msg.text))
                    elif isinstance(msg, platform_message.Image):
                        if keep_image_inputs:
-                            if msg.base64 is not None:
+                            if msg.base64:
                                content_list.append(provider_message.ContentElement.from_image_base64(msg.base64))
+                            elif msg.url:
+                                content_list.append(provider_message.ContentElement.from_image_url(msg.url))
                        else:
                            content_list.append(provider_message.ContentElement.from_text('[Image]'))
                            plain_text += '[Image]'
@@ -92,40 +92,40 @@ class DiscordAdapter(DiscordAPIMixin, abstract_platform_adapter.AbstractPlatform
                    await adapter_self.logger.error(f'Error in Discord interaction callback: {traceback.format_exc()}')

            async def on_message_edit(self: discord.Client, before: discord.Message, after: discord.Message):
+                if after.author.bot:
+                    return
                await adapter_self._dispatch_gateway_tuple(
                    'message_edit', (before, after), self.user.id if self.user else None
                )

-            async def on_message_delete(self: discord.Client, message: discord.Message):
-                await adapter_self._dispatch_gateway_tuple(
-                    'message_delete', message, self.user.id if self.user else None
-                )
+            async def on_raw_message_edit(self: discord.Client, payload: discord.RawMessageUpdateEvent):
+                # Cached edits are handled above. Ignore embed-only updates, which
+                # Discord also emits after resolving previews.
+                if payload.cached_message is not None or not {'content', 'attachments'} & payload.data.keys():
+                    return
+                if payload.data.get('author', {}).get('bot'):
+                    return
+                try:
+                    channel = await adapter_self._get_channel(payload.channel_id)
+                    after = await channel.fetch_message(payload.message_id)
+                except discord.HTTPException:
+                    await adapter_self.logger.warning('Unable to fetch edited Discord message')
+                    return
+                if not after.author.bot:
+                    await adapter_self._dispatch_gateway_tuple(
+                        'message_edit', (None, after), self.user.id if self.user else None
+                    )

            async def on_raw_message_delete(self: discord.Client, payload: discord.RawMessageDeleteEvent):
+                # Raw callbacks also fire for cached messages. Dispatch only once,
+                # retaining the cached payload when Discord supplies one.
+                cached = getattr(payload, 'cached_message', None)
                await adapter_self._dispatch_gateway_tuple(
-                    'raw_message_delete',
-                    payload,
+                    'message_delete' if cached is not None else 'raw_message_delete',
+                    cached if cached is not None else payload,
                    self.user.id if self.user else None,
                )

-            async def on_reaction_add(
-                self: discord.Client, reaction: discord.Reaction, user: discord.User | discord.Member
-            ):
-                if self.user and user.id == self.user.id:
-                    return
-                await adapter_self._dispatch_gateway_tuple(
-                    'reaction_add', (reaction, user), self.user.id if self.user else None
-                )
-
-            async def on_reaction_remove(
-                self: discord.Client, reaction: discord.Reaction, user: discord.User | discord.Member
-            ):
-                if self.user and user.id == self.user.id:
-                    return
-                await adapter_self._dispatch_gateway_tuple(
-                    'reaction_remove', (reaction, user), self.user.id if self.user else None
-                )
-
            async def on_raw_reaction_add(self: discord.Client, payload: discord.RawReactionActionEvent):
                if self.user and payload.user_id == self.user.id:
                    return
@@ -212,7 +212,15 @@ class DiscordAdapter(DiscordAPIMixin, abstract_platform_adapter.AbstractPlatform

    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
        content, files = await self.message_converter.yiri2target(message)
-        channel = await self._get_channel(target_id)
+        guild = self.bot.get_guild(int(target_id)) if target_type == 'group' else None
+        if guild is not None:
+            # Guild-level events have no originating channel. Use only the
+            # server's explicitly configured system channel for their replies.
+            channel = guild.system_channel
+            if channel is None:
+                raise ValueError('Discord guild has no system channel configured for event replies')
+        else:
+            channel = await self._get_channel(target_id)
        kwargs = {'content': content}
        if files:
            kwargs['files'] = files
@@ -323,58 +331,55 @@ class DiscordAdapter(DiscordAPIMixin, abstract_platform_adapter.AbstractPlatform
            else getattr(bot_message, 'resp_message_id', None)
        )

-        text_parts = [m.text for m in message if isinstance(m, platform_message.Plain)]
-        chunk_text = '\n\n'.join(t for t in text_parts if t)
-
+        tool_calls = (
+            bot_message.get('tool_calls') if isinstance(bot_message, dict) else getattr(bot_message, 'tool_calls', None)
+        )
+        terminal = is_final and not tool_calls
+        chunk_text = '\n\n'.join(m.text for m in message if isinstance(m, platform_message.Plain) and m.text)
        ctx = self._stream_buffer.get(msg_id) if msg_id else None
-        if ctx is not None:
-            ctx['updated_at'] = time.time()
-
        if ctx is None:
-            try:
-                if is_final and chunk_text:
-                    await self.reply_message(message_source, message, quote_origin)
-            finally:
-                self._stream_buffer.pop(msg_id, None)
+            if terminal and message:
+                await self.reply_message(message_source, message, quote_origin)
            return

-        # Progressive streaming path: send first chunk, edit subsequent.
+        ctx['updated_at'] = time.time()
        ctx['chunk_count'] += 1
-
-        # Runner yields the full accumulated text on each chunk, so we
-        # always replace (not append).
        if chunk_text:
-            ctx['last_content'] = chunk_text[:2000]
+            ctx['last_content'] = chunk_text

-        sent = ctx['sent_message']
-
-        if sent is None:
-            # First non-empty chunk — send the initial message.
-            if not ctx['last_content']:
-                return  # No content yet, wait for next chunk.
-            try:
-                sent = await ctx['channel'].send(ctx['last_content'])
-                ctx['sent_message'] = sent
-            except Exception:
-                await self.logger.error(f'Discord stream send failed: {traceback.format_exc()}')
+        try:
+            content = ctx['last_content']
+            sent = ctx['sent_message']
+            if content:
+                # Discord limits each message to 2000 characters. Keep the first
+                # message as the live preview and deliver the remainder at completion.
+                preview = content[:2000]
+                if sent is None:
+                    sent = await ctx['channel'].send(preview)
+                    ctx['sent_message'] = sent
+                elif is_final or ctx['chunk_count'] % self._STREAM_EDIT_INTERVAL == 0:
+                    if preview != sent.content:
+                        try:
+                            ctx['sent_message'] = await sent.edit(content=preview)
+                        except discord.NotFound:
+                            ctx['sent_message'] = await ctx['channel'].send(preview)
+                if terminal:
+                    for offset in range(2000, len(content), 2000):
+                        await ctx['channel'].send(content[offset : offset + 2000])
+            if terminal:
+                attachments = platform_message.MessageChain(
+                    [
+                        part
+                        for part in message
+                        if isinstance(part, (platform_message.Image, platform_message.File, platform_message.Voice))
+                    ]
+                )
+                if attachments:
+                    await self.reply_message(message_source, attachments, quote_origin)
+        finally:
+            # Tool-call completions are intermediate rounds, not the end of a reply.
+            if terminal:
                self._stream_buffer.pop(msg_id, None)
-                return
-
-        if is_final:
-            # Final chunk — edit to the full content, then clean up.
-            if ctx['last_content'] and ctx['last_content'] != sent.content:
-                try:
-                    await sent.edit(content=ctx['last_content'][:2000])
-                except Exception:
-                    pass  # Best-effort
-            self._stream_buffer.pop(msg_id, None)
-        elif (ctx['chunk_count'] % self._STREAM_EDIT_INTERVAL) == 0:
-            # Intermediate edit — throttle to avoid rate limits.
-            if ctx['last_content'] and ctx['last_content'] != sent.content:
-                try:
-                    await sent.edit(content=ctx['last_content'][:2000])
-                except Exception:
-                    pass  # Rate-limited or deleted — ignore.

    async def run_async(self):
        await self.bot.start(self.config['token'], reconnect=True)
@@ -69,7 +69,7 @@ class DiscordEventConverter(abstract_platform_adapter.AbstractEventConverter):

    @staticmethod
    async def message_edit_to_eba(
-        before: discord.Message, after: discord.Message
+        before: discord.Message | None, after: discord.Message
    ) -> platform_events.MessageEditedEvent:
        return platform_events.MessageEditedEvent(
            type='message.edited',
@@ -640,7 +640,6 @@ class RuntimeBot:
        group_id = cls._get_entity_id(group)
        if group_id:
            metadata['group_id'] = group_id
-            return 'group', group_id, metadata

        chat_id = getattr(event, 'chat_id', None)
        chat_type = getattr(event, 'chat_type', None)
@@ -651,6 +650,11 @@ class RuntimeBot:
                return 'group', str(chat_id), metadata
            return 'person', str(chat_id), metadata

+        # A guild can contain multiple delivery channels. Keep its id for
+        # membership APIs, but prefer the event's explicit chat for replies.
+        if group_id:
+            return 'group', group_id, metadata
+
        session_target_type, session_target_id = cls._session_to_reply_target(getattr(event, 'session_id', None))
        if session_target_type and session_target_id:
            return session_target_type, session_target_id, metadata
@@ -723,6 +727,8 @@ class RuntimeBot:

    @classmethod
    def _build_agent_input(cls, event: platform_events.EBAEvent) -> AgentInput:
+        from ..agent.runner.query_entry_adapter import QueryEntryAdapter
+
        text = platform_events.event_summary(event)
        contents: list[dict[str, typing.Any]] = []

@@ -744,7 +750,7 @@ class RuntimeBot:
        return AgentInput(
            text=text,
            contents=contents,
-            attachments=[],
+            attachments=QueryEntryAdapter._build_attachments(event, contents),
        )

    def _eba_event_to_agent_envelope(
@@ -1688,7 +1694,11 @@ class RuntimeBot:
        async for _ in self.ap.agent_run_orchestrator.run(
            envelope,
            binding,
-            adapter_context={'_delivery_adapter': adapter},
+            adapter_context={
+                '_delivery_adapter': adapter,
+                '_platform_event': event,
+                '_execution_context': self.execution_context,
+            },
        ):
            # Resuming an interaction follows the same explicit-action delivery policy.
            pass
@@ -1118,6 +1118,14 @@ class RuntimeConnectionHandler(handler.Handler):
                    message=f'Query with query_id {query_id} not found',
                )

+            from ..agent.runner.context_reset import reset_runner_context
+            from ..agent.runner.query_entry_adapter import QueryEntryAdapter
+
+            conversation_id = QueryEntryAdapter._build_conversation_context(query).conversation_id
+            async with self.ap.persistence_mgr.tenant_uow(action_context.workspace_uuid):
+                await reset_runner_context(
+                    self.ap.persistence_mgr, action_context.workspace_uuid, query.bot_uuid, conversation_id
+                )
            query.session.using_conversation = None

            return handler.ActionResponse.success(