diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index 1095faa7c..cb54384dd 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -6,6 +6,7 @@ without cluttering the main pipeline code. from __future__ import annotations +import asyncio import traceback import typing import time @@ -118,6 +119,25 @@ async def prepare_monitoring_identity( class MonitoringHelper: """Helper class for monitoring operations""" + @staticmethod + async def finish_query(ap, query, message_id: str, status: str) -> None: + """Bound terminal writes and retry transient failures without blocking cleanup.""" + if not message_id: + return + for attempt in range(3): + try: + async with asyncio.timeout(2): + await ap.monitoring_service.update_message_status( + get_query_execution_context(query), + message_id=message_id, + status=status, + level='error' if status == 'error' else 'info', + ) + return + except Exception as exc: + ap.logger.warning(f'Terminal monitoring write failed (attempt {attempt + 1}): {exc}') + ap.logger.error(f'Unable to finish monitoring message {message_id}: {status}') + @staticmethod async def record_query_start( ap: app.Application, @@ -283,6 +303,9 @@ class MonitoringHelper: ) -> str: """Record query processing error, returns message_id""" try: + original_message_id = (query.variables or {}).get('_monitoring_message_id') + if original_message_id: + await MonitoringHelper.finish_query(ap, query, original_message_id, 'error') session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}' # Get sender name from message event diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index fc7aaaf31..d1cfd5c62 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/src/langbot/pkg/pipeline/pipelinemgr.py @@ -497,6 +497,28 @@ class RuntimePipeline: processing_mode.reset(token) async def _process_query(self, query: pipeline_query.Query): + """Close monitoring on every exit, including cancellation and plugin veto.""" + from .monitoring_helper import MonitoringHelper + + status = 'error' + try: + await self._process_query_body(query) + if not query.variables.get('_monitoring_has_error'): + status = 'success' + finally: + try: + # Never bypass the placement fence to repair a stale execution. + async with asyncio.timeout(2): + await self._assert_execution_active(query) + await MonitoringHelper.finish_query( + self.ap, query, query.variables.get('_monitoring_message_id', ''), status + ) + except Exception as exc: + self.ap.logger.error(f'Failed to finalize query {query.query_id}: {exc}') + finally: + await self.ap.query_pool.remove_query(query) + + async def _process_query_body(self, query: pipeline_query.Query): """处理请求""" await self._assert_execution_active(query) # Get monitoring metadata @@ -602,8 +624,10 @@ class RuntimePipeline: self.ap.logger.error(f'Failed to record query response: {e}') except WorkspaceError as e: + query.variables['_monitoring_has_error'] = True self.ap.logger.info(f'Dropped query {query.query_id} because its Workspace execution binding is stale: {e}') except Exception as e: + query.variables['_monitoring_has_error'] = True inst_name = query.current_stage_name if query.current_stage_name else 'unknown' self.ap.logger.error(f'Error processing query {query.query_id} stage={inst_name} : {e}') self.ap.logger.error(f'Traceback: {traceback.format_exc()}') @@ -631,7 +655,6 @@ class RuntimePipeline: finally: self.ap.logger.debug(f'Query {query.query_id} processed') - await self.ap.query_pool.remove_query(query) def _record_lane_failure(self, query: pipeline_query.Query, reason: str) -> None: """Land one failed ``runner/execute`` stage on the owning execution trace.""" diff --git a/tests/unit_tests/pipeline/test_monitoring_failure.py b/tests/unit_tests/pipeline/test_monitoring_failure.py new file mode 100644 index 000000000..fccdd207d --- /dev/null +++ b/tests/unit_tests/pipeline/test_monitoring_failure.py @@ -0,0 +1,51 @@ +"""A stage exception must finish the original monitoring row.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from langbot.pkg.pipeline.monitoring_helper import MonitoringHelper +from langbot.pkg.pipeline.pool import get_query_execution_context + + +@pytest.mark.asyncio +async def test_exception_finishes_original_message_even_if_error_detail_write_fails(sample_query): + sample_query.variables['_monitoring_message_id'] = 'original-message' + service = SimpleNamespace( + update_message_status=AsyncMock(), record_message=AsyncMock(side_effect=RuntimeError('detail failure')) + ) + app = SimpleNamespace(monitoring_service=service, logger=Mock()) + await MonitoringHelper.record_query_error( + app, sample_query, 'bot', 'bot', 'pipeline', 'pipeline', KeyError('random') + ) + service.update_message_status.assert_awaited_once_with( + get_query_execution_context(sample_query), message_id='original-message', status='error', level='error' + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('original_id', ['original-message', None]) +async def test_error_detail_keeps_execution_context_and_original_status(sample_query, original_id): + if original_id: + sample_query.variables['_monitoring_message_id'] = original_id + service = SimpleNamespace( + update_message_status=AsyncMock(), + record_message=AsyncMock(return_value='error-detail'), + record_error=AsyncMock(), + ) + app = SimpleNamespace(monitoring_service=service, logger=Mock()) + result = await MonitoringHelper.record_query_error( + app, sample_query, 'bot', 'bot', 'pipeline', 'pipeline', ValueError('failed') + ) + assert result == 'error-detail' + context = get_query_execution_context(sample_query) + assert service.record_message.await_args.args == (context,) + assert service.record_error.await_args.args == (context,) + assert service.record_error.await_args.kwargs['message_id'] == 'error-detail' + if original_id: + service.update_message_status.assert_awaited_once_with( + context, message_id=original_id, status='error', level='error' + ) + else: + service.update_message_status.assert_not_awaited() diff --git a/tests/unit_tests/pipeline/test_terminal_monitoring.py b/tests/unit_tests/pipeline/test_terminal_monitoring.py new file mode 100644 index 000000000..cd828769d --- /dev/null +++ b/tests/unit_tests/pipeline/test_terminal_monitoring.py @@ -0,0 +1,108 @@ +"""Terminal monitoring must not prevent query cleanup.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from langbot.pkg.pipeline.monitoring_helper import MonitoringHelper +from langbot.pkg.pipeline.pipelinemgr import RuntimePipeline + + +@pytest.mark.asyncio +@pytest.mark.parametrize('failure', [None, RuntimeError('stage'), asyncio.CancelledError()]) +async def test_terminal_cleanup(sample_query, failure): + sample_query.variables['_monitoring_message_id'] = 'original' + service = SimpleNamespace(update_message_status=AsyncMock()) + app = SimpleNamespace( + monitoring_service=service, logger=Mock(), query_pool=SimpleNamespace(remove_query=AsyncMock()) + ) + runtime = SimpleNamespace( + ap=app, _assert_execution_active=AsyncMock(), _process_query_body=AsyncMock(side_effect=failure) + ) + if failure: + with pytest.raises(type(failure)): + await RuntimePipeline._process_query(runtime, sample_query) + else: + await RuntimePipeline._process_query(runtime, sample_query) + assert service.update_message_status.await_args.kwargs['status'] == ('error' if failure else 'success') + app.query_pool.remove_query.assert_awaited_once_with(sample_query) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('failures', [1, 3]) +async def test_terminal_write_retries_are_bounded(sample_query, failures): + service = SimpleNamespace(update_message_status=AsyncMock(side_effect=[RuntimeError('db')] * failures + [None])) + app = SimpleNamespace(monitoring_service=service, logger=Mock()) + await MonitoringHelper.finish_query(app, sample_query, 'original', 'error') + assert service.update_message_status.await_count == min(failures + 1, 3) + + +@pytest.mark.asyncio +async def test_stale_placement_does_not_write(sample_query): + sample_query.variables['_monitoring_message_id'] = 'original' + service = SimpleNamespace(update_message_status=AsyncMock()) + app = SimpleNamespace( + monitoring_service=service, logger=Mock(), query_pool=SimpleNamespace(remove_query=AsyncMock()) + ) + runtime = SimpleNamespace( + ap=app, _assert_execution_active=AsyncMock(side_effect=RuntimeError('stale')), _process_query_body=AsyncMock() + ) + await RuntimePipeline._process_query(runtime, sample_query) + service.update_message_status.assert_not_awaited() + app.query_pool.remove_query.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'veto,cancel_adapter,stage_error', [(True, False, False), (False, True, False), (False, False, True)] +) +async def test_real_body_early_exits(sample_query, veto, cancel_adapter, stage_error): + service = SimpleNamespace( + update_message_status=AsyncMock(), record_message=AsyncMock(return_value='original'), record_error=AsyncMock() + ) + event = SimpleNamespace(is_prevented_default=lambda: veto, event=sample_query) + app = SimpleNamespace( + monitoring_service=service, + logger=Mock(), + query_pool=SimpleNamespace(remove_query=AsyncMock()), + plugin_connector=SimpleNamespace(emit_event=AsyncMock(return_value=event)), + ) + runtime = object.__new__(RuntimePipeline) + runtime.ap = app + runtime.pipeline_entity = SimpleNamespace(uuid='pipeline') + runtime._assert_execution_active = AsyncMock() + runtime._record_lane_failure = Mock() + runtime._execute_from_stage = AsyncMock(side_effect=RuntimeError('stage') if stage_error else None) + sample_query.pipeline_config = {} + sample_query.adapter.on_monitoring_message_created = AsyncMock( + side_effect=asyncio.CancelledError() if cancel_adapter else None + ) + if cancel_adapter: + with pytest.raises(asyncio.CancelledError): + await runtime._process_query(sample_query) + else: + await runtime._process_query(sample_query) + assert service.update_message_status.await_args.kwargs['status'] == ( + 'error' if cancel_adapter or stage_error else 'success' + ) + app.query_pool.remove_query.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_error_detail_survives_status_write_failure(sample_query): + sample_query.variables['_monitoring_message_id'] = 'original' + service = SimpleNamespace( + update_message_status=AsyncMock(side_effect=RuntimeError('db')), + record_message=AsyncMock(return_value='detail'), + record_error=AsyncMock(), + ) + app = SimpleNamespace(monitoring_service=service, logger=Mock()) + assert ( + await MonitoringHelper.record_query_error( + app, sample_query, 'bot', 'bot', 'pipeline', 'pipeline', ValueError('failed') + ) + == 'detail' + ) + service.record_error.assert_awaited_once()