fix(monitoring): unify execution records and stabilize streaming cleanup

src/langbot/pkg/api/http/controller/groups/monitoring.py

@@ -86,7 +86,14 @@ class MonitoringRouterGroup(group.RouterGroup):
         async def get_execution_detail(source: str, execution_id: str, request_context: RequestContext) -> str:
             """Full trace of a single execution (agent run or pipeline query)."""
             try:
-                detail = await self.ap.monitoring_service.get_execution_detail(request_context, source, execution_id)
+                detail = await self.ap.monitoring_service.get_execution_detail(
+                    request_context,
+                    source,
+                    execution_id,
+                    section=quart.request.args.get('section'),
+                    offset=quart.request.args.get('offset', 0),
+                    limit=quart.request.args.get('limit', 100),
+                )
             except ValueError as exc:
                 return self.fail(404, str(exc))
             return self.success(data=detail)
@@ -107,6 +114,8 @@ class MonitoringRouterGroup(group.RouterGroup):
 
             stats = await self.ap.monitoring_service.get_token_statistics(
                 request_context,
+                execution_statuses=quart.request.args.getlist('status') or None,
+                mode=quart.request.args.get('mode', 'all'),
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -368,6 +377,10 @@ class MonitoringRouterGroup(group.RouterGroup):
         @self.route('/data', methods=['GET'], permission=Permission.RESOURCE_VIEW)
         async def get_all_data(request_context: RequestContext) -> str:
             """Get all monitoring data in a single request"""
+            execution_filters = {
+                'execution_statuses': quart.request.args.getlist('status') or None,
+                'mode': quart.request.args.get('mode', 'all'),
+            }
             # Parse query parameters
             bot_ids = quart.request.args.getlist('botId')
             pipeline_ids = quart.request.args.getlist('pipelineId')
@@ -382,6 +395,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get overview metrics
             overview = await self.ap.monitoring_service.get_overview_metrics(
                 request_context,
+                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -391,6 +405,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get messages
             messages, messages_total = await self.ap.monitoring_service.get_messages(
                 request_context,
+                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -402,6 +417,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get LLM calls
             llm_calls, llm_calls_total = await self.ap.monitoring_service.get_llm_calls(
                 request_context,
+                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -413,6 +429,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get tool calls
             tool_calls, tool_calls_total = await self.ap.monitoring_service.get_tool_calls(
                 request_context,
+                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -424,6 +441,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get sessions
             sessions, sessions_total = await self.ap.monitoring_service.get_sessions(
                 request_context,
+                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -436,6 +454,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get errors
             errors, errors_total = await self.ap.monitoring_service.get_errors(
                 request_context,
+                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -447,6 +466,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             # Get embedding calls
             embedding_calls, embedding_calls_total = await self.ap.monitoring_service.get_embedding_calls(
                 request_context,
+                **execution_filters,
                 start_time=start_time,
                 end_time=end_time,
                 limit=limit,
@@ -458,6 +478,7 @@ class MonitoringRouterGroup(group.RouterGroup):
                     'traffic': await get_traffic_series(
                         self.ap,
                         request_context,
+                        **execution_filters,
                         bot_ids=bot_ids or None,
                         pipeline_ids=pipeline_ids or None,
                         start_time=start_time,
@@ -725,6 +746,8 @@ class MonitoringRouterGroup(group.RouterGroup):
 
             stats = await self.ap.monitoring_service.get_feedback_stats(
                 request_context,
+                execution_statuses=quart.request.args.getlist('status') or None,
+                mode=quart.request.args.get('mode', 'all'),
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
@@ -754,6 +777,8 @@ class MonitoringRouterGroup(group.RouterGroup):
 
             feedback_list, total = await self.ap.monitoring_service.get_feedback_list(
                 request_context,
+                execution_statuses=quart.request.args.getlist('status') or None,
+                mode=quart.request.args.get('mode', 'all'),
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 feedback_type=feedback_type,

src/langbot/pkg/api/http/service/agent.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from contextlib import aclosing
+
 import datetime
 import copy
 import fnmatch
@@ -365,25 +367,28 @@ class AgentService:
             origin=origin,
             synthetic_event=event_type,
         ):
-            async for output in self.ap.agent_run_orchestrator.run(
-                event,
-                binding,
-                adapter_context={
-                    '_query': execution_query,
-                    '_execution_context': execution_context,
-                    '_result_observer': observe_result,
-                },
-            ):
-                output_text = self._provider_output_to_text(output)
-                if output_text:
-                    final_text = output_text
-                output_items.append(
-                    {
-                        'kind': output.__class__.__name__,
-                        'role': str(getattr(output, 'role', '') or ''),
-                        'text': output_text,
-                    }
+            async with aclosing(
+                self.ap.agent_run_orchestrator.run(
+                    event,
+                    binding,
+                    adapter_context={
+                        '_query': execution_query,
+                        '_execution_context': execution_context,
+                        '_result_observer': observe_result,
+                    },
                 )
+            ) as owned_stream:
+                async for output in owned_stream:
+                    output_text = self._provider_output_to_text(output)
+                    if output_text:
+                        final_text = output_text
+                    output_items.append(
+                        {
+                            'kind': output.__class__.__name__,
+                            'role': str(getattr(output, 'role', '') or ''),
+                            'text': output_text,
+                        }
+                    )
 
         return {
             'event_id': event_id,

src/langbot/pkg/api/http/service/monitoring.py

@@ -14,9 +14,11 @@ from ....entity.persistence import agent as persistence_agent
 from ....entity.persistence import monitoring as persistence_monitoring
 from ....entity.persistence import agent_run as persistence_agent_run
 from ....entity.persistence import agent_interaction as persistence_agent_interaction
+from ....entity.persistence.event_log import EventLog
 from ..authz import WorkspaceRequiredError
 from ..context import ExecutionContext
 from .tenant import TenantContext, require_workspace_uuid
+from .monitoring_execution_details import ExecutionDetailsMixin
 
 
 _DEFAULT_MONITORING_PAGE_ROWS = 1000
@@ -131,7 +133,7 @@ def _workspace_transaction(method):
     return wrapped
 
 
-class MonitoringService:
+class MonitoringService(ExecutionDetailsMixin):
     """Monitoring service"""
 
     ap: app.Application
@@ -535,9 +537,17 @@ class MonitoringService:
         runner_name: str | None = None,
         variables: str | None = None,
         role: str = 'user',
+        event_id: str | None = None,
+        run_id: str | None = None,
+        parent_message_id: str | None = None,
     ) -> str:
         """Record a message"""
         workspace_uuid = self._require_write_context(context)
+        from ....telemetry import trace
+
+        trace_state = trace.current()
+        event_id = event_id or (trace_state.event_id if trace_state else None)
+        run_id = run_id or trace.current_run() or None
         message_id = str(uuid.uuid4())
         message_content = self._sanitize_message_content(message_content)
         message_data = {
@@ -558,6 +568,9 @@ class MonitoringService:
             'runner_name': runner_name,
             'variables': variables,
             'role': role,
+            'event_id': event_id,
+            'run_id': run_id,
+            'parent_message_id': parent_message_id,
         }
 
         await self.ap.persistence_mgr.execute_async(
@@ -879,6 +892,8 @@ class MonitoringService:
         pipeline_ids: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> dict:
         """Get overview metrics"""
         workspace_uuid = require_workspace_uuid(context)
@@ -888,15 +903,47 @@ class MonitoringService:
         embedding_conditions = [persistence_monitoring.MonitoringEmbeddingCall.workspace_uuid == workspace_uuid]
         session_conditions = [persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid]
 
+        message_conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringMessage, workspace_uuid, execution_statuses, mode
+            )
+        )
+        llm_conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringLLMCall, workspace_uuid, execution_statuses, mode
+            )
+        )
+        session_conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringSession, workspace_uuid, execution_statuses, mode
+            )
+        )
+        embedding_conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringEmbeddingCall, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             message_conditions.append(persistence_monitoring.MonitoringMessage.bot_id.in_(bot_ids))
             llm_conditions.append(persistence_monitoring.MonitoringLLMCall.bot_id.in_(bot_ids))
             session_conditions.append(persistence_monitoring.MonitoringSession.bot_id.in_(bot_ids))
 
         if pipeline_ids:
-            message_conditions.append(persistence_monitoring.MonitoringMessage.pipeline_id.in_(pipeline_ids))
-            llm_conditions.append(persistence_monitoring.MonitoringLLMCall.pipeline_id.in_(pipeline_ids))
-            session_conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids))
+            message_conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringMessage, workspace_uuid, pipeline_ids
+                )
+            )
+            llm_conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringLLMCall, workspace_uuid, pipeline_ids
+                )
+            )
+            session_conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringSession, workspace_uuid, pipeline_ids
+                )
+            )
 
         if start_time:
             message_conditions.append(persistence_monitoring.MonitoringMessage.timestamp >= start_time)
