fix(telemetry): skip unrouted events and preserve matched deliveries

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,

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:

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

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.

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

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)

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',