diff --git a/src/langbot/pkg/pipeline/wrapper/wrapper.py b/src/langbot/pkg/pipeline/wrapper/wrapper.py index 2dcda3a05..bd0c75813 100644 --- a/src/langbot/pkg/pipeline/wrapper/wrapper.py +++ b/src/langbot/pkg/pipeline/wrapper/wrapper.py @@ -10,7 +10,6 @@ from .. import stage import langbot_plugin.api.entities.builtin.platform.message as platform_message import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query -import langbot_plugin.api.entities.builtin.provider.message as provider_message import langbot_plugin.api.entities.events as events @@ -27,50 +26,6 @@ class ResponseWrapper(stage.PipelineStage): async def initialize(self, pipeline_config: dict): pass - def _is_final_assistant_message(self, result) -> bool: - """Whether *result* is the agent's final, tool-call-free answer. - - Intermediate streaming chunks and tool-call rounds must NOT trigger - outbound attachment collection — only the terminal assistant message. - """ - if getattr(result, 'role', None) != 'assistant': - return False - if result.tool_calls: - return False - if isinstance(result, provider_message.MessageChunk): - return bool(result.is_final) - return True - - async def _append_outbound_attachments( - self, - query: pipeline_query.Query, - message_chain: platform_message.MessageChain, - ) -> None: - """Collect sandbox outbox files and append them to *message_chain*. - - Runs at most once per query (guarded by a query variable) and never - raises into the pipeline — attachment delivery is best-effort. - """ - if query.variables.get('_sandbox_outbound_collected'): - return - box_service = getattr(self.ap, 'box_service', None) - if box_service is None or not getattr(box_service, 'available', False): - return - query.variables['_sandbox_outbound_collected'] = True - try: - attachments = await box_service.collect_outbound_attachments(query) - except Exception as e: - self.ap.logger.warning(f'Outbound attachment collection failed: {e}') - return - for att in attachments: - att_type = att.get('type') - if att_type == 'Image': - message_chain.append(platform_message.Image(base64=att['base64'])) - elif att_type == 'Voice': - message_chain.append(platform_message.Voice(base64=att['base64'])) - else: - message_chain.append(platform_message.File(name=att.get('name', 'file'), base64=att['base64'])) - async def process( self, query: pipeline_query.Query, @@ -107,7 +62,7 @@ class ResponseWrapper(stage.PipelineStage): reply_text = '' - if result.content: # 有内容 + if result.content or getattr(result, 'attachments', None): reply_text = str(result.get_content_platform_message_chain()) # ============= 触发插件事件 =============== @@ -142,11 +97,6 @@ class ResponseWrapper(stage.PipelineStage): reply_chain = result.get_content_platform_message_chain() is_plugin_reply = False - # Only collect Box outbox files for a terminal assistant - # response; also accept provider final chunks here. - if self._is_final_assistant_message(result): - await self._append_outbound_attachments(query, reply_chain) - query.resp_message_chain.append(reply_chain) if is_plugin_reply: plugin_diagnostics.record_last_plugin_response_source( @@ -160,20 +110,6 @@ class ResponseWrapper(stage.PipelineStage): result_type=entities.ResultType.CONTINUE, new_query=query, ) - elif ( - isinstance(result, provider_message.MessageChunk) and result.is_final and not result.tool_calls - ): - # A blank final chunk may carry Box outbox files, but do not - # send an empty chain: streaming adapters treat it as final. - reply_chain = platform_message.MessageChain([]) - await self._append_outbound_attachments(query, reply_chain) - if len(reply_chain) > 0: - query.resp_message_chain.append(reply_chain) - yield entities.StageProcessResult( - result_type=entities.ResultType.CONTINUE, - new_query=query, - ) - if result.tool_calls is not None and len(result.tool_calls) > 0: # 有函数调用 function_names = [tc.function.name for tc in result.tool_calls] diff --git a/tests/unit_tests/pipeline/test_wrapper_outbound_attachments.py b/tests/unit_tests/pipeline/test_wrapper_outbound_attachments.py index f01d3655b..4971ec559 100644 --- a/tests/unit_tests/pipeline/test_wrapper_outbound_attachments.py +++ b/tests/unit_tests/pipeline/test_wrapper_outbound_attachments.py @@ -4,6 +4,12 @@ from langbot_plugin.api.entities.builtin.platform.message import MessageChain, F from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk import pytest +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from tests.factories import FakeApp, text_query +from langbot_plugin.api.entities.builtin.provider.session import Session, LauncherTypes + @pytest.mark.parametrize('cls', [Message, MessageChunk]) def test_explicit_output_attachments_survive_platform_conversion_only(cls): @@ -16,3 +22,46 @@ def test_explicit_output_attachments_survive_platform_conversion_only(cls): def test_file_only_output(cls): message = cls(role='assistant', attachments=MessageChain([File(name='result.txt', base64='YQ==')])) assert len(message.get_content_platform_message_chain()) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize('cls', [Message, MessageChunk]) +@pytest.mark.parametrize('content,with_file', [('answer', False), ('answer', True), ('', True)]) +async def test_wrapper_delivers_explicit_files_without_scanning_box(cls, content, with_file): + from langbot.pkg.pipeline.wrapper.wrapper import ResponseWrapper + + app = FakeApp() + app.box_service = SimpleNamespace(available=True, collect_outbound_attachments=AsyncMock(return_value=[])) + event = SimpleNamespace(reply_message_chain=None) + app.plugin_connector.emit_event = AsyncMock( + return_value=SimpleNamespace(event=event, is_prevented_default=lambda: False) + ) + query = text_query('hello') + query.pipeline_config = { + 'output': {'misc': {'at-sender': False, 'quote-origin': False, 'track-function-calls': False}} + } + message = cls( + role='assistant', + content=content, + attachments=MessageChain([File(name='result.txt', base64='YQ==')]) if with_file else None, + ) + if isinstance(message, MessageChunk): + message.is_final = True + query.resp_messages = [message] + query.resp_message_chain = [] + app.sess_mgr.get_session = AsyncMock( + return_value=Session( + launcher_type=LauncherTypes.PERSON, + launcher_id=12345, + sender_id=12345, + use_prompt_name='default', + using_conversation=None, + conversations=[], + ) + ) + stage = ResponseWrapper(app) + results = [result async for result in stage.process(query, 'ResponseWrapper')] + assert results + if with_file: + assert query.resp_message_chain[-1][-1].name == 'result.txt' + app.box_service.collect_outbound_attachments.assert_not_awaited()