@@ -973,23 +1020,109 @@ class MonitoringService:
 
     # ========== Execution view ==========
 
+    @staticmethod
+    def _debug_run_condition(model):
+        return sqlalchemy.or_(
+            model.binding_id.like('debug:%'),
+            model.conversation_id.like('debug:%'),
+            model.bot_id == 'websocket-proxy-bot',
+        )
+
+    @staticmethod
+    def _debug_pipeline_condition(model):
+        # Dashboard/API pipeline debug uses this reserved proxy identity.
+        # Real Web Page bots have their own bot UUID, even on the same adapter.
+        return sqlalchemy.func.coalesce(model.bot_id == 'websocket-proxy-bot', False)
+
+    def _monitoring_execution_condition(self, model, workspace_uuid, statuses=None, mode='all'):
+        """Filter telemetry by its owning execution, not by the call's outcome."""
+        if not statuses and mode == 'all':
+            return sqlalchemy.true()
+        Run = persistence_agent_run.AgentRun
+        Message = persistence_monitoring.MonitoringMessage
+        roots = sqlalchemy.orm.aliased(Message)
+        run_conditions = [Run.workspace_id == workspace_uuid]
+        if statuses:
+            run_conditions.append(Run.status.in_(self._raw_agent_statuses(statuses)))
+        debug = self._debug_run_condition(Run)
+        if mode != 'all':
+            run_conditions.append(debug if mode == 'debug' else sqlalchemy.not_(sqlalchemy.func.coalesce(debug, False)))
+        runs = sqlalchemy.select(Run.run_id).where(*run_conditions)
+        root_conditions = [roots.workspace_uuid == workspace_uuid, roots.role != 'assistant']
+        if statuses:
+            root_conditions.append(roots.status.in_(self._raw_pipeline_statuses(statuses)))
+        if mode != 'all':
+            root_debug = self._debug_pipeline_condition(roots)
+            root_conditions.append(root_debug if mode == 'debug' else sqlalchemy.not_(root_debug))
+        root_ids = sqlalchemy.select(roots.id).where(*root_conditions)
+        linked = sqlalchemy.or_(
+            Message.id.in_(root_ids), Message.parent_message_id.in_(root_ids), Message.run_id.in_(runs)
+        )
+        message_ids = sqlalchemy.select(Message.id).where(Message.workspace_uuid == workspace_uuid, linked)
+        if model is Message:
+            return linked
+        if hasattr(model, 'message_id'):
+            return sqlalchemy.or_(model.message_id.in_(runs), model.message_id.in_(message_ids))
+        if model is persistence_monitoring.MonitoringSession:
+            return sqlalchemy.or_(
+                model.session_id.in_(sqlalchemy.select(roots.session_id).where(*root_conditions)),
+                model.session_id.in_(sqlalchemy.select(Run.conversation_id).where(*run_conditions)),
+            )
+        return sqlalchemy.false()
+
+    def _processor_runs(self, workspace_uuid, processor_ids):
+        Run = persistence_agent_run.AgentRun
+        return sqlalchemy.select(Run.run_id).where(
+            Run.workspace_id == workspace_uuid,
+            sqlalchemy.func.coalesce(self._execution_json_text(Run.metadata_json, 'processor_id'), Run.agent_id).in_(
+                processor_ids
+            ),
+        )
+
+    def _monitoring_processor_condition(self, model, workspace_uuid, processor_ids):
+        direct = model.pipeline_id.in_(processor_ids)
+        if hasattr(model, 'message_id'):
+            Message = persistence_monitoring.MonitoringMessage
+            return sqlalchemy.or_(
+                direct,
+                model.message_id.in_(self._processor_runs(workspace_uuid, processor_ids)),
+                model.message_id.in_(
+                    sqlalchemy.select(Message.id).where(
+                        Message.workspace_uuid == workspace_uuid, Message.pipeline_id.in_(processor_ids)
+                    )
+                ),
+            )
+        return direct
+
     def _agent_run_conditions(
         self,
         workspace_uuid: str,
         *,
         bot_ids: list[str] | None = None,
         agent_ids: list[str] | None = None,
+        processor_ids: list[str] | None = None,
         statuses: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
         debug: bool | None = None,
     ) -> list:
         AgentRun = persistence_agent_run.AgentRun
-        conditions = [AgentRun.workspace_id == workspace_uuid]
+        conditions = [
+            AgentRun.workspace_id == workspace_uuid,
+            ~sqlalchemy.exists(
+                sqlalchemy.select(persistence_monitoring.MonitoringMessage.id).where(
+                    persistence_monitoring.MonitoringMessage.workspace_uuid == workspace_uuid,
+                    persistence_monitoring.MonitoringMessage.run_id == AgentRun.run_id,
+                    persistence_monitoring.MonitoringMessage.role != 'assistant',
+                )
+            ),
+        ]
         if bot_ids:
             conditions.append(AgentRun.bot_id.in_(bot_ids))
         if agent_ids:
             conditions.append(AgentRun.agent_id.in_(agent_ids))
+        if processor_ids:
+            conditions.append(AgentRun.run_id.in_(self._processor_runs(workspace_uuid, processor_ids)))
         if statuses is not None:
             conditions.append(AgentRun.status.in_(statuses) if statuses else sqlalchemy.false())
         if start_time:
@@ -997,11 +1130,10 @@ class MonitoringService:
         if end_time:
             conditions.append(AgentRun.created_at <= end_time)
         if debug is not None:
