Fix migrated group response gating and failed query monitoring

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

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(

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

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)

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,

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

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'
+    )

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

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