From e212f9ead14109d1f03423649bea7aad9350906f Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Sat, 17 Oct 2026 19:42:31 +0800 Subject: [PATCH] Fix migrated group response gating and failed query monitoring Co-authored-by: multica-agent --- src/langbot/pkg/pipeline/controller.py | 20 ++++++++++++ src/langbot/pkg/pipeline/monitoring_helper.py | 10 ++++++ .../pkg/pipeline/resprule/rules/random.py | 8 ++++- src/langbot/pkg/platform/botmgr.py | 4 ++- tests/unit_tests/pipeline/test_controller.py | 32 +++++++++++++++++++ .../pipeline/test_monitoring_failure.py | 24 ++++++++++++++ tests/unit_tests/pipeline/test_resprule.py | 14 ++++++++ .../platform/test_botmgr_execution_trace.py | 14 ++++++++ 8 files changed, 124 insertions(+), 2 deletions(-) create mode 100644 tests/unit_tests/pipeline/test_monitoring_failure.py diff --git a/src/langbot/pkg/pipeline/controller.py b/src/langbot/pkg/pipeline/controller.py index 11ebeae09..161aecfeb 100644 --- a/src/langbot/pkg/pipeline/controller.py +++ b/src/langbot/pkg/pipeline/controller.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import copy import traceback from ..core import app @@ -59,6 +60,25 @@ class Controller: query.pipeline_config = pipeline.pipeline_entity.config query.variables['_pipeline_bound_plugins'] = pipeline.bound_plugins query.variables['_pipeline_bound_mcp_servers'] = pipeline.bound_mcp_servers + if getattr(getattr(query, 'launcher_type', None), 'value', None) == 'group': + from .resprule.resprule import GroupRespondRuleCheckStage + from .entities import ResultType + + # Busy sessions reach steering before the normal stage chain. + # Check on a copy: matchers strip mentions/prefixes, and a + # failed claim must leave the queued message untouched. + candidate = copy.copy(query) + candidate.message_chain = query.message_chain.model_copy(deep=True) + candidate.variables = dict(query.variables) + gate = GroupRespondRuleCheckStage(self.ap) + await gate.initialize(query.pipeline_config) + result = await gate.process(candidate, 'GroupRespondRuleCheckStage') + random_key = '_group_response_random_draw' + if random_key in candidate.variables: + query.variables[random_key] = candidate.variables[random_key] + if result.result_type == ResultType.INTERRUPT: + return False + query = candidate return await self.ap.agent_run_orchestrator.try_claim_steering_from_query(query) except Exception as exc: self.ap.logger.warning( diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index 14ff187db..1fcc6a368 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -314,6 +314,16 @@ class MonitoringHelper: ) -> str: """Record query processing error, returns message_id""" try: + # Finish the original inbound row before writing optional error + # details. Otherwise the queue can be empty while WebUI says pending. + original_message_id = query.variables.get('_monitoring_message_id') + if original_message_id: + await ap.monitoring_service.update_message_status( + get_query_execution_context(query), + message_id=original_message_id, + status='error', + level='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/resprule/rules/random.py b/src/langbot/pkg/pipeline/resprule/rules/random.py index 2bfe8b71c..4eae7cc19 100644 --- a/src/langbot/pkg/pipeline/resprule/rules/random.py +++ b/src/langbot/pkg/pipeline/resprule/rules/random.py @@ -18,4 +18,10 @@ class RandomRespRule(rule_model.GroupRespondRule): ) -> entities.RuleJudgeResult: random_rate = rule_dict['random'] - return entities.RuleJudgeResult(matching=random.random() < random_rate, replacement=message_chain) + # A busy query may be checked for steering several times before its + # normal pipeline turn. Draw once so queue wakeups do not increase + # its chance of being admitted. + key = '_group_response_random_draw' + if key not in query.variables: + query.variables[key] = random.random() + return entities.RuleJudgeResult(matching=query.variables[key] < random_rate, replacement=message_chain) diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index dfb99d292..e9198cf0f 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -1184,7 +1184,9 @@ class RuntimeBot: event, adapter, pipeline_uuid=event_binding.get('target_uuid'), - routed_by_event_binding=True, + # A catch-all binding replaces the legacy default pipeline, + # not an explicit matching rule. Preserve its response gate. + routed_by_event_binding=bool(event_binding.get('filters')), ) return await self._record_event_route_trace( event_type=event_type, diff --git a/tests/unit_tests/pipeline/test_controller.py b/tests/unit_tests/pipeline/test_controller.py index b0b10a4d6..dce3b47b5 100644 --- a/tests/unit_tests/pipeline/test_controller.py +++ b/tests/unit_tests/pipeline/test_controller.py @@ -81,6 +81,38 @@ async def test_try_claim_steering_sets_pipeline_context_before_claiming(): app.agent_run_orchestrator.try_claim_steering_from_query.assert_awaited_once_with(query) +@pytest.mark.asyncio +@pytest.mark.parametrize('text,expected', [('ordinary chat', False), ('qa: hello', True)]) +async def test_busy_group_steering_obeys_response_rules(sample_query, text, expected): + from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes, Session + from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain + + app = make_app() + app.sess_mgr.get_session.return_value = Session( + instance_uuid='test-instance', + workspace_uuid='test-workspace', + placement_generation=1, + launcher_type=LauncherTypes.GROUP, + launcher_id=sample_query.launcher_id, + ) + pipeline = make_pipeline() + pipeline.pipeline_entity.config['trigger'] = { + 'group-respond-rules': {'at': True, 'prefix': ['qa:'], 'regexp': [], 'random': 0} + } + app.pipeline_mgr.get_pipeline_by_uuid.return_value = pipeline + app.agent_run_orchestrator.try_claim_steering_from_query.return_value = True + sample_query.launcher_type = LauncherTypes.GROUP + sample_query.message_chain = MessageChain([Plain(text=text)]) + claimed = await Controller(app)._try_claim_steering_before_session_slot(sample_query) + assert claimed is expected, app.logger.warning.call_args + assert str(sample_query.message_chain) == text + if expected: + forwarded = app.agent_run_orchestrator.try_claim_steering_from_query.await_args.args[0] + assert str(forwarded.message_chain) == ' hello' + else: + app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited() + + @pytest.mark.asyncio async def test_consumer_transfers_query_from_queue_to_running_task(): app = make_app() 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..869c327f1 --- /dev/null +++ b/tests/unit_tests/pipeline/test_monitoring_failure.py @@ -0,0 +1,24 @@ +"""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' + ) diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index 63dfbfd0d..a8be87a3d 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -138,3 +138,17 @@ async def test_atbot_rule_no_match(mock_app, sample_query): result = await atbot_rule.match(str(message_chain), message_chain, {}, sample_query) assert result.matching is False + + +@pytest.mark.asyncio +async def test_random_response_draw_is_reused_across_queue_rechecks(sample_query, mock_app, monkeypatch): + from langbot.pkg.pipeline.resprule.rules.random import RandomRespRule + from unittest.mock import Mock + + draw = Mock(side_effect=[0.9, 0.1]) + monkeypatch.setattr('langbot.pkg.pipeline.resprule.rules.random.random.random', draw) + matcher = RandomRespRule(mock_app) + for _ in range(3): + result = await matcher.match('hello', sample_query.message_chain, {'random': 0.5}, sample_query) + assert result.matching is False + draw.assert_called_once() diff --git a/tests/unit_tests/platform/test_botmgr_execution_trace.py b/tests/unit_tests/platform/test_botmgr_execution_trace.py index 2002f3045..798af6124 100644 --- a/tests/unit_tests/platform/test_botmgr_execution_trace.py +++ b/tests/unit_tests/platform/test_botmgr_execution_trace.py @@ -481,6 +481,20 @@ async def test_admission_does_not_override_telemetry_opt_out(monkeypatch, editio await assert_no_upload(counters, manager) +@pytest.mark.parametrize( + 'filters,bypass', [([], False), ([{'field': 'chat_type', 'operator': 'eq', 'value': 'person'}], True)] +) +async def test_pipeline_default_route_preserves_response_gate(filters, bypass): + bot, manager, counters = make_bot([business_route(target_type='pipeline', filters=filters)]) + bot._dispatch_eba_message_to_pipeline = AsyncMock() + try: + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + bot._dispatch_eba_message_to_pipeline.assert_awaited_once() + assert bot._dispatch_eba_message_to_pipeline.await_args.kwargs['routed_by_event_binding'] is bypass + finally: + await counters.shutdown() + + async def test_unrelated_broken_subscription_does_not_poison_matched_delivery(): bot, manager, counters = make_bot([business_route(target_type='pipeline')]) bot._dispatch_eba_message_to_pipeline = AsyncMock()