-            debug_condition = sqlalchemy.or_(
-                AgentRun.binding_id.like('debug:%'),
-                AgentRun.conversation_id.like('debug:%'),
+            debug_condition = self._debug_run_condition(AgentRun)
+            conditions.append(
+                debug_condition if debug else sqlalchemy.not_(sqlalchemy.func.coalesce(debug_condition, False))
             )
-            conditions.append(debug_condition if debug else sqlalchemy.not_(debug_condition))
         return conditions
 
     def _pipeline_execution_conditions(
@@ -1013,6 +1145,7 @@ class MonitoringService:
         statuses: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
+        debug: bool | None = None,
     ) -> list:
         MonitoringMessage = persistence_monitoring.MonitoringMessage
         conditions = [
@@ -1023,13 +1156,16 @@ class MonitoringService:
         if bot_ids:
             conditions.append(MonitoringMessage.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(MonitoringMessage.pipeline_id.in_(pipeline_ids))
+            conditions.append(self._monitoring_processor_condition(MonitoringMessage, workspace_uuid, pipeline_ids))
         if statuses is not None:
             conditions.append(MonitoringMessage.status.in_(statuses) if statuses else sqlalchemy.false())
         if start_time:
             conditions.append(MonitoringMessage.timestamp >= start_time)
         if end_time:
             conditions.append(MonitoringMessage.timestamp <= end_time)
+        if debug is not None:
+            debug_condition = self._debug_pipeline_condition(MonitoringMessage)
+            conditions.append(debug_condition if debug else sqlalchemy.not_(debug_condition))
         return conditions
 
     @staticmethod
@@ -1079,7 +1215,10 @@ class MonitoringService:
         A run stores the Agent it executed as a uuid; the list has to name the
         processor, so the names are resolved for the whole page at once.
         """
-        agent_ids = {row.agent_id for row in rows if row.agent_id}
+        agent_ids = {self._deserialize_json(row.metadata_json).get('processor_id') or row.agent_id for row in rows} - {
+            None,
+            '',
+        }
         if not agent_ids:
             return {}
         Agent = persistence_agent.Agent
@@ -1101,7 +1240,11 @@ class MonitoringService:
         created_at_ms = _epoch_ms(row.created_at)
         started_at_ms = _epoch_ms(row.started_at)
         finished_at_ms = _epoch_ms(row.finished_at)
-        is_debug = str(row.binding_id or '').startswith('debug:') or str(row.conversation_id or '').startswith('debug:')
+        is_debug = (
+            str(row.binding_id or '').startswith('debug:')
+            or str(row.conversation_id or '').startswith('debug:')
+            or row.bot_id == 'websocket-proxy-bot'
+        )
         return {
             'source': 'agent',
             'id': row.run_id,
@@ -1109,11 +1252,11 @@ class MonitoringService:
             'status': row.status,
             'status_group': _status_group(row.status, 'agent'),
             'title': metadata.get('event_type') or row.queue_name or row.runner_id,
-            'target_kind': 'agent' if row.agent_id else 'processor',
-            'target_id': row.agent_id or row.binding_id,
+            'target_kind': metadata.get('processor_type') or ('agent' if row.agent_id else 'processor'),
+            'target_id': metadata.get('processor_id') or row.agent_id or row.binding_id,
             'target_name': metadata.get('agent_name')
             or metadata.get('target_name')
-            or (agent_names or {}).get(row.agent_id or ''),
+            or (agent_names or {}).get(metadata.get('processor_id') or row.agent_id or ''),
             'bot_id': row.bot_id,
             'bot_name': metadata.get('bot_name'),
             'pipeline_id': None,
@@ -1130,6 +1273,8 @@ class MonitoringService:
             'queue_name': row.queue_name,
             'debug': is_debug,
             'has_error': row.status in {'failed', 'timeout'},
+            'status_reason': row.status_reason,
+            'input_preview': _message_preview(json.dumps(metadata.get('input') or {}, ensure_ascii=False), 200),
         }
 
     def _serialize_pipeline_execution(self, row) -> dict:
@@ -1137,7 +1282,7 @@ class MonitoringService:
         return {
             'source': 'pipeline',
             'id': row.id,
-            'event_id': None,
+            'event_id': row.event_id,
             'status': row.status,
             'status_group': _status_group(row.status, 'pipeline'),
             'title': _message_preview(row.message_content),
@@ -1158,8 +1303,9 @@ class MonitoringService:
             'usage': None,
             'cost': None,
             'queue_name': None,
-            'debug': False,
+            'debug': row.bot_id == 'websocket-proxy-bot',
             'has_error': row.status == 'error',
+            'input_preview': _message_preview(row.message_content),
             'platform': row.platform,
             'user_id': row.user_id,
             'user_name': row.user_name,
@@ -1173,13 +1319,20 @@ class MonitoringService:
         pipeline_ids: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> dict:
         LLMCall = persistence_monitoring.MonitoringLLMCall
         conditions = [LLMCall.workspace_uuid == workspace_uuid]
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringLLMCall, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(LLMCall.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(LLMCall.pipeline_id.in_(pipeline_ids))
+            conditions.append(self._monitoring_processor_condition(LLMCall, workspace_uuid, pipeline_ids))
         if start_time:
             conditions.append(LLMCall.timestamp >= start_time)
         if end_time:
@@ -1207,10 +1360,18 @@ class MonitoringService:
         pipeline_ids: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
+        statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> dict:
         AgentRun = persistence_agent_run.AgentRun
         run_conditions = self._agent_run_conditions(
-            workspace_uuid, bot_ids=bot_ids, start_time=start_time, end_time=end_time
+            workspace_uuid,
+            bot_ids=bot_ids,
+            processor_ids=pipeline_ids,
+            start_time=start_time,
+            end_time=end_time,
+            statuses=self._raw_agent_statuses(statuses),
+            debug=True if mode == 'debug' else False if mode == 'real' else None,
         )
         run_status_result = await self.ap.persistence_mgr.execute_async(
             sqlalchemy.select(AgentRun.status, sqlalchemy.func.count(AgentRun.id))
@@ -1224,10 +1385,7 @@ class MonitoringService:
             sqlalchemy.select(sqlalchemy.func.count(AgentRun.id)).where(
                 sqlalchemy.and_(
                     *run_conditions,
-                    sqlalchemy.or_(
-                        AgentRun.binding_id.like('debug:%'),
-                        AgentRun.conversation_id.like('debug:%'),
-                    ),
+                    self._debug_run_condition(AgentRun),
                 )
             )
         )
@@ -1252,7 +1410,13 @@ class MonitoringService:
         durations = sorted((row[1] - row[0]).total_seconds() * 1000 for row in timing_result.all() if row[0] and row[1])
 
         pipeline_conditions = self._pipeline_execution_conditions(
-            workspace_uuid, bot_ids=bot_ids, pipeline_ids=pipeline_ids, start_time=start_time, end_time=end_time
+            workspace_uuid,
+            bot_ids=bot_ids,
+            pipeline_ids=pipeline_ids,
+            statuses=self._raw_pipeline_statuses(statuses),
+            start_time=start_time,
+            end_time=end_time,
+            debug=True if mode == 'debug' else False if mode == 'real' else None,
         )
         pipeline_status_result = await self.ap.persistence_mgr.execute_async(
             sqlalchemy.select(
@@ -1264,6 +1428,12 @@ class MonitoringService:
         )
         pipeline_counts = {row[0]: row[1] for row in pipeline_status_result.all()}
         pipeline_total = sum(pipeline_counts.values())
+        pipeline_debug_result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringMessage.id)).where(
+                *pipeline_conditions, self._debug_pipeline_condition(persistence_monitoring.MonitoringMessage)
+            )
+        )
+        debug_count += pipeline_debug_result.scalar() or 0
 
         # Pending interactions are gated by the same bot/window scope as the runs
         # so "waiting input" cannot drift from the rest of the card.
@@ -1273,10 +1443,21 @@ class MonitoringService:
         ]
         if bot_ids:
             waiting_conditions.append(persistence_agent_interaction.AgentInteraction.bot_id.in_(bot_ids))
+        if pipeline_ids:
+            waiting_conditions.append(
+                persistence_agent_interaction.AgentInteraction.run_id.in_(
+                    self._processor_runs(workspace_uuid, pipeline_ids)
+                )
+            )
         if start_time:
             waiting_conditions.append(persistence_agent_interaction.AgentInteraction.created_at >= start_time)
         if end_time:
             waiting_conditions.append(persistence_agent_interaction.AgentInteraction.created_at <= end_time)
+        waiting_conditions.append(
+            persistence_agent_interaction.AgentInteraction.run_id.in_(
+                sqlalchemy.select(AgentRun.run_id).where(*run_conditions)
+            )
+        )
         waiting_result = await self.ap.persistence_mgr.execute_async(
             sqlalchemy.select(sqlalchemy.func.count(persistence_agent_interaction.AgentInteraction.id)).where(
                 sqlalchemy.and_(*waiting_conditions)
@@ -1334,71 +1515,102 @@ class MonitoringService:
         """List executions from both object types (agent runs and pipeline queries)."""
         workspace_uuid = require_workspace_uuid(context)
         limit, offset = self.normalize_page_window(limit, offset)
-        fetch_limit = min(limit + offset, self._configured_query_limit('page_rows', 1000, 5000))
-        if source not in {'all', 'agent', 'pipeline'}:
+        if source not in {'all', 'agent', 'pipeline', 'event'}:
             source = 'all'
         if mode not in {'all', 'real', 'debug'}:
             mode = 'all'
-        debug = None if mode == 'all' else (mode == 'debug')
-
-        include_agent = source in {'all', 'agent'} and (not pipeline_ids or bool(agent_ids))
-        include_pipeline = source in {'all', 'pipeline'} and (not agent_ids or bool(pipeline_ids))
-
-        items: list[dict] = []
-        total = 0
-
-        if include_agent:
-            run_conditions = self._agent_run_conditions(
+        debug = None if mode == 'all' else mode == 'debug'
+        branches = []
+        AgentRun = persistence_agent_run.AgentRun
+        Message = persistence_monitoring.MonitoringMessage
+        if source in {'all', 'agent'}:
+            conditions = self._agent_run_conditions(
                 workspace_uuid,
                 bot_ids=bot_ids,
                 agent_ids=agent_ids,
+                processor_ids=pipeline_ids,
                 statuses=self._raw_agent_statuses(statuses),
                 start_time=start_time,
                 end_time=end_time,
                 debug=debug,
             )
-            AgentRun = persistence_agent_run.AgentRun
-            result = await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.select(AgentRun)
-                .where(sqlalchemy.and_(*run_conditions))
-                .order_by(AgentRun.created_at.desc(), AgentRun.id.desc())
-                .limit(fetch_limit)
+            branches.append(
+                sqlalchemy.select(
+                    sqlalchemy.literal('agent').label('source'),
+                    AgentRun.run_id.label('id'),
+                    AgentRun.created_at.label('time'),
+                ).where(*conditions)
             )
-            rows = self._model_rows(result)
-            count_result = await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.select(sqlalchemy.func.count(AgentRun.id)).where(sqlalchemy.and_(*run_conditions))
-            )
-            total += count_result.scalar() or 0
-            agent_names = await self._agent_names(workspace_uuid, rows)
-            items.extend(self._serialize_agent_execution(row, agent_names=agent_names) for row in rows)
-
-        if include_pipeline and (debug is None or debug is False):
-            message_conditions = self._pipeline_execution_conditions(
+        if source in {'all', 'pipeline'} and (not agent_ids or pipeline_ids):
+            conditions = self._pipeline_execution_conditions(
                 workspace_uuid,
                 bot_ids=bot_ids,
                 pipeline_ids=pipeline_ids,
                 statuses=self._raw_pipeline_statuses(statuses),
                 start_time=start_time,
                 end_time=end_time,
+                debug=debug,
             )
-            MonitoringMessage = persistence_monitoring.MonitoringMessage
-            result = await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.select(MonitoringMessage)
-                .where(sqlalchemy.and_(*message_conditions))
-                .order_by(MonitoringMessage.timestamp.desc())
-                .limit(fetch_limit)
+            branches.append(
+                sqlalchemy.select(
+                    sqlalchemy.literal('pipeline').label('source'),
+                    Message.id.label('id'),
+                    Message.timestamp.label('time'),
+                ).where(*conditions)
             )
-            rows = self._model_rows(result)
-            count_result = await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.select(sqlalchemy.func.count(MonitoringMessage.id)).where(
-                    sqlalchemy.and_(*message_conditions)
+        if source in {'all', 'event'} and mode != 'debug' and not pipeline_ids and not agent_ids:
+            conditions = self._unhandled_event_conditions(workspace_uuid, bot_ids, start_time, end_time)
+            if statuses:
+                conditions.append(self._execution_json_text(EventLog.metadata_json, 'status').in_(statuses))
+            branches.append(
+                sqlalchemy.select(
+                    sqlalchemy.literal('event').label('source'),
+                    EventLog.event_id.label('id'),
+                    EventLog.created_at.label('time'),
+                ).where(*conditions)
+            )
+        total = 0
+        page = []
+        if branches:
+            combined = sqlalchemy.union_all(*branches).subquery()
+            count = await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.select(sqlalchemy.func.count()).select_from(combined)
+            )
+            total = count.scalar() or 0
+            refs = (
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.select(combined)
+                    .order_by(combined.c.time.desc(), combined.c.source, combined.c.id)
+                    .offset(offset)
+                    .limit(limit)
                 )
-            )
-            total += count_result.scalar() or 0
-            items.extend(self._serialize_pipeline_execution(row) for row in rows)
-
-        items.sort(key=lambda item: item.get('created_at_ms') or 0, reverse=True)
-        page = items[offset : offset + limit]
+            ).all()
+            by_key = {}
+            for kind, model, identity_column in [
+                ('agent', AgentRun, AgentRun.run_id),
+                ('pipeline', Message, Message.id),
+                ('event', EventLog, EventLog.event_id),
+            ]:
+                ids = [r.id for r in refs if r.source == kind]
+                if not ids:
+                    continue
+                workspace_column = model.workspace_uuid if kind == 'pipeline' else model.workspace_id
+                records = await self._execution_rows(
+                    workspace_uuid,
+                    sqlalchemy.select(model).where(workspace_column == workspace_uuid, identity_column.in_(ids)),
+                )
+                names = await self._agent_names(workspace_uuid, records) if kind == 'agent' else {}
+                for record in records:
+                    item = (
+                        self._serialize_agent_execution(record, names)
+                        if kind == 'agent'
+                        else self._serialize_pipeline_execution(record)
+                        if kind == 'pipeline'
+                        else self._serialize_event_execution(record)
+                    )
+                    by_key[(kind, item['id'])] = item
+            page = [by_key[(ref.source, ref.id)] for ref in refs if (ref.source, ref.id) in by_key]
+            await self.enrich_execution_rows(workspace_uuid, page)
         return {
             'items': page,
             'total': total,
@@ -1408,6 +1620,8 @@ class MonitoringService:
             'summary': {
                 'executions': await self._get_execution_summary(
                     workspace_uuid,
+                    statuses=statuses,
+                    mode=mode,
                     bot_ids=bot_ids,
                     pipeline_ids=pipeline_ids,
                     start_time=start_time,
@@ -1415,6 +1629,8 @@ class MonitoringService:
                 ),
                 'tokens': await self._get_token_coverage(
                     workspace_uuid,
+                    execution_statuses=statuses,
+                    mode=mode,
                     bot_ids=bot_ids,
                     pipeline_ids=pipeline_ids,
                     start_time=start_time,
@@ -1423,75 +1639,6 @@ class MonitoringService:
             },
         }
 
-    async def get_execution_detail(
-        self,
-        context: TenantContext,
-        source: str,
-        execution_id: str,
-    ) -> dict:
-        """Full trace of a single execution, keyed by object type."""
-        workspace_uuid = require_workspace_uuid(context)
-
-        if source == 'agent':
-            from ....agent.runner.run_ledger_store import RunLedgerStore
-
-            store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine())
-            run = await store.get_run(execution_id)
-            if run is None or run.get('workspace_id') != workspace_uuid:
-                raise ValueError('Execution not found')
-            items, next_cursor, _, has_more = await store.page_run_events(run_id=execution_id, limit=200)
-            return {
-                'source': 'agent',
-                'run': run,
-                'events': items,
-                'has_more': has_more,
-                'next_cursor': next_cursor,
-            }
-
-        if source == 'pipeline':
-            MonitoringMessage = persistence_monitoring.MonitoringMessage
-            result = await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.select(MonitoringMessage).where(
-                    sqlalchemy.and_(
-                        MonitoringMessage.id == execution_id,
-                        MonitoringMessage.workspace_uuid == workspace_uuid,
-                    )
-                )
-            )
-            rows = self._model_rows(result)
-            if not rows:
-                raise ValueError('Execution not found')
-            serialized = self.ap.persistence_mgr.serialize_model(MonitoringMessage, rows[0])
-
-            async def _related(model, order_column):
-                related = await self.ap.persistence_mgr.execute_async(
-                    sqlalchemy.select(model)
-                    .where(
-                        sqlalchemy.and_(
-                            model.workspace_uuid == workspace_uuid,
-                            model.message_id == execution_id,
-                        )
-                    )
-                    .order_by(order_column.asc())
-                )
-                return [self.ap.persistence_mgr.serialize_model(model, row) for row in self._model_rows(related)]
-
-            return {
-                'source': 'pipeline',
-                'message': serialized,
-                'llm_calls': await _related(
-                    persistence_monitoring.MonitoringLLMCall, persistence_monitoring.MonitoringLLMCall.timestamp
-                ),
-                'tool_calls': await _related(
-                    persistence_monitoring.MonitoringToolCall, persistence_monitoring.MonitoringToolCall.timestamp
-                ),
-                'errors': await _related(
-                    persistence_monitoring.MonitoringError, persistence_monitoring.MonitoringError.timestamp
-                ),
-            }
-
-        raise ValueError('Unknown execution source')
-
     async def get_token_statistics(
         self,
         context: TenantContext,
@@ -1500,6 +1647,8 @@ class MonitoringService:
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
         bucket: str = 'hour',
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> dict:
         """Get detailed token usage statistics for production observability.
 
@@ -1518,10 +1667,15 @@ class MonitoringService:
             bucket = 'hour'
 
         conditions = [LLMCall.workspace_uuid == workspace_uuid]
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringLLMCall, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(LLMCall.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(LLMCall.pipeline_id.in_(pipeline_ids))
+            conditions.append(self._monitoring_processor_condition(LLMCall, workspace_uuid, pipeline_ids))
         if start_time:
             conditions.append(LLMCall.timestamp >= start_time)
         if end_time:
@@ -1702,16 +1856,27 @@ class MonitoringService:
         end_time: datetime.datetime | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get messages with filters"""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringMessage.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringMessage, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringMessage.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringMessage.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringMessage, workspace_uuid, pipeline_ids
+                )
+            )
         if session_ids:
             conditions.append(persistence_monitoring.MonitoringMessage.session_id.in_(session_ids))
         if start_time:
@@ -1757,16 +1922,27 @@ class MonitoringService:
         end_time: datetime.datetime | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get LLM calls with filters"""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringLLMCall.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringLLMCall, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringLLMCall.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringLLMCall.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringLLMCall, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringLLMCall.timestamp >= start_time)
         if end_time:
@@ -1812,16 +1988,27 @@ class MonitoringService:
         end_time: datetime.datetime | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get tool calls with filters"""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringToolCall.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringToolCall, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringToolCall.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringToolCall.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringToolCall, workspace_uuid, pipeline_ids
+                )
+            )
         if session_ids:
             conditions.append(persistence_monitoring.MonitoringToolCall.session_id.in_(session_ids))
         if start_time:
@@ -1865,12 +2052,19 @@ class MonitoringService:
         knowledge_base_id: str | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get embedding calls with filters"""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringEmbeddingCall.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringEmbeddingCall, workspace_uuid, execution_statuses, mode
+            )
+        )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringEmbeddingCall.timestamp >= start_time)
         if end_time:
