From c05aac538f9b04103cda06fdeee92cd8852c626d Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Sat, 17 Oct 2026 15:11:29 +0800 Subject: [PATCH] fix(discord): complete streaming media and event delivery --- src/langbot/pkg/agent/runner/context_reset.py | 28 ++++ .../pkg/agent/runner/platform_tools.py | 2 +- .../pkg/api/http/service/monitoring.py | 27 +--- src/langbot/pkg/pipeline/preproc/preproc.py | 8 +- .../pkg/platform/adapters/discord/adapter.py | 143 +++++++++-------- .../adapters/discord/event_converter.py | 2 +- src/langbot/pkg/platform/botmgr.py | 16 +- src/langbot/pkg/plugin/handler.py | 8 + tests/unit_tests/agent/test_platform_tools.py | 13 ++ tests/unit_tests/pipeline/test_preproc.py | 20 ++- .../platform/test_botmgr_execution_trace.py | 53 +++++++ .../platform/test_discord_eba_adapter.py | 150 ++++++++++++++++++ .../unit_tests/platform/test_routing_rules.py | 9 +- tests/unit_tests/plugin/test_handler.py | 33 ++++ 14 files changed, 407 insertions(+), 105 deletions(-) diff --git a/src/langbot/pkg/agent/runner/context_reset.py b/src/langbot/pkg/agent/runner/context_reset.py index 73d6b3fd4..1ee3b2d1c 100644 --- a/src/langbot/pkg/agent/runner/context_reset.py +++ b/src/langbot/pkg/agent/runner/context_reset.py @@ -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 diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 5b7dc6509..b74ab8041 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -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( diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 335d334c6..0ca23d7e1 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -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( diff --git a/src/langbot/pkg/pipeline/preproc/preproc.py b/src/langbot/pkg/pipeline/preproc/preproc.py index 2a0878170..37c5d28b8 100644 --- a/src/langbot/pkg/pipeline/preproc/preproc.py +++ b/src/langbot/pkg/pipeline/preproc/preproc.py @@ -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]' diff --git a/src/langbot/pkg/platform/adapters/discord/adapter.py b/src/langbot/pkg/platform/adapters/discord/adapter.py index 8ace25ba7..374c008cf 100644 --- a/src/langbot/pkg/platform/adapters/discord/adapter.py +++ b/src/langbot/pkg/platform/adapters/discord/adapter.py @@ -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) diff --git a/src/langbot/pkg/platform/adapters/discord/event_converter.py b/src/langbot/pkg/platform/adapters/discord/event_converter.py index d34839725..e8bb49d4d 100644 --- a/src/langbot/pkg/platform/adapters/discord/event_converter.py +++ b/src/langbot/pkg/platform/adapters/discord/event_converter.py @@ -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', diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index bb3beb0e8..ae0ecdc4e 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -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 diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index 12fade67b..413ccd678 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -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( diff --git a/tests/unit_tests/agent/test_platform_tools.py b/tests/unit_tests/agent/test_platform_tools.py index a4dbf3190..160c9f625 100644 --- a/tests/unit_tests/agent/test_platform_tools.py +++ b/tests/unit_tests/agent/test_platform_tools.py @@ -207,6 +207,19 @@ def test_platform_resources_require_runner_call_permission() -> None: assert capabilities['unavailable_tools'] == [{'name': 'event_reply', 'reason': 'runner_call_permission_missing'}] +def test_interaction_resume_can_reply_only_to_a_frozen_delivery_target() -> None: + event = _event().model_copy(update={'event_type': 'interaction.submitted'}) + names = resolve_agent_platform_tool_names({}, event.event_type) + assert 'event_reply' in names + resources, _ = build_platform_tool_resources(event, names, ['detail', 'call']) + assert 'event_reply' in {item['tool_name'] for item in resources} + + event.delivery.reply_target = None + resources, capabilities = build_platform_tool_resources(event, names, ['detail', 'call']) + assert 'event_reply' not in {item['tool_name'] for item in resources} + assert {'name': 'event_reply', 'reason': 'event_target_unavailable'} in capabilities['unavailable_tools'] + + @pytest.mark.asyncio @pytest.mark.parametrize( ('name', 'params', 'api'), diff --git a/tests/unit_tests/pipeline/test_preproc.py b/tests/unit_tests/pipeline/test_preproc.py index 46cccadad..5ab2cafc6 100644 --- a/tests/unit_tests/pipeline/test_preproc.py +++ b/tests/unit_tests/pipeline/test_preproc.py @@ -234,7 +234,8 @@ class TestPreProcessorImageSegment: """Tests for image segment handling.""" @pytest.mark.asyncio - async def test_image_with_vision_model(self): + @pytest.mark.parametrize('image_source', ['base64', 'url']) + async def test_image_with_vision_model(self, image_source): """Image should be included when model supports vision.""" preproc = get_preproc_module() @@ -268,7 +269,13 @@ class TestPreProcessorImageSegment: chain = platform_message.MessageChain( [ platform_message.Plain(text='look at this'), - platform_message.Image(base64='data:image/png;base64,abc123'), + platform_message.Image( + **( + {'base64': 'data:image/png;base64,abc123'} + if image_source == 'base64' + else {'url': 'https://cdn.example/image.png'} + ) + ), ] ) query.message_chain = chain @@ -278,6 +285,15 @@ class TestPreProcessorImageSegment: assert result.result_type == preproc.entities.ResultType.CONTINUE # User message should have content assert result.new_query.user_message.content is not None + from langbot.pkg.agent.runner.query_entry_adapter import QueryEntryAdapter + + contents = [part.model_dump() for part in result.new_query.user_message.content] + attachments = QueryEntryAdapter._build_attachments(result.new_query, contents) + assert len(attachments) == 1 + if image_source == 'url': + assert attachments[0]['url'] == 'https://cdn.example/image.png' + else: + assert attachments[0].get('content') or attachments[0].get('url') @pytest.mark.asyncio async def test_image_without_vision_model(self): diff --git a/tests/unit_tests/platform/test_botmgr_execution_trace.py b/tests/unit_tests/platform/test_botmgr_execution_trace.py index 1d805f77f..d0b2b27e0 100644 --- a/tests/unit_tests/platform/test_botmgr_execution_trace.py +++ b/tests/unit_tests/platform/test_botmgr_execution_trace.py @@ -563,3 +563,56 @@ async def test_valid_interaction_success_keeps_acknowledgement_and_execution(): assert chain['outcome'] == 'success' assert {node['family'] for node in execution_nodes(records)} == {'platform_event', 'platform_api', 'runner'} await counters.shutdown() + + +@pytest.mark.parametrize('channel_id', ['channel-1', 'channel-2']) +def test_discord_reply_target_uses_channel_and_preserves_guild(channel_id): + from langbot.pkg.platform.botmgr import RuntimeBot + from langbot_plugin.api.entities.builtin.platform import entities, events, message + + event = events.MessageReceivedEvent( + message_id='message-1', + message_chain=message.MessageChain([message.Plain(text='hello')]), + sender=entities.User(id='user-1'), + chat_type=entities.ChatType.GROUP, + chat_id=channel_id, + group=entities.UserGroup(id='guild-1'), + adapter_name='discord-omni', + ) + assert RuntimeBot._infer_reply_target(event) == ( + 'group', + channel_id, + {'group_id': 'guild-1', 'chat_id': channel_id}, + ) + + +def test_group_event_without_chat_retains_group_reply_target(): + from langbot.pkg.platform.botmgr import RuntimeBot + from langbot_plugin.api.entities.builtin.platform import entities, events + + event = events.MemberJoinedEvent(group=entities.UserGroup(id='group-1'), member=entities.User(id='user-1')) + assert RuntimeBot._infer_reply_target(event) == ('group', 'group-1', {'group_id': 'group-1'}) + + +def test_agent_event_keeps_image_file_and_voice_attachments(): + from langbot.pkg.platform.botmgr import RuntimeBot + from langbot_plugin.api.entities.builtin.platform import entities, events, message + + event = events.MessageReceivedEvent( + message_id='media-1', + message_chain=message.MessageChain( + [ + message.Image(url='https://cdn.example.test/image.png'), + message.File(name='input.txt', url='https://cdn.example.test/input.txt'), + message.Voice(base64='dGVzdA=='), + ] + ), + sender=entities.User(id='user-1'), + chat_type=entities.ChatType.PRIVATE, + chat_id='channel-1', + ) + agent_input = RuntimeBot._build_agent_input(event) + assert [attachment.type for attachment in agent_input.attachments] == ['image', 'file', 'voice'] + assert agent_input.attachments[1].name == 'input.txt' + assert agent_input.attachments[1].url == 'https://cdn.example.test/input.txt' + assert agent_input.attachments[2].content == 'dGVzdA==' diff --git a/tests/unit_tests/platform/test_discord_eba_adapter.py b/tests/unit_tests/platform/test_discord_eba_adapter.py index a4bdc4b08..50521c55a 100644 --- a/tests/unit_tests/platform/test_discord_eba_adapter.py +++ b/tests/unit_tests/platform/test_discord_eba_adapter.py @@ -344,3 +344,153 @@ async def test_discord_send_and_reply_omit_empty_files_and_return_message_result assert channel.send.await_args.kwargs['content'] == 'reply' assert channel.send.await_args.kwargs['reference'] is source_message assert channel.send.await_args.kwargs['mention_author'] is False + + +def stream_adapter(): + adapter = make_adapter() + sent = SimpleNamespace(content='', edit=AsyncMock()) + sent.edit.return_value = sent + channel = SimpleNamespace(send=AsyncMock(return_value=sent)) + adapter._stream_buffer['stream'] = { + 'channel': channel, + 'sent_message': None, + 'last_content': '', + 'chunk_count': 0, + 'updated_at': 0, + } + return adapter, channel, sent + + +@pytest.mark.asyncio +async def test_discord_stream_delivers_all_text_across_message_limit(): + adapter, channel, sent = stream_adapter() + source = platform_events.MessageReceivedEvent(message_id=1) + await adapter.reply_message_chunk( + source, + {'resp_message_id': 'stream'}, + platform_message.MessageChain([platform_message.Plain(text='preview')]), + ) + sent.content = 'preview' + content = '0123456789' * 450 + 'END' + await adapter.reply_message_chunk( + source, + {'resp_message_id': 'stream'}, + platform_message.MessageChain([platform_message.Plain(text=content)]), + is_final=True, + ) + sent.edit.assert_awaited_once_with(content=content[:2000]) + remainder = ''.join(call.args[0] for call in channel.send.await_args_list[1:]) + assert content[:2000] + remainder == content + assert all(len(call.args[0]) <= 2000 for call in channel.send.await_args_list) + assert not adapter._stream_buffer + + +@pytest.mark.asyncio +async def test_discord_stream_keeps_tool_round_and_delivers_terminal_attachment(): + adapter, channel, sent = stream_adapter() + object.__setattr__(adapter, 'reply_message', AsyncMock()) + source = platform_events.MessageReceivedEvent(message_id=1) + attachment = platform_message.File(name='result.txt', base64='T0s=') + message = platform_message.MessageChain([attachment]) + await adapter.reply_message_chunk( + source, + {'resp_message_id': 'stream', 'tool_calls': [{'id': 'tool'}]}, + message, + is_final=True, + ) + assert 'stream' in adapter._stream_buffer + adapter.reply_message.assert_not_awaited() + await adapter.reply_message_chunk(source, {'resp_message_id': 'stream'}, message, is_final=True) + adapter.reply_message.assert_awaited_once() + assert adapter.reply_message.await_args.args[1].root == [attachment] + channel.send.assert_not_awaited() + assert not adapter._stream_buffer + + +@pytest.mark.asyncio +async def test_discord_stream_final_failure_is_not_reported_as_success(): + adapter, channel, _ = stream_adapter() + channel.send.side_effect = RuntimeError('delivery failed') + with pytest.raises(RuntimeError, match='delivery failed'): + await adapter.reply_message_chunk( + platform_events.MessageReceivedEvent(message_id=1), + {'resp_message_id': 'stream'}, + platform_message.MessageChain([platform_message.Plain(text='reply')]), + is_final=True, + ) + assert not adapter._stream_buffer + + +@pytest.mark.asyncio +async def test_discord_gateway_uses_one_raw_path_for_cached_events(): + adapter = make_adapter() + object.__setattr__(adapter, '_dispatch_gateway_tuple', AsyncMock()) + assert not hasattr(adapter.bot, 'on_reaction_add') + assert not hasattr(adapter.bot, 'on_reaction_remove') + assert not hasattr(adapter.bot, 'on_message_delete') + payload = SimpleNamespace(user_id=123) + await adapter.bot.on_raw_reaction_add(payload) + adapter._dispatch_gateway_tuple.assert_awaited_once_with('raw_reaction_add', payload, None) + adapter._dispatch_gateway_tuple.reset_mock() + cached = fake_message() + await adapter.bot.on_raw_message_delete(SimpleNamespace(cached_message=cached)) + adapter._dispatch_gateway_tuple.assert_awaited_once_with('message_delete', cached, None) + + +@pytest.mark.asyncio +async def test_discord_does_not_route_its_own_stream_edits(): + adapter = make_adapter() + object.__setattr__(adapter, '_dispatch_gateway_tuple', AsyncMock()) + before = fake_message(author=fake_user(bot=True)) + await adapter.bot.on_message_edit(before, before) + adapter._dispatch_gateway_tuple.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_discord_uncached_edit_fetches_message_without_duplicate_or_preview_events(): + adapter = make_adapter() + after = fake_message() + channel = SimpleNamespace(fetch_message=AsyncMock(return_value=after)) + object.__setattr__(adapter, '_get_channel', AsyncMock(return_value=channel)) + object.__setattr__(adapter, '_dispatch_gateway_tuple', AsyncMock()) + payload = SimpleNamespace(cached_message=None, data={'content': 'edited'}, channel_id=12, message_id=34) + await adapter.bot.on_raw_message_edit(payload) + channel.fetch_message.assert_awaited_once_with(34) + adapter._dispatch_gateway_tuple.assert_awaited_once_with('message_edit', (None, after), None) + adapter._dispatch_gateway_tuple.reset_mock() + channel.fetch_message.reset_mock() + payload.cached_message = after + await adapter.bot.on_raw_message_edit(payload) + payload.cached_message = None + payload.data = {'embeds': []} + await adapter.bot.on_raw_message_edit(payload) + payload.data = {'content': 'bot edit', 'author': {'bot': True}} + await adapter.bot.on_raw_message_edit(payload) + channel.fetch_message.assert_not_awaited() + adapter._dispatch_gateway_tuple.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_guild_event_reply_uses_system_channel(monkeypatch): + adapter = make_adapter() + channel = SimpleNamespace(send=AsyncMock(return_value=SimpleNamespace(id=456))) + monkeypatch.setattr(adapter.bot, 'get_guild', lambda guild_id: SimpleNamespace(system_channel=channel)) + adapter._get_channel = AsyncMock() + result = await adapter.send_message( + 'group', '123', platform_message.MessageChain([platform_message.Plain(text='Welcome')]) + ) + assert result.message_id == 456 + channel.send.assert_awaited_once_with(content='Welcome') + adapter._get_channel.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_guild_event_reply_without_system_channel_does_not_pick_another_channel(monkeypatch): + adapter = make_adapter() + monkeypatch.setattr(adapter.bot, 'get_guild', lambda guild_id: SimpleNamespace(system_channel=None)) + adapter._get_channel = AsyncMock() + with pytest.raises(ValueError, match='no system channel'): + await adapter.send_message( + 'group', '123', platform_message.MessageChain([platform_message.Plain(text='Welcome')]) + ) + adapter._get_channel.assert_not_awaited() diff --git a/tests/unit_tests/platform/test_routing_rules.py b/tests/unit_tests/platform/test_routing_rules.py index c35788a39..30455643d 100644 --- a/tests/unit_tests/platform/test_routing_rules.py +++ b/tests/unit_tests/platform/test_routing_rules.py @@ -645,7 +645,8 @@ class TestInteractionResumeRouting: send_message=AsyncMock(), ) - await bot._handle_interaction_submission(self._event(), adapter) + event = self._event() + await bot._handle_interaction_submission(event, adapter) envelope, binding, adapter_context = captured[0] assert envelope.event_type == 'interaction.submitted' @@ -653,7 +654,11 @@ class TestInteractionResumeRouting: assert binding.binding_id == 'binding-1' assert binding.processor_type == 'agent' assert binding.processor_id == 'agent-1' - assert adapter_context == {'_delivery_adapter': adapter} + assert adapter_context == { + '_delivery_adapter': adapter, + '_platform_event': event, + '_execution_context': TEST_CONTEXT, + } assert envelope.delivery.supports_streaming is False assert binding.delivery_policy.enable_reply is False adapter.send_message.assert_not_awaited() diff --git a/tests/unit_tests/plugin/test_handler.py b/tests/unit_tests/plugin/test_handler.py index 23be3490b..98dfeba04 100644 --- a/tests/unit_tests/plugin/test_handler.py +++ b/tests/unit_tests/plugin/test_handler.py @@ -5,6 +5,7 @@ Tests handler helper methods that don't require full handler setup. from __future__ import annotations +from contextlib import asynccontextmanager from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, Mock import pytest @@ -18,6 +19,38 @@ from langbot_plugin.entities.io.context import ( ) +@pytest.mark.asyncio +async def test_plugin_reset_clears_runner_history_in_the_authorized_scope(monkeypatch): + reset = AsyncMock() + monkeypatch.setattr('langbot.pkg.agent.runner.context_reset.reset_runner_context', reset) + transactions = [] + + @asynccontextmanager + async def tenant_uow(workspace): + transactions.append(workspace) + yield + + query = SimpleNamespace( + bot_uuid='bot-a', + launcher_type=SimpleNamespace(value='group'), + launcher_id='guild-a', + sender_id='user-a', + variables={}, + session=SimpleNamespace(using_conversation=SimpleNamespace(uuid=None)), + ) + app = SimpleNamespace( + query_pool=SimpleNamespace(cached_queries={'query-a': query}), + persistence_mgr=SimpleNamespace(tenant_uow=tenant_uow), + logger=SimpleNamespace(debug=Mock(), warning=Mock()), + ) + h = make_handler(app) + response = await h.actions[PluginToRuntimeAction.CREATE_NEW_CONVERSATION.value]({'query_id': 'query-a'}) + assert response.code == 0 + reset.assert_awaited_once_with(app.persistence_mgr, 'workspace-a', 'bot-a', 'group_guild-a') + assert transactions == ['workspace-a'] + assert query.session.using_conversation is None + + def make_handler(app): """Create a RuntimeConnectionHandler with mocked external connection.""" from langbot.pkg.plugin.handler import RuntimeConnectionHandler