From 4b370f0a92d198775974354d82d349bc5beab0fa Mon Sep 17 00:00:00 2001 From: Hyu Date: Thu, 8 Oct 2026 13:33:57 +0800 Subject: [PATCH] fix(telemetry): skip unrouted events and preserve matched deliveries --- src/langbot/pkg/platform/botmgr.py | 97 ++-- src/langbot/pkg/telemetry/execution.py | 18 +- src/langbot/pkg/telemetry/trace.py | 28 ++ src/langbot/templates/config.yaml | 12 +- .../platform/test_botmgr_execution_trace.py | 415 ++++++++++++++++-- .../platform/test_plugin_subscriptions.py | 45 ++ .../unit_tests/platform/test_routing_rules.py | 9 +- 7 files changed, 556 insertions(+), 68 deletions(-) diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index e3e0efad6..291acd1db 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -396,10 +396,15 @@ class RuntimeBot: } log_method = getattr(self.logger, level, self.logger.info) await log_method(text, metadata=metadata) - if status in {'delivered', 'failed', 'discarded', 'not_matched'}: + if status in {'delivered', 'failed'}: from ..telemetry import trace as trace_mod from ..telemetry.execution import record + state = trace_mod.current() + if status == 'failed' and state is not None and state.reportable and not trace_mod.reporting_suppressed(): + # The root event can finish successfully after a delivery failed. + # Preserve that failure for closure reporting in failures mode. + state.mark_failure(failure_code or 'route_failed') with trace_mod.scope(route_ref=self._route_ref(binding, target_type, target_uuid)): # The routing decision is its own workflow step; the Pipeline or # processor it delivers to hangs under this node. @@ -901,7 +906,13 @@ class RuntimeBot: event.bot_uuid = self.bot_entity.uuid execution_id = self._platform_event_execution_id(event) - with ingress(getattr(self, 'ap', None), 'event_done', getattr(self, 'execution_context', None), execution_id): + with ingress( + getattr(self, 'ap', None), + 'event_done', + getattr(self, 'execution_context', None), + execution_id, + require_route=True, + ): # The inbound event is the root workflow step: everything recorded # while handling it hangs under this node. with trace_mod.stage_scope() as node: @@ -969,20 +980,24 @@ class RuntimeBot: await self._record_adapter_event(event, adapter) await self._persist_monitoring_ingress(event, adapter, execution_id) - primary = ( - self._handle_interaction_submission(event, adapter, execution_id) + primary_handler = ( + self._handle_interaction_submission if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted' - else self._dispatch_eba_event_to_processor(event, adapter, execution_id=execution_id) + else self._dispatch_eba_event_to_processor ) subscriptions = self._get_event_bindings_from_value(getattr(self.bot_entity, 'plugin_processors', [])) seen = set() - tasks = [primary] + tasks = [self._run_event_delivery(primary_handler, event, adapter, execution_id=execution_id)] for subscription in subscriptions: processor_uuid = subscription.get('processor_uuid') if not processor_uuid or processor_uuid in seen or not subscription.get('enabled', True): continue seen.add(processor_uuid) - tasks.append(self._dispatch_plugin_subscription(event, adapter, processor_uuid, execution_id)) + tasks.append( + self._run_event_delivery( + self._dispatch_plugin_subscription, event, adapter, processor_uuid, execution_id + ) + ) # Start all deliveries together. A slow or failed subscriber cannot block # another subscriber or the primary route from receiving the event. results = await asyncio.gather(*tasks, return_exceptions=True) @@ -1009,6 +1024,18 @@ class RuntimeBot: results, ) + async def _run_event_delivery(self, handler, *args, **kwargs): + # gather creates child tasks. Each delivery needs its own database scope; + # copying the listener's ContextVar does not transfer task ownership. + persistence = getattr(getattr(self, 'ap', None), 'persistence_mgr', None) + if getattr(getattr(persistence, 'mode', None), 'value', None) == 'cloud_runtime': + tenant_scope = getattr(persistence, 'tenant_scope', None) + if not callable(tenant_scope): + raise RuntimeError('Cloud event deliveries require a tenant scope') + async with tenant_scope(self.workspace_uuid): + return await handler(*args, **kwargs) + return await handler(*args, **kwargs) + async def _dispatch_plugin_subscription(self, event, adapter, processor_uuid, execution_id=None): event_type = event.type event_binding = { @@ -1028,22 +1055,27 @@ class RuntimeBot: patterns = descriptor.supported_event_patterns if not patterns or not self._agent_supports_event_type(patterns, event_type): return - agent = {**agent, 'supported_event_patterns': patterns} - return await self._dispatch_eba_event_to_processor( - event, adapter, event_binding, agent, execution_id=execution_id - ) except Exception as exc: - return await self._record_event_route_trace( - event_type=event_type, - status='failed', - level='error', - binding=event_binding, - target_type='event_processor', - target_uuid=processor_uuid, - failure_code='runner_failed', - reason=str(exc), - text=f'Plugin processor {processor_uuid} failed: {exc}', - ) + from ..telemetry import trace as trace_mod + + # No matching declaration was resolved: log locally without + # reporting a failed delivery for every unrelated incoming event. + with trace_mod.suppress_reporting(): + return await self._record_event_route_trace( + event_type=event_type, + status='failed', + level='error', + binding=event_binding, + target_type='event_processor', + target_uuid=processor_uuid, + failure_code='runner_failed', + reason=str(exc), + text=f'Plugin processor {processor_uuid} failed: {exc}', + ) + agent = {**agent, 'supported_event_patterns': patterns} + return await self._dispatch_eba_event_to_processor( + event, adapter, event_binding, agent, execution_id=execution_id + ) async def _dispatch_eba_event_to_processor( self, @@ -1067,6 +1099,12 @@ class RuntimeBot: ) target_type = event_binding.get('target_type') + from ..telemetry import trace as trace_mod + + if target_type != 'discard': + # Matching precedes target lookup/execution, so genuine delivery + # failures remain observable even when the target is unavailable. + trace_mod.admit() await self._record_event_route_trace( event_type=event_type, status='matched', @@ -1079,7 +1117,7 @@ class RuntimeBot: if isinstance(event, platform_events.MessageReceivedEvent): from ..telemetry import trace as trace_mod - with trace_mod.scope(route_ref=self._route_ref(event_binding)): + with trace_mod.suppress_reporting(), trace_mod.scope(route_ref=self._route_ref(event_binding)): await self._dispatch_eba_message_to_pipeline( event, adapter, @@ -1202,6 +1240,10 @@ class RuntimeBot: envelope = self._eba_event_to_agent_envelope(event, adapter, execution_id) if target_type == 'event_processor': envelope.data = event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'}) + if isinstance(event, platform_events.MessageReceivedEvent): + # Use the SDK wire format for Source timestamps and nested chains. + # Pydantic's enclosing event serializer bypasses model_dump overrides. + envelope.data['message_chain'] = event.message_chain.model_dump() from ..telemetry import trace as trace_mod try: @@ -1442,13 +1484,18 @@ class RuntimeBot: conversation_id=conversation_id, actor_id=actor_id, ) + if record['processor_type'] not in {'agent', 'pipeline'}: + raise ValueError(f'Unsupported interaction processor type: {record["processor_type"]}') + from ..telemetry import trace as trace_mod + + # Only a validated callback resolving an original processor is eligible. + # Admit before acknowledgement and target lookup to preserve failures. + trace_mod.admit() await self.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission(record, adapter) if record['processor_type'] == 'agent': await self._resume_agent_interaction(record, event, adapter, actor_id, execution_id) return - if record['processor_type'] != 'pipeline': - raise ValueError(f'Unsupported interaction processor type: {record["processor_type"]}') pipeline = await self.ap.pipeline_service.get_pipeline( self.execution_context, diff --git a/src/langbot/pkg/telemetry/execution.py b/src/langbot/pkg/telemetry/execution.py index 8a72288f2..d914deadd 100644 --- a/src/langbot/pkg/telemetry/execution.py +++ b/src/langbot/pkg/telemetry/execution.py @@ -7,6 +7,11 @@ execution closes. The sender never assembles a trace locally, so there is no per-trace buffer, no TTL and no per-process trace cap; memory is bounded by the outbound record buffer, the flush batch size and the small cross-task registry. +Platform events are reportable only after a business route, matching plugin +subscription or validated interaction callback admits the execution. Pending +events enqueue neither nodes nor closures, even on failure or in Cloud mode. +Independent API/WebUI executions remain eligible without a Bot route. + Sampling is decided exactly once per event, at ingress, from a stable hash of the execution identity, so every record of one execution carries the same decision. A chain that broke is always emitted even when sampling excluded it, @@ -245,7 +250,7 @@ class ExecutionCounters: node: str = '', ): try: - if not self._enabled(): + if not self._enabled() or trace_mod.reporting_suppressed(): return if family not in FAMILIES or mode not in MODES or outcome not in OUTCOMES: return @@ -261,7 +266,7 @@ class ExecutionCounters: lookup = str(execution_id or '').strip() or current_execution_id() if lookup: state = self._resolve(lookup) - if state is None or state.closed: + if state is None or state.closed or not state.reportable: # Every observation belongs to exactly one execution chain. return self.configure(state) @@ -400,7 +405,7 @@ class ExecutionCounters: resources_end() def _chain_emitted(self, state: TraceState) -> bool: - if state.mode == 'off': + if not state.reportable or state.mode == 'off': return False if state.emit_nodes or state.debug: return True @@ -562,9 +567,14 @@ def ingress( debug: bool = False, origin: str = 'platform', synthetic_event: str = '', + require_route: bool = False, ): """Open one execution boundary; only the owner of the chain closes it. + ``require_route`` starts a platform event without reporting permission; + dispatch calls ``trace.admit`` when a delivery matches. Nested entrypoints + must reuse that decision instead of implicitly admitting the parent. + Nested boundaries (an event routed into a Pipeline, a Runner invoked from a dispatch) reuse the in-flight chain instead of starting a second one, and only alias their own execution id onto it. ``debug``/``origin`` apply when @@ -575,6 +585,8 @@ def ingress( binding = trace_mod.bind(execution_id) state = binding.state if binding.created: + if require_route: + state.reportable = False resources_begin() counters = _execution_counters(ap) if counters is not None: diff --git a/src/langbot/pkg/telemetry/trace.py b/src/langbot/pkg/telemetry/trace.py index 6ae8da2d1..ddc5f5d05 100644 --- a/src/langbot/pkg/telemetry/trace.py +++ b/src/langbot/pkg/telemetry/trace.py @@ -35,6 +35,9 @@ from uuid import uuid4 _current: contextvars.ContextVar['TraceState | None'] = contextvars.ContextVar('telemetry_trace', default=None) _route: contextvars.ContextVar[str] = contextvars.ContextVar('telemetry_trace_route', default='') _run: contextvars.ContextVar[str] = contextvars.ContextVar('telemetry_trace_run', default='') +_reporting_suppressed: contextvars.ContextVar[bool] = contextvars.ContextVar( + 'telemetry_reporting_suppressed', default=False +) # Open workflow step frames of the current task, innermost last. The tuple is # never mutated in place: opening a step publishes a new tuple on this task's # context only, and closing it restores the previous one. @@ -60,6 +63,7 @@ class TraceState: 'event_id', 'identity', 'configured', + 'reportable', 'mode', 'denominator', 'emit_nodes', @@ -90,6 +94,9 @@ class TraceState: self.event_id = str(execution_id).strip() if execution_id else str(uuid4()) self.identity: dict[str, str] = {} self.configured = False + # Platform ingress starts pending; an actual route admits the shared + # execution before work starts. Independent API/debug runs are eligible. + self.reportable = not reporting_suppressed() # Sampling, decided exactly once per event (at ingress or first touch). self.mode = 'all' self.denominator = 1 @@ -216,6 +223,27 @@ def current() -> TraceState | None: return _current.get() +def reporting_suppressed() -> bool: + return _reporting_suppressed.get() + + +def admit() -> None: + """Admit a matched delivery without revoking any concurrent sibling's choice.""" + state = current() + if state is not None and not state.closed and not reporting_suppressed(): + state.reportable = True + + +@contextlib.contextmanager +def suppress_reporting(): + """Keep discarded branch side effects local even if a sibling is admitted.""" + token = _reporting_suppressed.set(True) + try: + yield + finally: + _reporting_suppressed.reset(token) + + def node_stack() -> tuple[NodeFrame, ...]: """This task's own open-step frames, innermost last.""" return _node.get() diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index f8752314f..856563e39 100644 --- a/src/langbot/templates/config.yaml +++ b/src/langbot/templates/config.yaml @@ -444,10 +444,14 @@ space: # heartbeat and execution reporting; restart after changing this setting. disable_telemetry: false # Per-event execution traces: off | failures | sampled | all. - # Every execution is reported by default: one record per workflow node plus - # one terminal chain record. 'failures' keeps only broken chains, 'sampled' - # keeps broken chains plus every N-th successful one (N is - # execution_trace_sample), and 'off' reports no executions at all. + # Every eligible execution is reported by default: one record per workflow + # node plus one terminal chain record. 'failures' keeps only broken chains; + # 'sampled' keeps broken chains plus about 1/N of successful ones, selected + # by event hash (N is execution_trace_sample); 'off' reports no executions. + # Unmatched/discarded platform events are never reported unless a plugin + # subscription matches. Matched delivery failures and valid interaction + # resumptions remain eligible, as do independent API/WebUI executions. + # Cloud uses 'all' but still applies this route eligibility filter. execution_trace: all execution_trace_sample: 20 # How often buffered execution records are uploaded, in seconds. diff --git a/tests/unit_tests/platform/test_botmgr_execution_trace.py b/tests/unit_tests/platform/test_botmgr_execution_trace.py index 8fbad1463..2002f3045 100644 --- a/tests/unit_tests/platform/test_botmgr_execution_trace.py +++ b/tests/unit_tests/platform/test_botmgr_execution_trace.py @@ -4,6 +4,7 @@ from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock, Mock +import asyncio import pytest @@ -93,38 +94,16 @@ def execution_chains(records: list[dict]) -> list[dict]: @pytest.mark.asyncio -async def test_route_miss_emits_one_closed_chain(): +async def test_route_miss_emits_nothing(): bot, manager, counters = make_bot([]) await bot._handle_platform_event(message_received_event(), FakeAdapter()) - records = await flushed_records(counters, manager) - assert len(manager.sent) == 1 - chains = execution_chains(records) - assert len(chains) == 1 - chain = chains[0] - assert chain['schema'] == 2 - # The execution identity is deterministic: one inbound event, one chain. - assert chain['event_id'] == 'platform:bot-1:message-1' - assert chain['instance_id'] == 'instance-test' - assert chain['workspace_uuid'] == 'workspace-test' - assert chain['closed_by'] == 'event_done' - assert chain['stage_count'] == 2 - - rows = execution_nodes(records) - assert [(row['family'], row['operation'], row['outcome']) for row in rows] == [ - ('event_route', 'message.received', 'skipped'), - ('platform_event', 'message.received', 'success'), - ] - assert {row['seq'] for row in rows} == {0, 1} - assert all(row['event_id'] == chain['event_id'] for row in rows) - assert all(row['adapter'] == 'FakeAdapter' for row in rows) - # The inbound event is the root step; the routing decision nests under it. - root = next(row for row in rows if row['root']) - route = next(row for row in rows if row['family'] == 'event_route') - assert (root['family'], root['parent_node_id']) == ('platform_event', '') - assert route['parent_node_id'] == root['node_id'] - # The chain is closed and deregistered: nothing dangles after the event. + await counters.flush() + await counters.shutdown() + assert manager.sent == [] + assert counters.records == [] + assert counters.dropped == 0 assert counters.registry == {} @@ -154,7 +133,7 @@ async def test_unavailable_route_target_carries_route_identity(): @pytest.mark.asyncio -async def test_discarded_route_is_visible_without_user_values(): +async def test_discarded_route_is_only_logged_locally(): bot, manager, counters = make_bot( [ { @@ -168,10 +147,10 @@ async def test_discarded_route_is_visible_without_user_values(): await bot._handle_platform_event(SimpleNamespace(type='platform.member.joined'), FakeAdapter()) - nodes = execution_nodes(await flushed_records(counters, manager)) - assert [node['outcome'] for node in nodes] == ['skipped', 'success'] - assert nodes[0]['operation'] == 'platform.member.joined' - assert nodes[0]['mode'] == 'none' + await counters.flush() + assert manager.sent == [] + assert counters.records == [] + assert bot.logger.info.await_args.kwargs['metadata']['status'] == 'discarded' @pytest.mark.asyncio @@ -206,11 +185,377 @@ async def test_route_trace_scope_is_restored_after_dispatch(): @pytest.mark.asyncio -async def test_adapter_call_without_ingress_still_counts(): +async def test_mock_adapter_route_miss_is_not_reported(): """Mock adapter objects used by other tests must not break the ingress path.""" bot, manager, counters = make_bot([]) adapter = Mock() await bot._handle_platform_event(message_received_event(), adapter) - assert await flushed_records(counters, manager) + await counters.flush() + assert manager.sent == [] + + +def business_route(**overrides): + return { + 'id': 'route', + 'enabled': True, + 'event_pattern': '*', + 'target_type': 'agent', + 'target_uuid': 'agent-1', + **overrides, + } + + +async def assert_no_upload(counters, manager): + assert counters.records == [] + assert counters.task is None + await counters.flush() + await counters.shutdown() + assert manager.sent == [] + assert counters.health() == { + 'records_buffered': 0, + 'records_sent': 0, + 'records_dropped': 0, + 'executions_in_flight': 0, + } + + +@pytest.mark.parametrize('edition', ['community', 'cloud']) +@pytest.mark.parametrize('mode', ['all', 'sampled', 'failures', 'off']) +@pytest.mark.parametrize( + 'routes', + [ + [], + [business_route(enabled=False)], + [business_route(event_pattern='group.member_joined')], + [business_route(filters=[{'field': 'chat_id', 'operator': 'eq', 'value': 'another-chat'}])], + [business_route(target_type='discard', priority=10), business_route(priority=1)], + ], +) +async def test_unrouted_events_never_upload_in_any_mode(monkeypatch, edition, mode, routes): + from langbot.pkg.utils import constants + + monkeypatch.setattr(constants, 'edition', edition) + bot, manager, counters = make_bot(routes, config={'url': 'https://space.example.test', 'execution_trace': mode}) + bot._dispatch_eba_message_to_pipeline = AsyncMock() + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + await assert_no_upload(counters, manager) + # Normal local route diagnostics still exist, including explicit discard. + assert bot.logger.info.await_args.kwargs['metadata']['status'] in {'not_matched', 'discarded'} + + +@pytest.mark.parametrize('mode', ['all', 'sampled', 'failures']) +async def test_matched_target_failure_is_retained_even_after_root_success(mode): + bot, manager, counters = make_bot( + [business_route()], + config={ + 'url': 'https://space.example.test', + 'execution_trace': mode, + 'execution_trace_sample': 100000, + }, + ) + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + records = await flushed_records(counters, manager) + (chain,) = execution_chains(records) + assert chain['outcome'] == 'failed' + assert chain['error'] == 'processor_not_found' + if mode == 'failures': + assert execution_nodes(records) == [] + await counters.shutdown() + + +def install_processor(bot, patterns=None): + bot.bot_entity.plugin_processors = [{'processor_uuid': 'processor', 'enabled': True}] + bot.ap.agent_service.get_agent = AsyncMock( + return_value={ + 'kind': 'event_processor', + 'component_ref': 'plugin:test/runner/default', + } + ) + bot.ap.runner_registry = SimpleNamespace( + get=AsyncMock( + return_value=SimpleNamespace( + usages=['event'], + supported_event_patterns=['*'] if patterns is None else patterns, + ) + ) + ) + bot._agent_product_to_binding = Mock(return_value=SimpleNamespace()) + bot._eba_event_to_agent_envelope = Mock(return_value=SimpleNamespace()) + + async def run(*args, **kwargs): + from langbot.pkg.telemetry.execution import record + + record(bot.ap, TEST_CONTEXT, family='runner', operation='execute', outcome='success') + if False: + yield None + + bot.ap.agent_run_orchestrator = SimpleNamespace(run=run) + + +@pytest.mark.parametrize('primary', ['miss', 'discard']) +@pytest.mark.parametrize('primary_first', [True, False]) +async def test_matching_subscription_admits_chain_regardless_of_primary_order(primary, primary_first): + from langbot.pkg.telemetry.execution import record + + routes = [] if primary == 'miss' else [business_route(target_type='discard')] + bot, manager, counters = make_bot(routes) + install_processor(bot) + bot.bot_entity.plugin_processors *= 2 # Duplicate subscriptions still execute once. + original = bot._dispatch_eba_event_to_processor + first_done = asyncio.Event() + run_calls = [] + + async def ordered_dispatch(event, adapter, event_binding=None, agent=None, execution_id=None): + is_primary = event_binding is None + if is_primary != primary_first: + await first_done.wait() + result = await original(event, adapter, event_binding, agent, execution_id) + if not is_primary: + run_calls.append(result) + if is_primary == primary_first: + first_done.set() + return result + + async def discard_side_effect(*args, **kwargs): + # A discarded message can still trigger a webhook/adapter side effect. + # A concurrent admitted subscriber must not expose this branch's node. + record(bot.ap, TEST_CONTEXT, family='platform_api', operation='discard_side_effect', outcome='success') + await counters.flush() + + bot._dispatch_eba_event_to_processor = ordered_dispatch + bot._dispatch_eba_message_to_pipeline = discard_side_effect + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + records = await flushed_records(counters, manager) + (chain,) = execution_chains(records) + nodes = execution_nodes(records) + assert len(run_calls) == 1 + assert chain['stage_count'] == 3 + assert {node['family'] for node in nodes} == {'runner', 'event_route', 'platform_event'} + assert all(node['outcome'] == 'success' for node in nodes) + (root,) = [node for node in nodes if node['root']] + assert root['family'] == 'platform_event' + assert all(node['parent_node_id'] == root['node_id'] for node in nodes if not node['root']) + assert all(node['event_id'] == chain['event_id'] for node in nodes) + await counters.shutdown() + + +@pytest.mark.parametrize('kind', ['disabled', 'pattern_miss', 'no_patterns', 'missing', 'declaration_error']) +async def test_unmatched_subscription_cannot_admit_event(kind): + bot, manager, counters = make_bot([]) + install_processor(bot) + if kind == 'disabled': + bot.bot_entity.plugin_processors[0]['enabled'] = False + elif kind in {'pattern_miss', 'no_patterns'}: + install_processor(bot, [] if kind == 'no_patterns' else ['group.member_joined']) + elif kind == 'missing': + bot.ap.agent_service.get_agent.return_value = None + else: + bot.ap.runner_registry.get.side_effect = RuntimeError('unavailable') + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + await assert_no_upload(counters, manager) + + +@pytest.mark.parametrize('failure', [RuntimeError('runner failed'), TimeoutError('timeout'), asyncio.CancelledError()]) +async def test_matched_subscription_keeps_failed_timeout_and_cancelled_execution(failure): + bot, manager, counters = make_bot([], config={'url': 'https://space.example.test', 'execution_trace': 'failures'}) + install_processor(bot) + + async def run(*args, **kwargs): + raise failure + yield + + bot.ap.agent_run_orchestrator.run = run + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + (chain,) = execution_chains(await flushed_records(counters, manager)) + assert chain['outcome'] == 'failed' + assert chain['error'] + await counters.shutdown() + + +@pytest.mark.parametrize('processor', ['agent', 'pipeline', 'invalid', 'expired', 'duplicate']) +async def test_interaction_admission_requires_valid_callback(processor): + from langbot_plugin.api.entities.builtin.platform import events + + bot, manager, counters = make_bot([]) + callbacks = SimpleNamespace( + consume_callback=AsyncMock(return_value={'processor_type': processor, 'processor_id': 'target'}), + acknowledge_submission=AsyncMock(), + ) + if processor in {'expired', 'duplicate'}: + callbacks.consume_callback.side_effect = ValueError(processor) + bot.ap.agent_run_orchestrator = SimpleNamespace(interaction_manager=callbacks) + bot._resume_agent_interaction = AsyncMock(side_effect=ValueError('target unavailable')) + event = events.PlatformSpecificEvent(platform='test', action='interaction.submitted', data={}) + await bot._handle_platform_event(event, FakeAdapter()) + if processor in {'agent', 'pipeline'}: + (chain,) = execution_chains(await flushed_records(counters, manager)) + assert chain['outcome'] == 'failed' + callbacks.acknowledge_submission.assert_awaited_once() + await counters.shutdown() + else: + callbacks.acknowledge_submission.assert_not_awaited() + await assert_no_upload(counters, manager) + + +async def test_rejected_event_does_not_contaminate_next_matched_event_or_monitoring(): + from langbot.pkg.telemetry import trace + + bot, manager, counters = make_bot([]) + bot._persist_monitoring_ingress = AsyncMock() + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + assert bot._persist_monitoring_ingress.await_count == 2 + assert counters.records == [] + assert trace.current() is None + assert trace.node_stack() == () + assert counters.registry == {} + bot.bot_entity.event_bindings = [business_route()] + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + assert len(execution_chains(await flushed_records(counters, manager))) == 1 + await counters.shutdown() + + +async def test_pending_gate_blocks_nested_and_rpc_records_and_failure_fallback(): + import contextvars + from langbot.pkg.telemetry import execution, trace + + bot, manager, counters = make_bot([]) + with pytest.raises(RuntimeError): + with execution.ingress(bot.ap, context=TEST_CONTEXT, execution_id='pending', require_route=True, debug=True): + with execution.ingress(bot.ap, context=TEST_CONTEXT, debug=True): + execution.bind_trace(bot.ap, 'run-alias') + execution.record(bot.ap, TEST_CONTEXT, family='tool', operation='nested', outcome='failed') + + async def rpc(): + execution.record( + bot.ap, + TEST_CONTEXT, + execution_id='run-alias', + family='platform_api', + operation='rpc', + outcome='failed', + ) + await counters.flush() + + await contextvars.Context().run(asyncio.create_task, rpc()) + assert counters.records == [] + raise RuntimeError('failed before a route matched') + assert trace.current() is None + await assert_no_upload(counters, manager) + + +@pytest.mark.parametrize('origin', ['api', 'webui']) +async def test_independent_debug_entry_still_reports(origin): + from langbot.pkg.telemetry import execution + + bot, manager, counters = make_bot([], config={'url': 'https://space.example.test', 'execution_trace': 'failures'}) + with execution.ingress( + bot.ap, + context=TEST_CONTEXT, + execution_id='debug', + debug=True, + origin=origin, + synthetic_event='message.received', + ): + execution.record(bot.ap, TEST_CONTEXT, family='runner', operation='execute', outcome='success') + (chain,) = execution_chains(await flushed_records(counters, manager)) + assert chain['debug'] is True + assert chain['origin'] == origin + await counters.shutdown() + + +@pytest.mark.parametrize('edition', ['community', 'cloud']) +async def test_admission_does_not_override_telemetry_opt_out(monkeypatch, edition): + from langbot.pkg.utils import constants + + monkeypatch.setattr(constants, 'edition', edition) + bot, manager, counters = make_bot( + [business_route()], + config={ + 'url': 'https://space.example.test', + 'disable_telemetry': True, + }, + ) + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + await assert_no_upload(counters, manager) + + +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() + bot.bot_entity.plugin_processors = [{'processor_uuid': 'missing'}] + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + records = await flushed_records(counters, manager) + (chain,) = execution_chains(records) + assert chain['outcome'] == 'success' + assert chain['error'] == '' + assert all(node['outcome'] == 'success' for node in execution_nodes(records)) + bot.logger.error.assert_awaited_once() + await counters.shutdown() + + +async def test_pending_event_cancelled_during_dispatch_never_uploads(): + bot, manager, counters = make_bot([]) + started = asyncio.Event() + + async def waiting_dispatch(*args, **kwargs): + started.set() + await asyncio.Event().wait() + + bot._dispatch_eba_event_to_processor = waiting_dispatch + task = asyncio.create_task(bot._handle_platform_event(message_received_event(), FakeAdapter())) + await started.wait() + await counters.flush() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + await assert_no_upload(counters, manager) + + +async def test_retry_of_matched_batch_cannot_include_intervening_unrouted_event(): + bot, manager, counters = make_bot([business_route()]) + await bot._handle_platform_event(message_received_event(), FakeAdapter()) + expected = list(counters.records) + original_send = manager.send + manager.send = AsyncMock(return_value=False) + assert await counters.flush() is False + assert counters.records == expected + bot.bot_entity.event_bindings = [] + ignored = message_received_event() + ignored.message_id = 'ignored-message' + await bot._handle_platform_event(ignored, FakeAdapter()) + assert counters.records == expected + manager.send = original_send + await counters.shutdown() + sent = [record for batch in manager.sent for record in batch['records']] + assert sent == expected + assert counters.dropped == 0 + + +async def test_valid_interaction_success_keeps_acknowledgement_and_execution(): + from langbot.pkg.telemetry.execution import record + from langbot_plugin.api.entities.builtin.platform import events + + bot, manager, counters = make_bot([]) + + async def acknowledge(*args): + record(bot.ap, TEST_CONTEXT, family='platform_api', operation='ack', outcome='success') + + async def resume(*args): + record(bot.ap, TEST_CONTEXT, family='runner', operation='execute', outcome='success') + + bot.ap.agent_run_orchestrator = SimpleNamespace( + interaction_manager=SimpleNamespace( + consume_callback=AsyncMock(return_value={'processor_type': 'agent'}), + acknowledge_submission=acknowledge, + ) + ) + bot._resume_agent_interaction = resume + event = events.PlatformSpecificEvent(platform='test', action='interaction.submitted', data={}) + await bot._handle_platform_event(event, FakeAdapter()) + records = await flushed_records(counters, manager) + (chain,) = execution_chains(records) + assert chain['outcome'] == 'success' + assert {node['family'] for node in execution_nodes(records)} == {'platform_event', 'platform_api', 'runner'} + await counters.shutdown() diff --git a/tests/unit_tests/platform/test_plugin_subscriptions.py b/tests/unit_tests/platform/test_plugin_subscriptions.py index b69bc9d4f..d074bf659 100644 --- a/tests/unit_tests/platform/test_plugin_subscriptions.py +++ b/tests/unit_tests/platform/test_plugin_subscriptions.py @@ -1,6 +1,7 @@ """Independent bot subscriptions must not change primary route delivery.""" import asyncio +import contextvars from types import SimpleNamespace from unittest.mock import AsyncMock @@ -8,6 +9,7 @@ import pytest from langbot_plugin.api.entities.builtin.platform.events import MemberJoinedEvent from langbot.pkg.platform.botmgr import RuntimeBot +from langbot.pkg.persistence.tenant_uow import PersistenceScope, PersistenceScopeBoundary def make_bot(subscriptions): @@ -105,3 +107,46 @@ async def test_missing_processor_is_logged_without_unscoped_lookup(): await bot._dispatch_plugin_subscription(MemberJoinedEvent(), None, 'missing') bot.ap.agent_service.get_agent.assert_awaited_once_with('workspace', 'missing') assert bot._record_event_route_trace.await_args.kwargs['status'] == 'failed' + + +@pytest.mark.parametrize('primary_fails', [False, True]) +async def test_cloud_deliveries_own_their_scope_and_preserve_parent(primary_fails): + bot = make_bot([{'processor_uuid': 'plugin'}]) + bot.workspace_uuid = 'workspace' + active_scope = contextvars.ContextVar('test_event_scope', default=None) + active_transaction = contextvars.ContextVar('test_event_transaction', default=None) + + def tenant_scope(workspace): + return PersistenceScopeBoundary( + PersistenceScope.workspace(workspace), + active_scope=active_scope, + active_transaction=active_transaction, + ) + + bot.ap = SimpleNamespace( + persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime'), tenant_scope=tenant_scope) + ) + seen = [] + + async def delivery(event, adapter, *args, **kwargs): + state = active_scope.get() + assert state.owner_task is asyncio.current_task() + assert state.scope == PersistenceScope.workspace('workspace') + seen.append(state) + await asyncio.sleep(0) + assert active_scope.get() is state + if primary_fails and not args: + raise ValueError('primary failed') + + bot._dispatch_eba_event_to_processor = delivery + bot._dispatch_plugin_subscription = delivery + async with tenant_scope('workspace'): + parent = active_scope.get() + await bot._handle_platform_event(MemberJoinedEvent(), None) + assert active_scope.get() is parent + assert parent.owner_task is asyncio.current_task() + assert active_scope.get() is None + assert len(seen) == 2 + assert seen[0] is not seen[1] + assert all(state is not parent and state.depth == 0 for state in seen) + assert bot.logger.error.await_count == int(primary_fails) diff --git a/tests/unit_tests/platform/test_routing_rules.py b/tests/unit_tests/platform/test_routing_rules.py index 1a339c936..ce93c6099 100644 --- a/tests/unit_tests/platform/test_routing_rules.py +++ b/tests/unit_tests/platform/test_routing_rules.py @@ -767,6 +767,7 @@ async def test_processor_outputs_require_explicit_platform_actions(kind, output_ """Draining runner results must not send text, duplicate replies, or stream cards.""" from langbot_plugin.api.entities.builtin.platform import entities, events, message from langbot_plugin.api.entities.builtin.provider import message as provider_message + from datetime import datetime from langbot.pkg.agent.runner.platform_tools import execute_platform_tool, freeze_platform_context @@ -783,6 +784,10 @@ async def test_processor_outputs_require_explicit_platform_actions(kind, output_ completed = [] async def run(envelope, binding, adapter_context): + if kind == 'event_processor': + restored = message.MessageChain.model_validate(envelope.data['message_chain']) + assert restored.source.id == 'source-1' + assert restored.source.time == datetime(2026, 10, 8, 12) assert envelope.delivery.supports_streaming is False assert binding.delivery_policy.enable_streaming is False assert binding.delivery_policy.enable_reply is False @@ -823,7 +828,9 @@ async def test_processor_outputs_require_explicit_platform_actions(kind, output_ ) event = events.MessageReceivedEvent( message_id='message-1', - message_chain=message.MessageChain([message.Plain(text='hello')]), + message_chain=message.MessageChain( + [message.Source(id='source-1', time=datetime(2026, 10, 8, 12)), message.Plain(text='hello')] + ), sender=entities.User(id='user-1', nickname='QA'), chat_type=entities.ChatType.PRIVATE, chat_id='user-1',