@@ -1919,16 +2113,27 @@ class MonitoringService:
         is_active: bool | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get sessions with filters"""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringSession, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringSession.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringSession, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time)
         if end_time:
@@ -1983,16 +2188,27 @@ class MonitoringService:
         end_time: datetime.datetime | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get errors with filters"""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringError.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringError, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringError.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringError.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringError, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringError.timestamp >= start_time)
         if end_time:
@@ -2458,7 +2674,11 @@ class MonitoringService:
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringMessage.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringMessage.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringMessage, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringMessage.timestamp >= start_time)
         if end_time:
@@ -2514,7 +2734,11 @@ class MonitoringService:
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringLLMCall.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringLLMCall.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringLLMCall, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringLLMCall.timestamp >= start_time)
         if end_time:
@@ -2621,7 +2845,11 @@ class MonitoringService:
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringError.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringError.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringError, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringError.timestamp >= start_time)
         if end_time:
@@ -2672,7 +2900,11 @@ class MonitoringService:
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringSession.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringSession, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time)
         if end_time:
@@ -2825,6 +3057,8 @@ class MonitoringService:
         pipeline_ids: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> dict:
         """Get feedback statistics.
 
@@ -2834,10 +3068,19 @@ class MonitoringService:
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringFeedback.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringFeedback, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringFeedback.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringFeedback.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringFeedback, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringFeedback.timestamp >= start_time)
         if end_time:
@@ -2918,16 +3161,27 @@ class MonitoringService:
         end_time: datetime.datetime | None = None,
         limit: int = 100,
         offset: int = 0,
+        execution_statuses: list[str] | None = None,
+        mode: str = 'all',
     ) -> tuple[list[dict], int]:
         """Get feedback list with filters."""
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         conditions = [persistence_monitoring.MonitoringFeedback.workspace_uuid == workspace_uuid]
 
+        conditions.append(
+            self._monitoring_execution_condition(
+                persistence_monitoring.MonitoringFeedback, workspace_uuid, execution_statuses, mode
+            )
+        )
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringFeedback.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringFeedback.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringFeedback, workspace_uuid, pipeline_ids
+                )
+            )
         if feedback_type is not None:
             conditions.append(persistence_monitoring.MonitoringFeedback.feedback_type == feedback_type)
         if start_time:
@@ -2980,7 +3234,11 @@ class MonitoringService:
         if bot_ids:
             conditions.append(persistence_monitoring.MonitoringFeedback.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(persistence_monitoring.MonitoringFeedback.pipeline_id.in_(pipeline_ids))
+            conditions.append(
+                self._monitoring_processor_condition(
+                    persistence_monitoring.MonitoringFeedback, workspace_uuid, pipeline_ids
+                )
+            )
         if start_time:
             conditions.append(persistence_monitoring.MonitoringFeedback.timestamp >= start_time)
         if end_time:

src/langbot/pkg/api/http/service/monitoring_execution_details.py

@@ -0,0 +1,580 @@
+"""Tenant-scoped execution inputs, outputs, related runs and conversation history."""
+
+from __future__ import annotations
+
+import datetime
+import json
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ....entity.persistence.agent_run import AgentRun, AgentRunEvent
+from ....entity.persistence.event_log import EventLog
+from ....entity.persistence.transcript import Transcript
+from ....entity.persistence import monitoring as models
+from .tenant import require_workspace_uuid
+
+
+def epoch(value):
+    if value is None:
+        return None
+    return int(value.replace(tzinfo=datetime.timezone.utc).timestamp() * 1000)
+
+
+def parsed(value, fallback=None):
+    try:
+        return json.loads(value) if value else fallback
+    except (TypeError, ValueError):
+        return fallback
+
+
+def event_content(content, event_data):
+    """Keep event-specific properties alongside the normalized message input."""
+    if not isinstance(event_data, dict) or not event_data:
+        return content
+    if not isinstance(content, dict):
+        return event_data if not content else {**event_data, 'input': content}
+    merged = dict(event_data)
+    conflicts = {}
+    for key, value in content.items():
+        if key not in merged:
+            merged[key] = value
+        elif merged[key] != value and value is not None and value != '' and value != [] and value != {}:
+            conflicts[key] = value
+    if conflicts:
+        # Never silently overwrite a custom field that shares an input name.
+        return {'event': merged, 'input': conflicts}
+    return merged
+
+
+def event_item(row):
+    metadata = parsed(row.metadata_json, {})
+    return {
+        'id': row.event_id,
+        'event_type': row.event_type,
+        'source': row.source,
+        'timestamp_ms': epoch(row.event_time or row.created_at),
+        'actor_id': row.actor_id,
+        'actor_name': row.actor_name,
+        'conversation_id': row.conversation_id,
+        'thread_id': row.thread_id,
+        'content': event_content(parsed(row.input_json, None) or row.input_summary, metadata.get('input_event')),
+        'metadata': metadata,
+    }
+
+
+def message_item(row):
+    return {
+        'id': row.id,
+        'role': row.role or 'user',
+        'timestamp_ms': epoch(row.timestamp),
+        'content': parsed(row.message_content, None) or row.message_content,
+        'actor_id': row.user_id,
+        'actor_name': row.user_name,
+        'status': row.status,
+        'event_id': row.event_id,
+        'run_id': row.run_id,
+        'origin': 'delivery' if row.role == 'assistant' else 'input',
+        'delivery': parsed(row.variables, {}) if row.role == 'assistant' else None,
+    }
+
+
+def transcript_item(row):
+    return {
+        'id': row.transcript_id,
+        'role': row.role,
+        'timestamp_ms': epoch(row.created_at),
+        'content': parsed(row.content_json, None) or row.content,
+        'attachments': parsed(row.attachment_refs_json, []),
+        'event_id': row.event_id,
+        'run_id': row.run_id,
+        'origin': 'generated' if row.role == 'assistant' else 'input',
+    }
+
+
+class ExecutionDetailsMixin:
+    def _execution_json_text(self, column, key):
+        # SQLite JSON functions accept TEXT directly; PostgreSQL operators
+        # require an explicit JSON cast for the existing text-backed journals.
+        if self.ap.persistence_mgr.get_db_engine().dialect.name == 'sqlite':
+            expression = sa.type_coerce(column, sa.JSON)
+        else:
+            expression = sa.cast(column, sa.JSON)
+        return expression[key].as_string()
+
+    async def link_execution_message(self, context, message_id, run_id, event_id):
+        workspace = self._require_write_context(context)
+        statement = (
+            sa.update(models.MonitoringMessage)
+            .where(
+                models.MonitoringMessage.workspace_uuid == workspace,
+                models.MonitoringMessage.id == message_id,
+            )
+            .values(run_id=run_id, event_id=sa.func.coalesce(models.MonitoringMessage.event_id, event_id))
+        )
+        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
+        if callable(tenant_uow):
+            async with tenant_uow(workspace) as uow:
+                await uow.session.execute(statement)
+        else:
+            await self.ap.persistence_mgr.execute_async(statement)
+
+    async def record_ingress_event(
+        self,
+        context,
+        *,
+        event_id,
+        event_type,
+        bot_id,
+        bot_name,
+        platform,
+        payload,
+        actor_id=None,
+        actor_name=None,
+        status='running',
+        routes=None,
+    ):
+        """Persist entry facts even when routing never creates a processor run."""
+        from ....agent.runner.event_log_store import EventLogStore
+        from .monitoring import _message_preview
+
+        workspace = self._require_write_context(context)
+        store = EventLogStore(self.ap.persistence_mgr.get_db_engine())
+        # Reuse the message sanitizer (including inline-media handling) instead
+        # of storing an unbounded adapter object in the event journal.
+        content = parsed(self._sanitize_message_content(json.dumps(payload, ensure_ascii=False, default=str)), {})
+        summary = _message_preview(json.dumps(content, ensure_ascii=False), 1000) if content else event_type
+        await store.append_event(
+            event_id=event_id,
+            event_type=event_type,
+            source='platform',
+            workspace_id=workspace,
+            bot_id=bot_id,
+            actor_id=actor_id,
+            actor_name=actor_name,
+            actor_type='user' if actor_id else 'system',
+            input_summary=summary,
+            input_json=content,
+            metadata={'status': status, 'bot_name': bot_name, 'platform': platform},
+        )
+        if routes is not None:
+            safe_routes = []
+            for route in routes:
+                if isinstance(route, BaseException):
+                    safe_routes.append({'status': 'failed', 'reason': str(route)})
+                elif isinstance(route, dict):
+                    safe_routes.append(
+                        {
+                            key: route.get(key)
+                            for key in (
+                                'status',
+                                'reason',
+                                'target_type',
+                                'target_uuid',
+                                'run_id',
+                                'failure_code',
+                            )
+                        }
+                    )
+            metadata = {'status': status, 'bot_name': bot_name, 'platform': platform, 'routes': safe_routes}
+            async with AsyncSession(self.ap.persistence_mgr.get_db_engine()) as session:
+                await session.execute(
+                    sa.update(EventLog)
+                    .where(
+                        EventLog.workspace_id == workspace,
+                        EventLog.event_id == event_id,
+                    )
+                    .values(metadata_json=json.dumps(metadata, ensure_ascii=False))
+                )
+                await session.commit()
+
+    async def _execution_query(self, workspace, statement):
+        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
+        if callable(tenant_uow):
+            async with tenant_uow(workspace) as uow:
+                return await uow.session.execute(statement)
+        async with AsyncSession(self.ap.persistence_mgr.get_db_engine()) as session:
+            return await session.execute(statement)
+
+    async def _execution_rows(self, workspace, statement):
+        return list((await self._execution_query(workspace, statement)).scalars().all())
+
+    async def _execution_page(self, workspace, statement, convert, offset, limit):
+        rows = await self._execution_rows(workspace, statement.offset(offset).limit(limit + 1))
+        return {
+            'items': [convert(row) for row in rows[:limit]],
+            'has_more': len(rows) > limit,
+            'next_offset': offset + min(len(rows), limit),
+        }
+
+    def _unhandled_event_conditions(self, workspace, bot_ids=None, start_time=None, end_time=None):
+        # Absorbed steering events belong to an existing run, too.
+        conditions = [
+            EventLog.workspace_id == workspace,
+            ~sa.exists(
+                sa.select(AgentRun.id).where(
+                    AgentRun.workspace_id == workspace,
+                    sa.or_(AgentRun.event_id == EventLog.event_id, AgentRun.run_id == EventLog.run_id),
+                )
+            ),
+            ~sa.exists(
+                sa.select(models.MonitoringMessage.id).where(
+                    models.MonitoringMessage.workspace_uuid == workspace,
+                    models.MonitoringMessage.event_id == EventLog.event_id,
+                )
+            ),
+        ]
+        if bot_ids:
+            conditions.append(EventLog.bot_id.in_(bot_ids))
+        if start_time:
+            conditions.append(EventLog.created_at >= start_time)
+        if end_time:
+            conditions.append(EventLog.created_at <= end_time)
+        return conditions
+
+    def _serialize_event_execution(self, row):
+        metadata = parsed(row.metadata_json, {})
+        status = metadata.get('status', 'ignored')
+        return {
+            'source': 'event',
+            'id': row.event_id,
+            'event_id': row.event_id,
+            'status': status,
+            'status_group': status,
+            'title': row.event_type,
+            'input_preview': row.input_summary or '',
+            'target_kind': 'event',
+            'target_id': None,
+            'target_name': None,
+            'bot_id': row.bot_id,
+            'bot_name': metadata.get('bot_name'),
+            'platform': metadata.get('platform') or row.source,
+            'user_id': row.actor_id,
+            'user_name': row.actor_name,
+            'conversation_id': row.conversation_id,
+            'session_id': row.conversation_id,
+            'created_at_ms': epoch(row.created_at),
+            'started_at_ms': None,
+            'finished_at_ms': None,
+            'duration_ms': None,
+            'usage': None,
+            'cost': None,
+            'queue_name': None,
+            'pipeline_id': None,
+            'pipeline_name': None,
+            'runner_id': None,
+            'debug': row.source == 'debug',
+            'has_error': status == 'failed',
+            'status_reason': metadata.get('reason'),
+        }
+
+    async def enrich_execution_rows(self, workspace, items):
+        ids = {item['event_id'] for item in items if item.get('event_id')}
+        if not ids:
+            return
+        events = await self._execution_rows(
+            workspace,
+            sa.select(EventLog).where(
+                EventLog.workspace_id == workspace,
+                EventLog.event_id.in_(ids),
+            ),
+        )
+        by_id = {row.event_id: row for row in events}
+        for item in items:
+            event = by_id.get(item.get('event_id'))
+            if event:
+                meta = parsed(event.metadata_json, {})
+                item.update(
+                    event_type=event.event_type,
+                    input_preview=event.input_summary or item.get('input_preview') or '',
+                    user_id=event.actor_id,
+                    user_name=event.actor_name,
+                    platform=meta.get('platform') or event.source,
+                    bot_name=item.get('bot_name') or meta.get('bot_name'),
+                )
+
+    async def get_execution_detail(self, context, source, execution_id, *, section=None, offset=0, limit=100):
+        workspace = require_workspace_uuid(context)
+        limit, offset = self.normalize_page_window(limit, offset)
+        limit = min(limit, 100)
+        sections = {
+            'inputs',
+            'outputs',
+            'deliveries',
+            'conversation',
+            'related',
+            'events',
+            'llm_calls',
+            'tool_calls',
+            'errors',
+        }
+        if section is not None and section not in sections:
+            raise ValueError('Unknown execution section')
+
+        run = message = event = None
+        if source in {'agent', 'auto'}:
+            rows = await self._execution_rows(
+                workspace,
+                sa.select(AgentRun).where(
+                    AgentRun.workspace_id == workspace,
+                    AgentRun.run_id == execution_id,
+                ),
+            )
+            run = rows[0] if rows else None
+        if run is None and source in {'pipeline', 'auto'}:
+            rows = await self._execution_rows(
+                workspace,
+                sa.select(models.MonitoringMessage).where(
+                    models.MonitoringMessage.workspace_uuid == workspace,
+                    models.MonitoringMessage.id == execution_id,
+                ),
+            )
+            message = rows[0] if rows else None
+            if message and message.parent_message_id:
+                return await self.get_execution_detail(
+                    context, 'auto', message.parent_message_id, section=section, offset=offset, limit=limit
+                )
+        if run is None and message is None and source in {'event', 'auto'}:
+            rows = await self._execution_rows(
+                workspace,
+                sa.select(EventLog).where(
+                    EventLog.workspace_id == workspace,
+                    EventLog.event_id == execution_id,
+                ),
+            )
+            event = rows[0] if rows else None
+        if run is None and message is None and event is None:
+            raise ValueError('Execution not found')
+
+        if message and message.run_id:
+            linked = await self._execution_rows(
+                workspace,
+                sa.select(AgentRun).where(AgentRun.workspace_id == workspace, AgentRun.run_id == message.run_id),
+            )
+            if linked:
+                linked_run = linked[0]
+            else:
+                linked_run = None
+        else:
+            linked_run = None
+
+        if run:
+            names = await self._agent_names(workspace, [run])
+            row = self._serialize_agent_execution(run, names)
+            event_id, bot, conversation, thread = run.event_id, run.bot_id, run.conversation_id, run.thread_id
+            run_id = run.run_id
+        elif message:
+            row = self._serialize_pipeline_execution(message)
+            event_id, bot, conversation, thread = message.event_id, message.bot_id, message.session_id, None
+            run_id = message.run_id
+            if linked_run:
+                row['duration_ms'] = (
+                    int((linked_run.finished_at - linked_run.started_at).total_seconds() * 1000)
+                    if linked_run.started_at and linked_run.finished_at
+                    else None
+                )
+                row['usage'] = parsed(linked_run.usage_json)
+                row['status_reason'] = linked_run.status_reason
+        else:
+            row = self._serialize_event_execution(event)
+            event_id, bot, conversation, thread = event.event_id, event.bot_id, event.conversation_id, event.thread_id
+            run_id = event.run_id
+        await self.enrich_execution_rows(workspace, [row])
+
+        ledger = run or linked_run
+        ingress_id = parsed(ledger.metadata_json, {}).get('ingress_event_id') if ledger else None
+        event_conditions = [
+            EventLog.workspace_id == workspace,
+            sa.or_(
+                EventLog.event_id == ingress_id if ingress_id else sa.false(),
+                EventLog.event_id == ledger.event_id if ledger else sa.false(),
+                EventLog.event_id == event_id if event_id else sa.false(),
+                EventLog.run_id == run_id if run_id else sa.false(),
+            ),
+        ]
+        event_ids = sa.select(EventLog.event_id).where(*event_conditions)
+        # All joins use explicit identities. Session/time matching is reserved
+        # for the visibly separate conversation history section.
+        message_scope = [models.MonitoringMessage.workspace_uuid == workspace]
+        related_message = sa.or_(
+            models.MonitoringMessage.id == message.id if message else sa.false(),
+            models.MonitoringMessage.parent_message_id == (message.id if message else run_id)
+            if message or run_id
+            else sa.false(),
+            models.MonitoringMessage.run_id == run_id if run_id else sa.false(),
+        )
+        transcript_scope = [Transcript.workspace_id == workspace, Transcript.bot_id == bot]
+        transcripts = sa.select(Transcript).where(
+            *transcript_scope,
+            sa.or_(
+                Transcript.run_id == run_id if run_id else sa.false(),
+                sa.and_(Transcript.role == 'user', Transcript.event_id.in_(event_ids)),
+            ),
+        )
+        call_ids = sa.select(models.MonitoringMessage.id).where(*message_scope, related_message)
+        statements = {
+            'inputs': (sa.select(EventLog).where(*event_conditions).order_by(EventLog.id.asc()), event_item),
+            'outputs': (transcripts.where(Transcript.role != 'user').order_by(Transcript.seq.asc()), transcript_item),
+            'deliveries': (
+                sa.select(models.MonitoringMessage)
+                .where(*message_scope, related_message, models.MonitoringMessage.role == 'assistant')
+                .order_by(models.MonitoringMessage.timestamp.asc(), models.MonitoringMessage.id.asc()),
+                message_item,
+            ),
+            'events': (
+                sa.select(AgentRunEvent)
+                .where(AgentRunEvent.run_id == run_id if run_id else sa.false())
+                .order_by(AgentRunEvent.sequence.asc()),
+                lambda r: {
+                    'id': r.id,
+                    'sequence': r.sequence,
+                    'type': r.type,
+                    'data': parsed(r.data_json, {}),
+                    'timestamp_ms': epoch(r.created_at),
+                },
+            ),
+        }
+        for name, model in [
+            ('llm_calls', models.MonitoringLLMCall),
+            ('tool_calls', models.MonitoringToolCall),
+            ('errors', models.MonitoringError),
+        ]:
+            statements[name] = (
+                sa.select(model)
+                .where(
+                    model.workspace_uuid == workspace,
+                    sa.or_(
+                        model.message_id.in_(call_ids),
+                        model.message_id == run_id if run_id else sa.false(),
+                    ),
+                )
+                .order_by(model.timestamp.asc(), model.id.asc()),
+                lambda r, model=model: self.ap.persistence_mgr.serialize_model(model, r),
+            )
+        if message:
+            statements['conversation'] = (
+                sa.select(models.MonitoringMessage)
+                .where(
+                    *message_scope,
+                    models.MonitoringMessage.bot_id == bot,
+                    models.MonitoringMessage.session_id == conversation,
+                )
+                .order_by(models.MonitoringMessage.timestamp.desc(), models.MonitoringMessage.id.desc()),
+                message_item,
+            )
+        else:
+            statements['conversation'] = (
+                sa.select(Transcript)
+                .where(
+                    *transcript_scope,
+                    Transcript.conversation_id == conversation if conversation else sa.false(),
+                    Transcript.thread_id == thread,
+                )
+                .order_by(Transcript.seq.desc()),
+                transcript_item,
+            )
+
+        pages = {}
+        for name, (query, convert) in statements.items():
+            if section is None or section == name:
+                pages[name] = await self._execution_page(workspace, query, convert, offset, limit)
+        if run:
+            meta = parsed(run.metadata_json, {})
+            for item in pages.get('inputs', {}).get('items', []):
+                # Older event logs omitted data, but the root run may retain it.
+                # Do not apply the root snapshot to later steering events.
+                if item['id'] == run.event_id and not item.get('metadata', {}).get('input_event'):
+                    item['content'] = event_content(item['content'], meta.get('input_event'))
+        if message and (section is None or section == 'inputs'):
+            # The original rich message is authoritative even when the event
+            # log is absent or only retained a compact event summary.
+            if offset == 0:
+                pages['inputs']['items'].insert(0, message_item(message))
+        if (
+            run
+            and not pages.get('inputs', {}).get('items')
+            and offset == 0
+            and (section is None or section == 'inputs')
+        ):
+            meta = parsed(run.metadata_json, {})
+            content = event_content(meta.get('input'), meta.get('input_event'))
+            if content:
+                pages['inputs']['items'] = [
+                    {'id': run.event_id or run.run_id, 'content': content, 'timestamp_ms': epoch(run.created_at)}
+                ]
+
+        if section is None or section == 'related':
+            # Union of ledger executions and legacy pipeline roots. SQL paging
+            # avoids silently dropping siblings on highly fanned-out events.
+            related_runs = sa.select(
+                sa.literal('agent').label('source'), AgentRun.run_id.label('id'), AgentRun.created_at.label('time')
+            ).where(
+                AgentRun.workspace_id == workspace,
+                sa.or_(
+                    AgentRun.event_id.in_(event_ids),
+                    self._execution_json_text(AgentRun.metadata_json, 'ingress_event_id').in_(event_ids),
+                ),
+                AgentRun.run_id != (run_id or ''),
+                ~sa.exists().where(
+                    models.MonitoringMessage.workspace_uuid == workspace,
+                    models.MonitoringMessage.run_id == AgentRun.run_id,
+                    models.MonitoringMessage.role != 'assistant',
+                ),
+            )
+            related_messages = sa.select(
+                sa.literal('pipeline'), models.MonitoringMessage.id, models.MonitoringMessage.timestamp
+            ).where(
+                *message_scope,
+                models.MonitoringMessage.event_id.in_(event_ids),
+                models.MonitoringMessage.role != 'assistant',
+                models.MonitoringMessage.id != (message.id if message else ''),
+            )
+            union = sa.union_all(related_runs, related_messages).subquery()
+            refs = (
+                await self._execution_query(
+                    workspace,
+                    sa.select(union).order_by(union.c.time, union.c.source, union.c.id).offset(offset).limit(limit + 1),
+                )
+            ).all()
+            related = []
+            for ref in refs[:limit]:
+                if ref.source == 'agent':
+                    records = await self._execution_rows(
+                        workspace,
+                        sa.select(AgentRun).where(AgentRun.workspace_id == workspace, AgentRun.run_id == ref.id),
+                    )
+                    related.append(
+                        self._serialize_agent_execution(records[0], await self._agent_names(workspace, records))
+                    )
+                else:
+                    records = await self._execution_rows(
+                        workspace,
+                        sa.select(models.MonitoringMessage).where(
+                            *message_scope, models.MonitoringMessage.id == ref.id
+                        ),
+                    )
+                    related.append(self._serialize_pipeline_execution(records[0]))
+            pages['related'] = {
+                'items': related,
+                'has_more': len(refs) > limit,
+                'next_offset': offset + min(len(refs), limit),
+            }
+
+        result = {
+            'source': row['source'],
+            'row': row,
+            'pages': pages,
+            'legacy_context': bool(message and not message.event_id),
+        }
+        if run:
+            # Do not expose runtime authorization, leases or claim tokens.
+            result['run'] = {'run_id': run.run_id, 'status': run.status, 'status_reason': run.status_reason}
+        if message:
+            result['message'] = self.ap.persistence_mgr.serialize_model(models.MonitoringMessage, message)
+        # Backwards-compatible fields for existing API consumers.
+        for key in ('events', 'llm_calls', 'tool_calls', 'errors'):
+            if key in pages:
+                result[key] = pages[key]['items']
+        result['has_more'] = pages.get('events', {}).get('has_more', False)
+        result['next_cursor'] = pages.get('events', {}).get('next_offset')
+        return result

src/langbot/pkg/api/http/service/monitoring_traffic.py

@@ -24,9 +24,14 @@ async def get_traffic_series(
     pipeline_ids: list[str] | None = None,
     start_time: datetime.datetime | None = None,
     end_time: datetime.datetime | None = None,
+    execution_statuses: list[str] | None = None,
+    mode: str = 'all',
 ) -> dict:
     """Count all matching records in UTC buckets, returning at most 1000 points."""
     workspace_uuid = require_workspace_uuid(context)
+    from .monitoring import MonitoringService
+
+    service = MonitoringService(ap)
     bucket = 'hour' if start_time and end_time and end_time - start_time <= datetime.timedelta(days=7) else 'day'
     step = datetime.timedelta(hours=1) if bucket == 'hour' else datetime.timedelta(days=1)
     postgres = ap.persistence_mgr.get_db_engine().dialect.name == 'postgresql'
@@ -40,10 +45,11 @@ async def get_traffic_series(
             pattern = '%Y-%m-%dT%H:00:00' if bucket == 'hour' else '%Y-%m-%dT00:00:00'
             time_bucket = sqlalchemy.func.strftime(pattern, timestamp)
         conditions = [model.workspace_uuid == workspace_uuid]
+        conditions.append(service._monitoring_execution_condition(model, workspace_uuid, execution_statuses, mode))
         if bot_ids:
             conditions.append(model.bot_id.in_(bot_ids))
         if pipeline_ids:
-            conditions.append(model.pipeline_id.in_(pipeline_ids))
+            conditions.append(service._monitoring_processor_condition(model, workspace_uuid, pipeline_ids))
         if start_time is not None:
             conditions.append(timestamp >= start_time)
         if end_time is not None: