LANG-134: Finish original inbound monitoring row on pipeline error (#2729)

* fix(inbound-error-state): Finish original inbound monitoring row on pipeline error (LANG-134)

* fix(pipeline): finalize monitoring on early exits and cancellation

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: multica-agent <[email protected]>

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

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."""

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()

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()