diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index d96e56f2f..2647573b4 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -174,3 +174,22 @@ Legacy EventListener plugins remain in the Pipeline lifecycle. `list_processor_runs` includes `created_at_ms`, `started_at_ms`, and `finished_at_ms`: Host lifecycle times in epoch milliseconds. Use the start and finish times for elapsed processing time; select a run and call `get_processor_run_events` for its logs and action results. These times are not internal plugin profiling data. + +## Unified execution monitoring + +Use `get_monitoring_executions` for the execution list and +its legacy `pipeline_ids` parameter to filter any processor kind (Agent, +Pipeline or event processor); the summary uses the same processor scope. Use +`get_monitoring_execution_detail` with `source=auto` to resolve a run, +message or event identifier outside the current list page. The detail exposes +`inputs`, `outputs` (generated content), `deliveries` (recorded platform sends), +`events`, `llm_calls`, `tool_calls`, `errors`, `related`, and `conversation` in +`pages`. Follow each section's `has_more` and `next_offset` independently. +Conversation history supplies context; historical messages without explicit +links must not be asserted to belong to the selected execution. Events without +a processor run use `source=event`. All lookups remain Workspace-scoped. + +Monitoring record filters accept `mode` (`all`, `real`, `debug`) and +`execution_statuses` (normalized execution statuses). These select the owning +execution, not the individual model/tool call outcome. Calls without a recorded +execution link are excluded when an execution filter is active. diff --git a/src/langbot/pkg/agent/runner/event_log_store.py b/src/langbot/pkg/agent/runner/event_log_store.py index 1dbd8a8b2..10b3a1860 100644 --- a/src/langbot/pkg/agent/runner/event_log_store.py +++ b/src/langbot/pkg/agent/runner/event_log_store.py @@ -128,7 +128,33 @@ class EventLogStore: created_at=as_naive_utc(_utc_now()), ) session.add(event) - await session.commit() + try: + await session.commit() + except sqlalchemy.exc.IntegrityError: + await session.rollback() + existing = ( + (await session.execute(sqlalchemy.select(EventLog).where(EventLog.event_id == event_id))) + .scalars() + .first() + ) + if existing is None or existing.workspace_id != workspace_id or existing.bot_id != bot_id: + raise + # Complete the ingress snapshot once a runner claims it. A + # second subscriber must not replace another run's identity. + for field, value in { + 'run_id': run_id, + 'runner_id': runner_id, + 'conversation_id': conversation_id, + 'thread_id': thread_id, + 'actor_id': actor_id, + 'actor_name': actor_name, + }.items(): + if getattr(existing, field) is None and value is not None: + setattr(existing, field, value) + if input_json and ('contents' in input_json or 'attachments' in input_json): + existing.input_json = json.dumps(input_json) + existing.input_summary = input_summary + await session.commit() return event_id diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 9d3db00d2..74f862d79 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -264,6 +264,11 @@ class AgentRunOrchestrator: context=context, authorization=run_authorization, ) + monitoring_id = (execution_query.variables or {}).get('_monitoring_message_id') + if monitoring_id and monitoring_id != run_id: + await self.ap.monitoring_service.link_execution_message( + get_query_execution_context(execution_query), monitoring_id, run_id, event.event_id + ) await self._session_registry.register( run_id=run_id, runner_id=descriptor.id, @@ -346,16 +351,25 @@ class AgentRunOrchestrator: ): continue + journal_started = time.monotonic() await self.journal.append_run_result( result_dict=result_dict, run_id=run_id, sequence=sequence_int, ) + journal_ms = (time.monotonic() - journal_started) * 1000 # Trusted Host observers receive validated events before message-only normalization. result_observer = (adapter_context or {}).get('_result_observer') + observer_started = time.monotonic() if result_observer is not None: await result_observer(result_dict) + observer_ms = (time.monotonic() - observer_started) * 1000 + if journal_ms + observer_ms >= 100: + self.ap.logger.debug( + f'Slow runner event consumer run={run_id} type={result_type} ' + f'journal_ms={journal_ms:.1f} observer_ms={observer_ms:.1f}' + ) if result_type == 'state.updated': await self.journal.handle_state_updated_event( @@ -441,9 +455,15 @@ class AgentRunOrchestrator: terminal_status or '', 'unknown' ) execution_error = '' if execution_outcome == 'success' else (terminal_reason or execution_outcome) - except asyncio.CancelledError: + except (asyncio.CancelledError, GeneratorExit): execution_outcome = 'cancelled' execution_error = 'cancelled' + await self.journal.finalize_run( + run_id=run_id, + status='cancelled', + status_reason='consumer_cancelled', + usage=terminal_usage, + ) raise except Exception as exc: execution_outcome = 'timeout' if self._is_deadline_exhausted(context) else 'failed' diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 70795f304..6591542ed 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -743,10 +743,43 @@ async def execute_platform_tool( from ...telemetry.platform import processing_mode token = processing_mode.set(authorization.get('processor_type', 'none')) + delivery_error = None try: return await api_func(**normalized) + except Exception as exc: + delivery_error = str(exc) + raise finally: processing_mode.reset(token) + if definition.api == 'send_message' and getattr(ap, 'monitoring_service', None) is not None: + try: + import json + + chain = normalized.get('message') + content = chain.model_dump(mode='json') if hasattr(chain, 'model_dump') else str(chain or '') + await ap.monitoring_service.record_message( + execution_context, + bot_id=bot_id, + bot_name=str(getattr(getattr(bot, 'bot_entity', None), 'name', bot_id)), + pipeline_id=str(authorization.get('processor_id') or ''), + pipeline_name=str(authorization.get('processor_id') or authorization.get('runner_id') or ''), + message_content=json.dumps(content, ensure_ascii=False), + session_id=str(authorization.get('conversation_id') or ''), + role='assistant', + status='error' if delivery_error else 'success', + level='error' if delivery_error else 'info', + run_id=execution_id, + parent_message_id=execution_id, + variables=json.dumps( + { + 'delivery_error': delivery_error, + 'target_type': normalized.get('target_type'), + 'target_id': normalized.get('target_id'), + } + ), + ) + except Exception as exc: + ap.logger.warning(f'Failed to record platform delivery: {exc}') finally: reset_execution_id(execution_token) diff --git a/src/langbot/pkg/agent/runner/run_journal.py b/src/langbot/pkg/agent/runner/run_journal.py index dd2bfbf27..3a0217f89 100644 --- a/src/langbot/pkg/agent/runner/run_journal.py +++ b/src/langbot/pkg/agent/runner/run_journal.py @@ -68,6 +68,26 @@ class AgentRunJournal: def _sanitize_attachments(cls, attachments: typing.Iterable[typing.Any]) -> list[dict[str, typing.Any]]: return [cls._sanitize_attachment_ref(attachment) for attachment in attachments] + @classmethod + def _sanitize_event_data(cls, value: typing.Any) -> typing.Any: + """Retain arbitrary event fields without copying inline binary payloads.""" + if hasattr(value, 'model_dump'): + value = value.model_dump(mode='json') + if isinstance(value, list): + return [cls._sanitize_event_data(item) for item in value] + if not isinstance(value, dict): + return value + result = {} + for key, item in value.items(): + if key in ('image_base64', 'file_base64', 'base64') and item: + result[key] = None + result['content_redacted'] = True + elif key == 'attachments' and isinstance(item, list): + result[key] = [cls._sanitize_event_data(cls._sanitize_attachment_ref(ref)) for ref in item] + else: + result[key] = cls._sanitize_event_data(item) + return result + async def create_run( self, *, @@ -78,6 +98,9 @@ class AgentRunJournal: authorization: dict[str, typing.Any], ) -> dict[str, typing.Any]: """Create the Host-owned run ledger record.""" + from ...telemetry import trace + + trace_state = trace.current() runtime = context.get('runtime') if isinstance(context, dict) else {} return await self._get_run_ledger_store().create_run( run_id=context['run_id'], @@ -93,17 +116,21 @@ class AgentRunJournal: authorization=authorization, metadata={ 'event_type': event.event_type, + 'ingress_event_id': trace_state.event_id if trace_state else event.event_id, 'source': event.source, 'processor_id': binding.processor_id, 'processor_type': binding.processor_type, **( - {'input_event': event.data, 'delivery': event.delivery.model_dump(mode='json')} + { + 'input_event': self._sanitize_event_data(event.data), + 'delivery': event.delivery.model_dump(mode='json'), + } if binding.processor_type == 'event_processor' else {} ), **( { - **({'input_event': event.data} if event.event_type != 'message.received' else {}), + 'input_event': self._sanitize_event_data(event.data), 'input': { 'text': context.get('input', {}).get('text', ''), 'contents': self._sanitize_contents(context.get('input', {}).get('contents', [])), @@ -260,7 +287,7 @@ class AgentRunJournal: event_time=( datetime.datetime.fromtimestamp(event.event_time, datetime.timezone.utc) if event.event_time else None ), - metadata=metadata, + metadata={**(metadata or {}), 'input_event': self._sanitize_event_data(event.data)}, ) async def write_user_transcript( diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 1d9656761..8df374b98 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/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, diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index 5cfcf116d..932d436fd 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/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, diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index bb7407248..653be4c93 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/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: diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py new file mode 100644 index 000000000..1b5fa94f9 --- /dev/null +++ b/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 diff --git a/src/langbot/pkg/api/http/service/monitoring_traffic.py b/src/langbot/pkg/api/http/service/monitoring_traffic.py index 702d34e01..c8469c488 100644 --- a/src/langbot/pkg/api/http/service/monitoring_traffic.py +++ b/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: diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 58cc9be71..104532f5d 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -165,7 +165,7 @@ class LangBotMCPServer: return _dump(ap.box_service.get_recent_errors(context)) @mcp.tool( - description='Read bounded Workspace runtime records. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, knowledge_base_id, user_query, is_active as supported by the record kind.' + description='Read bounded Workspace runtime records. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind.' ) async def get_monitoring_records( kind: typing.Literal['messages', 'llm_calls', 'tool_calls', 'embedding_calls', 'sessions', 'errors'], @@ -176,7 +176,7 @@ class LangBotMCPServer: import datetime context = _authorized(Permission.RESOURCE_VIEW) - allowed = {'start_time', 'end_time'} + allowed = {'start_time', 'end_time', 'mode', 'execution_statuses'} if kind != 'embedding_calls': allowed |= {'bot_ids', 'pipeline_ids'} if kind in {'messages', 'tool_calls'}: @@ -207,6 +207,52 @@ class LangBotMCPServer: return _dump(await ap.monitoring_service.get_message_details(context, identifier)) return _dump(await ap.monitoring_service.get_session_analysis(context, identifier)) + @mcp.tool( + description='List Workspace executions, including events that did not start a processor. The legacy pipeline_ids filter accepts Agent, Pipeline and event processor IDs. Returns bounded rows and summary metrics.' + ) + async def get_monitoring_executions( + limit: int = 50, + offset: int = 0, + bot_ids: list[str] | None = None, + pipeline_ids: list[str] | None = None, + statuses: list[str] | None = None, + mode: typing.Literal['all', 'real', 'debug'] = 'all', + ) -> str: + context = _authorized(Permission.RESOURCE_VIEW) + return _dump( + await ap.monitoring_service.get_executions( + context, + limit=limit, + offset=offset, + bot_ids=bot_ids, + pipeline_ids=pipeline_ids, + statuses=statuses, + mode=mode, + ) + ) + + @mcp.tool( + description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Each section is independently paginated; follow pages[section].next_offset when has_more is true.' + ) + async def get_monitoring_execution_detail( + source: typing.Literal['auto', 'agent', 'pipeline', 'event'], + identifier: str, + section: str | None = None, + offset: int = 0, + limit: int = 100, + ) -> str: + context = _authorized(Permission.RESOURCE_VIEW) + return _dump( + await ap.monitoring_service.get_execution_detail( + context, + source, + identifier, + section=section, + offset=offset, + limit=limit, + ) + ) + # ----- Bots ---------------------------------------------------- # @mcp.tool(description='List all messaging-platform bots. Secrets are redacted.') async def list_bots() -> str: diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index 1d2e8ed83..68d68d867 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -29,6 +29,9 @@ class MonitoringMessage(Base): runner_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True) # Runner name for this query variables = sqlalchemy.Column(sqlalchemy.Text, nullable=True) # Query variables as JSON string role = sqlalchemy.Column(sqlalchemy.String(50), nullable=True, default='user') # user, assistant + event_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) + run_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) + parent_message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) __table_args__ = ( sqlalchemy.Index('ix_monitoring_messages_workspace_timestamp', 'workspace_uuid', 'timestamp'), diff --git a/src/langbot/pkg/persistence/alembic/versions/0034_monitoring_execution_links.py b/src/langbot/pkg/persistence/alembic/versions/0034_monitoring_execution_links.py new file mode 100644 index 000000000..4cc720d07 --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0034_monitoring_execution_links.py @@ -0,0 +1,36 @@ +"""Link monitoring inputs and deliveries to events and runs.""" + +from alembic import op +import sqlalchemy as sa + +revision = '0034_monitoring_execution_links' +down_revision = '0033_operation_traceability' +branch_labels = None +depends_on = None + + +def upgrade(): + inspector = sa.inspect(op.get_bind()) + if not inspector.has_table('monitoring_messages'): + return + columns = {c['name'] for c in inspector.get_columns('monitoring_messages')} + indexes = {i['name'] for i in inspector.get_indexes('monitoring_messages')} + for name in ('event_id', 'run_id', 'parent_message_id'): + if name not in columns: + op.add_column('monitoring_messages', sa.Column(name, sa.String(255), nullable=True)) + index = f'ix_monitoring_messages_{name}' + if index not in indexes: + op.create_index(index, 'monitoring_messages', [name]) + + +def downgrade(): + inspector = sa.inspect(op.get_bind()) + if not inspector.has_table('monitoring_messages'): + return + columns = {c['name'] for c in inspector.get_columns('monitoring_messages')} + indexes = {i['name'] for i in inspector.get_indexes('monitoring_messages')} + for name in ('parent_message_id', 'run_id', 'event_id'): + if f'ix_monitoring_messages_{name}' in indexes: + op.drop_index(f'ix_monitoring_messages_{name}', table_name='monitoring_messages') + if name in columns: + op.drop_column('monitoring_messages', name) diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index 0eacc85c5..6eaf860c3 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/src/langbot/pkg/persistence/tenant_uow.py @@ -483,9 +483,16 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s and getattr(getattr(source, 'table', None), 'name', None) == 'legacy_pipelines' and source.name in {'config', 'stages', 'extensions_preferences'} ) - if not pipeline_json_cas: + monitoring_metadata = ( + type(element.type) is sqlalchemy.JSON + and isinstance(source, sqlalchemy.Column) + and type(source.type) is sqlalchemy.Text + and getattr(getattr(source, 'table', None), 'name', None) in {'event_log', 'agent_run'} + and source.name == 'metadata_json' + ) + if not (pipeline_json_cas or monitoring_metadata): raise ScopedSessionTransactionError( - 'TenantUnitOfWork only allows trusted pgvector and Pipeline JSON CAS casts' + 'TenantUnitOfWork only allows trusted pgvector, Pipeline JSON CAS and monitoring metadata casts' ) if isinstance(element, sqlalchemy.sql.functions.FunctionElement): diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index c4c7fc8f1..3fae71d9c 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -296,6 +296,7 @@ class MonitoringHelper: user_name=sender_name, runner_name=runner_name, role='assistant', + parent_message_id=(query.variables or {}).get('_monitoring_message_id'), ) except Exception as e: ap.logger.error(f'Failed to record query response: {e}') diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 7b942c54d..e3e0efad6 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -931,6 +931,32 @@ class RuntimeBot: node=node, ) + async def _persist_monitoring_ingress(self, event, adapter, execution_id, status='running', routes=None): + service = getattr(getattr(self, 'ap', None), 'monitoring_service', None) + if service is None or not execution_id: + return + try: + sender = getattr(event, 'sender', None) + await service.record_ingress_event( + self.execution_context, + event_id=execution_id, + event_type=getattr(event, 'type', None) or event.__class__.__name__, + bot_id=self.bot_entity.uuid, + bot_name=self.bot_entity.name, + platform=getattr(self.bot_entity, 'adapter', None) or adapter.__class__.__name__, + payload={ + key: value + for key, value in self._safe_model_dump(event).items() + if key != 'source_platform_object' and not key.startswith('_') + }, + actor_id=self._get_entity_id(sender), + actor_name=self._get_entity_name(sender), + status=status, + routes=routes, + ) + except Exception as exc: + await self.logger.warning(f'Failed to persist monitoring ingress: {exc}') + async def _handle_platform_event_body( self, event: platform_events.EBAEvent, @@ -941,6 +967,7 @@ class RuntimeBot: results: list[typing.Any] = [] try: 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) @@ -961,6 +988,7 @@ class RuntimeBot: results = await asyncio.gather(*tasks, return_exceptions=True) except BaseException as exc: self._record_platform_event(event, adapter, 'failed', f'{type(exc).__name__}: {exc}', node) + await self._persist_monitoring_ingress(event, adapter, execution_id, 'failed', [exc]) raise # Recorded after handling so the inbound stage carries the real outcome @@ -973,6 +1001,13 @@ class RuntimeBot: except Exception: pass self._record_platform_event(event, adapter, outcome, error_detail, node) + await self._persist_monitoring_ingress( + event, + adapter, + execution_id, + {'success': 'completed', 'skipped': 'ignored'}.get(outcome, 'failed'), + results, + ) async def _dispatch_plugin_subscription(self, event, adapter, processor_uuid, execution_id=None): event_type = event.type diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index c8d12c340..5b2356802 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -1,5 +1,7 @@ from __future__ import annotations +from contextlib import aclosing + import asyncio import inspect import typing @@ -1582,22 +1584,25 @@ class RuntimeConnectionHandler(handler.Handler): remove_think = _resolve_remove_think(data, query) effective_funcs = funcs_obj if 'func_call' in (llm_model.model_entity.abilities or []) else [] - async for chunk in llm_model.provider.invoke_llm_stream( - query=query, - model=llm_model, - messages=messages_obj, - funcs=effective_funcs, - extra_args=effective_extra_args, - remove_think=remove_think, - execution_context=execution_context, - ): - if chunk is None: - continue - yield handler.ActionResponse.success( - data={ - 'chunk': chunk.model_dump(), - }, + async with aclosing( + llm_model.provider.invoke_llm_stream( + query=query, + model=llm_model, + messages=messages_obj, + funcs=effective_funcs, + extra_args=effective_extra_args, + remove_think=remove_think, + execution_context=execution_context, ) + ) as owned_stream: + async for chunk in owned_stream: + if chunk is None: + continue + yield handler.ActionResponse.success( + data={ + 'chunk': chunk.model_dump(), + }, + ) usage = _pop_query_llm_usage(query) if usage is not None: yield handler.ActionResponse.success( diff --git a/src/langbot/pkg/provider/modelmgr/requester.py b/src/langbot/pkg/provider/modelmgr/requester.py index 069c9981e..99f6e885d 100644 --- a/src/langbot/pkg/provider/modelmgr/requester.py +++ b/src/langbot/pkg/provider/modelmgr/requester.py @@ -1,6 +1,9 @@ from __future__ import annotations +from contextlib import aclosing + import abc +import asyncio import typing import time @@ -204,15 +207,18 @@ class RuntimeProvider: try: # Stream the response - async for chunk in self.requester.invoke_llm_stream( - query=query, - model=model, - messages=messages, - funcs=funcs, - extra_args=extra_args, - remove_think=remove_think, - ): - yield chunk + async with aclosing( + self.requester.invoke_llm_stream( + query=query, + model=model, + messages=messages, + funcs=funcs, + extra_args=extra_args, + remove_think=remove_think, + ) + ) as owned_stream: + async for chunk in owned_stream: + yield chunk # Extract usage from stream if available (stored by LiteLLM requester) if query: if query.variables is None: @@ -223,6 +229,10 @@ class RuntimeProvider: input_tokens = usage_info.get('prompt_tokens', 0) output_tokens = usage_info.get('completion_tokens', 0) del query.variables[STREAM_USAGE_QUERY_VARIABLE] + except (asyncio.CancelledError, GeneratorExit): + status = 'error' + error_message = 'Streaming generation cancelled by consumer' + raise except Exception as e: status = 'error' error_message = str(e) diff --git a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py index 0bbaa1db5..23a69bea8 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py +++ b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py @@ -3,6 +3,7 @@ from __future__ import annotations import typing +import asyncio import litellm from litellm import acompletion, aembedding, arerank @@ -1245,6 +1246,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester): reasoning_started = False reasoning_closed = False thinking_blocks_state: list[dict[str, typing.Any]] = [] + response = None try: response = await acompletion(**args) @@ -1372,6 +1374,16 @@ class LiteLLMRequester(requester.ProviderAPIRequester): except Exception as e: self._handle_litellm_error(e) + finally: + # Closing a downstream async generator must release the provider's + # HTTP stream immediately, including cancellation while at yield. + close = getattr(response, 'aclose', None) + if callable(close): + try: + await asyncio.wait_for(close(), timeout=5.0) + except Exception as exc: + self.ap.logger.debug(f'Failed to close LLM stream: {type(exc).__name__}') + async def invoke_embedding( self, model: requester.RuntimeEmbeddingModel, diff --git a/tests/unit_tests/agent/test_orchestrator_integration.py b/tests/unit_tests/agent/test_orchestrator_integration.py index 1404a1537..bd8cf61a1 100644 --- a/tests/unit_tests/agent/test_orchestrator_integration.py +++ b/tests/unit_tests/agent/test_orchestrator_integration.py @@ -80,6 +80,7 @@ class FakePluginConnector: self.results = results or [] self.error = error self.delay = delay + self.delay_entered = asyncio.Event() self.calls: list[dict] = [] self.contexts: list[dict] = [] self.sessions_during_run: list[dict | None] = [] @@ -100,6 +101,7 @@ class FakePluginConnector: for result in self.results: if self.delay: + self.delay_entered.set() await asyncio.sleep(self.delay) yield result @@ -600,6 +602,35 @@ async def test_orchestrator_stops_after_cancel_request(clean_agent_state): assert run['status_reason'] == 'user stopped' +@pytest.mark.asyncio +@pytest.mark.parametrize('cancel_while_waiting', [False, True]) +async def test_consumer_exit_finalizes_cancelled_ledger(clean_agent_state, cancel_while_waiting): + connector = FakePluginConnector( + results=[ + {'type': 'message.delta', 'data': {'chunk': {'role': 'assistant', 'content': 'first'}}}, + {'type': 'run.completed', 'data': {'finish_reason': 'stop'}}, + ] + ) + orchestrator = AgentRunOrchestrator(FakeApplication(connector, clean_agent_state), FakeRegistry(make_descriptor())) + plan = orchestrator.query_bridge.build_plan(make_query()) + stream = orchestrator.run(plan.event, plan.binding, adapter_context={'_execution_context': TEST_CONTEXT}) + await anext(stream) + if cancel_while_waiting: + connector.delay = 10 + pending = asyncio.create_task(anext(stream)) + # Cancel the provider wait, not an arbitrary journal DB operation. + await asyncio.wait_for(connector.delay_entered.wait(), timeout=2) + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + else: + await stream.aclose() + run = await RunLedgerStore(clean_agent_state).get_run(connector.contexts[0]['run_id']) + assert run['status'] == 'cancelled' + assert run['status_reason'] == 'consumer_cancelled' + assert await get_session_registry().list_active_runs() == [] + + @pytest.mark.asyncio async def test_orchestrator_does_not_package_query_messages_into_context(clean_agent_state): """Host should not build an agent working-context window from query.messages.""" diff --git a/tests/unit_tests/agent/test_run_journal.py b/tests/unit_tests/agent/test_run_journal.py index 8908681ba..1ae134a01 100644 --- a/tests/unit_tests/agent/test_run_journal.py +++ b/tests/unit_tests/agent/test_run_journal.py @@ -52,7 +52,19 @@ async def test_agent_run_snapshot_records_identity_and_redacts_inline_attachment assert raw_input['contents'][0]['image_base64'] == 'large-image-data' assert raw_input['attachments'][0]['content'] == 'large-file-data' - if event_type != 'message.received': - assert saved['metadata']['input_event'] == event.data - else: - assert 'input_event' not in saved['metadata'] + assert saved['metadata']['input_event'] == event.data + + +def test_event_snapshot_preserves_custom_values_and_redacts_nested_binary(): + data = { + 'member': {'id': 'u1', 'active': False}, + 'count': 0, + 'custom': {'contents': [{'type': 'image_base64', 'image_base64': 'secret'}]}, + 'attachments': [{'name': 'file.txt', 'content': 'secret'}], + } + saved = AgentRunJournal._sanitize_event_data(data) + assert saved['member'] == data['member'] + assert saved['count'] == 0 + assert saved['custom']['contents'][0]['image_base64'] is None + assert saved['attachments'][0]['content'] is None + assert data['custom']['contents'][0]['image_base64'] == 'secret' diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index cd8d767ec..ce3d79f34 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -23,6 +23,73 @@ from langbot.pkg.entity.persistence.base import Base WORKSPACE = 'workspace-1' +@pytest.mark.asyncio +@pytest.mark.parametrize('has_log', [False, True]) +async def test_event_properties_survive_detail_snapshot_fallback(engine, service, has_log): + import json + from langbot.pkg.entity.persistence.event_log import EventLog + + data = {'member': {'id': 'u1', 'active': False}, 'count': 0, 'custom': {'reason': 'joined'}} + async with AsyncSession(engine) as session: + session.add( + persistence_agent_run.AgentRun( + run_id='custom-run', + event_id='custom-event', + workspace_id=WORKSPACE, + runner_id='runner', + status='completed', + created_at=_dt(10), + metadata_json=json.dumps( + { + 'event_type': 'group.member_joined', + 'input_event': data, + 'input': {'text': '', 'contents': [], 'attachments': []}, + } + ), + ) + ) + if has_log: + session.add( + EventLog( + event_id='custom-event', + event_type='group.member_joined', + source='webui', + workspace_id=WORKSPACE, + run_id='custom-run', + input_json='{"text":"","contents":[]}', + created_at=_dt(10), + ) + ) + session.add( + EventLog( + event_id='later-event', + event_type='message.received', + source='webui', + workspace_id=WORKSPACE, + run_id='custom-run', + input_json='{"text":"later"}', + created_at=_dt(10, 1), + ) + ) + await session.commit() + detail = await service.get_execution_detail(WORKSPACE, 'agent', 'custom-run') + items = detail['pages']['inputs']['items'] + assert items[0]['content']['member'] == data['member'] + assert items[0]['content']['count'] == 0 + assert items[0]['content']['custom'] == data['custom'] + if has_log: + assert items[1]['content'] == {'text': 'later'} + + +def test_event_content_preserves_conflicting_custom_input_fields(): + from langbot.pkg.api.http.service.monitoring_execution_details import event_content + + assert event_content({'text': 'normalized'}, {'text': {'custom': False}}) == { + 'event': {'text': {'custom': False}}, + 'input': {'text': 'normalized'}, + } + + class FakePersistenceManager: """Runs statements against a real in-memory SQLite engine.""" @@ -33,6 +100,11 @@ class FakePersistenceManager: def get_db_engine(self): return self._engine + def tenant_uow(self, workspace_uuid): + from langbot.pkg.persistence.tenant_uow import TenantUnitOfWork + + return TenantUnitOfWork(self._engine, workspace_uuid) + async def execute_async(self, statement): # Match PersistenceManager's connection-level result shape. async with self._engine.connect() as connection: @@ -348,3 +420,319 @@ async def test_execution_detail_for_agent_and_pipeline(engine, service): with pytest.raises(ValueError): await service.get_execution_detail(WORKSPACE, 'pipeline', 'missing-message') + + +@pytest.mark.asyncio +async def test_explicit_deliveries_context_and_tenant_isolation(engine, service): + import sqlalchemy as sa + + await _seed(engine) + async with AsyncSession(engine) as session: + await session.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-assistant') + .values(parent_message_id='msg-user') + ) + await session.commit() + detail = await service.get_execution_detail(WORKSPACE, 'auto', 'msg-assistant') + assert detail['row']['id'] == 'msg-user' + assert [m['id'] for m in detail['pages']['deliveries']['items']] == ['msg-assistant'] + assert detail['pages']['inputs']['items'][0]['id'] == 'msg-user' + assert detail['pages']['conversation']['items'] + with pytest.raises(ValueError, match='not found'): + await service.get_execution_detail('another-workspace', 'auto', 'msg-assistant') + + +@pytest.mark.asyncio +async def test_legacy_replies_are_context_not_guessed_deliveries(engine, service): + await _seed(engine) + detail = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user') + assert detail['legacy_context'] is True + assert detail['pages']['deliveries']['items'] == [] + assert 'msg-assistant' in [m['id'] for m in detail['pages']['conversation']['items']] + + +@pytest.mark.asyncio +async def test_fanout_steering_orphans_and_paged_trace(engine, service): + import json + import sqlalchemy as sa + from langbot.pkg.entity.persistence.event_log import EventLog + from langbot.pkg.entity.persistence.transcript import Transcript + + await _seed(engine) + async with AsyncSession(engine) as session: + await session.execute( + sa.update(persistence_agent_run.AgentRun) + .where(persistence_agent_run.AgentRun.run_id == 'run-done') + .values( + conversation_id='shared', + thread_id='thread-1', + metadata_json=json.dumps({'processor_type': 'event_processor'}), + ) + ) + session.add_all( + [ + EventLog( + event_id='evt-1', + event_type='message.received', + source='platform', + workspace_id=WORKSPACE, + bot_id='bot-1', + input_json='{"text":"original input"}', + actor_name='Alice', + created_at=_dt(10), + ), + EventLog( + event_id='steer', + event_type='message.received', + source='platform', + workspace_id=WORKSPACE, + bot_id='bot-1', + run_id='run-done', + input_json='{"text":"additional input"}', + created_at=_dt(10, 1), + ), + EventLog( + event_id='unrouted', + event_type='group.member.joined', + source='platform', + workspace_id=WORKSPACE, + bot_id='bot-1', + metadata_json='{"status":"ignored"}', + created_at=_dt(10, 7), + ), + persistence_agent_run.AgentRun( + run_id='sibling', + event_id='evt-1', + workspace_id=WORKSPACE, + bot_id='bot-1', + runner_id='plugin', + status='completed', + created_at=_dt(10), + ), + Transcript( + transcript_id='output', + event_id='generated', + conversation_id='shared', + thread_id='thread-1', + workspace_id=WORKSPACE, + bot_id='bot-1', + role='assistant', + content='generated only', + run_id='run-done', + seq=1, + ), + Transcript( + transcript_id='other-thread', + event_id='other', + conversation_id='shared', + thread_id='thread-2', + workspace_id=WORKSPACE, + bot_id='bot-1', + role='user', + content='private other thread', + seq=2, + ), + ] + ) + session.add_all( + [ + persistence_agent_run.AgentRunEvent(run_id='run-done', sequence=i, type='message.delta', data_json='{}') + for i in range(2, 206) + ] + ) + await session.commit() + result = await service.get_executions(WORKSPACE) + assert 'unrouted' in [r['id'] for r in result['items']] + assert 'steer' not in [r['id'] for r in result['items']] + ignored = await service.get_executions(WORKSPACE, statuses=['ignored']) + assert 'unrouted' in [r['id'] for r in ignored['items']] + detail = await service.get_execution_detail(WORKSPACE, 'agent', 'run-done') + assert detail['row']['target_kind'] == 'event_processor' + assert detail['row']['user_name'] == 'Alice' + assert [i['id'] for i in detail['pages']['inputs']['items']] == ['evt-1', 'steer'] + assert detail['pages']['outputs']['items'][0]['content'] == 'generated only' + assert [i['id'] for i in detail['pages']['related']['items']] == ['sibling'] + assert 'other-thread' not in [i['id'] for i in detail['pages']['conversation']['items']] + assert len(detail['pages']['events']['items']) == 100 + assert detail['pages']['events']['has_more'] + final = await service.get_execution_detail(WORKSPACE, 'agent', 'run-done', section='events', offset=200) + assert len(final['pages']['events']['items']) == 5 + assert not final['pages']['events']['has_more'] + orphan = await service.get_execution_detail(WORKSPACE, 'event', 'unrouted') + assert orphan['pages']['deliveries']['items'] == [] + assert orphan['pages']['conversation']['items'] == [] + + +@pytest.mark.asyncio +async def test_pipeline_runner_is_not_counted_twice(engine, service): + import sqlalchemy as sa + + await _seed(engine) + async with AsyncSession(engine) as session: + await session.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-user') + .values(run_id='run-done') + ) + await session.commit() + result = await service.get_executions(WORKSPACE) + assert result['total'] == 4 + assert result['summary']['executions']['total'] == 4 + assert 'run-done' not in [r['id'] for r in result['items']] + detail = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user') + assert detail['row']['duration_ms'] == 2000 + assert detail['events'][0]['type'] == 'run.completed' + + +@pytest.mark.asyncio +async def test_event_log_fanout_is_idempotent_and_cannot_cross_tenants(engine): + from langbot.pkg.agent.runner.event_log_store import EventLogStore + + store = EventLogStore(engine) + await store.append_event('event-id', 'message.received', 'platform', workspace_id=WORKSPACE, bot_id='bot') + await store.append_event( + 'event-id', + 'message.received', + 'platform', + workspace_id=WORKSPACE, + bot_id='bot', + run_id='first-run', + input_json={'text': 'input', 'attachments': []}, + ) + await store.append_event( + 'event-id', 'message.received', 'platform', workspace_id=WORKSPACE, bot_id='bot', run_id='second-run' + ) + event = await store.get_event('event-id') + assert event['run_id'] == 'first-run' + assert event['input_json']['text'] == 'input' + with pytest.raises(Exception): + await store.append_event('event-id', 'message.received', 'platform', workspace_id='other', bot_id='bot') + + +@pytest.mark.asyncio +async def test_processor_filter_scopes_each_kind_and_legacy_calls(engine, service): + import json + + await _seed(engine) + async with AsyncSession(engine) as session: + session.add( + persistence_agent_run.AgentRun( + run_id='event-run', + workspace_id=WORKSPACE, + runner_id='event-runner', + status='completed', + created_at=_dt(11), + metadata_json=json.dumps({'processor_type': 'event_processor', 'processor_id': 'event-processor-1'}), + ) + ) + session.add( + persistence_monitoring.MonitoringLLMCall( + id='event-call', + bot_id='bot-1', + bot_name='Bot One', + pipeline_name='Event Processor', + session_id='session-event', + input_tokens=6, + output_tokens=6, + duration=1, + timestamp=_dt(11), + workspace_uuid=WORKSPACE, + model_name='model', + message_id='event-run', + pipeline_id='', + status='success', + total_tokens=12, + ) + ) + await session.commit() + for processor, expected in [ + ('agent-1', {'run-done', 'run-debug'}), + ('pipeline-1', {'msg-user', 'msg-discarded'}), + ('event-processor-1', {'event-run'}), + ('missing', set()), + ]: + result = await service.get_executions(WORKSPACE, pipeline_ids=[processor]) + assert {row['id'] for row in result['items']} == expected + assert result['total'] == result['summary']['executions']['total'] == len(expected) + calls, total = await service.get_llm_calls(WORKSPACE, pipeline_ids=['event-processor-1']) + assert total == 1 + assert [row['id'] for row in calls] == ['event-call'] + other = await service.get_executions('another-workspace', pipeline_ids=['event-processor-1']) + assert other['total'] == 0 + + +@pytest.mark.asyncio +async def test_execution_filters_also_scope_cards_calls_and_traffic(engine, service): + import sqlalchemy as sa + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + await _seed(engine) + async with AsyncSession(engine) as session: + await session.execute( + sa.update(persistence_monitoring.MonitoringLLMCall) + .where(persistence_monitoring.MonitoringLLMCall.id == 'llm-1') + .values(message_id='run-failed', pipeline_id='') + ) + await session.commit() + failed = await service.get_executions(WORKSPACE, statuses=['failed']) + assert failed['total'] == failed['summary']['executions']['total'] == 1 + assert failed['summary']['executions']['waiting'] == 0 + calls, total = await service.get_llm_calls(WORKSPACE, execution_statuses=['failed']) + assert total == 1 and calls[0]['id'] == 'llm-1' + # A successful model call may belong to a failed execution. + assert calls[0]['status'] == 'success' + overview = await service.get_overview_metrics(WORKSPACE, execution_statuses=['failed']) + assert overview['llm_calls'] == 1 and overview['total_messages'] == 0 + traffic = await get_traffic_series(service.ap, WORKSPACE, execution_statuses=['failed']) + assert sum(p['llm_calls'] for p in traffic['points']) == 1 + assert sum(p['messages'] for p in traffic['points']) == 0 + + debug = await service.get_executions(WORKSPACE, mode='debug') + assert debug['total'] == debug['summary']['executions']['total'] == 1 + real = await service.get_executions(WORKSPACE, mode='real') + assert real['total'] == real['summary']['executions']['total'] == 4 + traffic = await get_traffic_series(service.ap, WORKSPACE, mode='debug') + assert sum(p['messages'] for p in traffic['points']) == 0 + + +@pytest.mark.asyncio +async def test_pipeline_debug_is_consistent_across_rows_rollups_and_telemetry(engine, service): + import sqlalchemy as sa + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + await _seed(engine) + async with AsyncSession(engine) as session: + await session.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-user') + .values(bot_id='websocket-proxy-bot') + ) + await session.execute( + sa.update(persistence_monitoring.MonitoringLLMCall) + .where(persistence_monitoring.MonitoringLLMCall.id == 'llm-1') + .values(message_id='msg-user', pipeline_id='pipeline-1') + ) + await session.execute( + sa.update(persistence_agent_run.AgentRun) + .where(persistence_agent_run.AgentRun.run_id == 'run-done') + .values(bot_id='websocket-proxy-bot') + ) + await session.commit() + debug = await service.get_executions(WORKSPACE, mode='debug') + ids = {r['id'] for r in debug['items']} + assert ids == {'msg-user', 'run-done', 'run-debug'} + assert all(r['debug'] for r in debug['items']) + assert debug['summary']['executions']['total'] == 3 + assert debug['summary']['executions']['debug'] == 3 + real = await service.get_executions(WORKSPACE, mode='real') + assert not ids.intersection(r['id'] for r in real['items']) + assert real['summary']['executions']['debug'] == 0 + filtered = await service.get_executions(WORKSPACE, pipeline_ids=['pipeline-1'], mode='debug') + assert [r['id'] for r in filtered['items']] == ['msg-user'] + calls, total = await service.get_llm_calls(WORKSPACE, pipeline_ids=['pipeline-1'], mode='debug') + assert total == 1 and calls[0]['id'] == 'llm-1' + real_rows, real_calls = await service.get_llm_calls(WORKSPACE, pipeline_ids=['pipeline-1'], mode='real') + assert real_calls == 1 and real_rows[0]['id'] == 'llm-2' + traffic = await get_traffic_series(service.ap, WORKSPACE, mode='debug', pipeline_ids=['pipeline-1']) + assert sum(p['llm_calls'] for p in traffic['points']) == 1 diff --git a/tests/unit_tests/persistence/test_tenant_uow.py b/tests/unit_tests/persistence/test_tenant_uow.py index 0c374f201..7b624d2ee 100644 --- a/tests/unit_tests/persistence/test_tenant_uow.py +++ b/tests/unit_tests/persistence/test_tenant_uow.py @@ -980,6 +980,18 @@ async def test_scoped_sql_structure_allows_only_the_production_vocabulary(statem _validate_scoped_statement_call((statement,), {}) +async def test_monitoring_json_casts_are_limited_to_journal_metadata() -> None: + from langbot.pkg.entity.persistence.agent_run import AgentRun + from langbot.pkg.entity.persistence.event_log import EventLog + + for column in (AgentRun.metadata_json, EventLog.metadata_json): + statement = sa.select(sa.cast(column, sa.JSON)['status'].as_string()) + _validate_scoped_statement_call((statement,), {}) + for column in (AgentRun.authorization_json, EventLog.input_json, sa.literal('{}')): + with pytest.raises(ScopedSessionTransactionError, match='casts'): + _validate_scoped_statement_call((sa.select(sa.cast(column, sa.JSON)),), {}) + + async def test_scoped_session_executes_sqlite_strftime() -> None: engine = create_async_engine('sqlite+aiosqlite:///:memory:') try: diff --git a/tests/unit_tests/provider/test_litellmchat.py b/tests/unit_tests/provider/test_litellmchat.py index 490f7aa12..4c089816d 100644 --- a/tests/unit_tests/provider/test_litellmchat.py +++ b/tests/unit_tests/provider/test_litellmchat.py @@ -53,6 +53,37 @@ class MockRuntimeRerankModel: self.provider.token_mgr.get_token = Mock(return_value=api_key) +@pytest.mark.asyncio +async def test_closing_consumer_closes_provider_http_stream(): + import asyncio + from types import SimpleNamespace + + app = Mock() + app.tool_mgr.generate_tools_for_openai = AsyncMock(return_value=None) + requester = litellmchat.LiteLLMRequester(ap=app, config={}) + closed = asyncio.Event() + + async def provider(): + try: + yield SimpleNamespace( + choices=[ + SimpleNamespace( + delta=SimpleNamespace(model_dump=lambda: {'role': 'assistant', 'content': 'first'}), + finish_reason=None, + ) + ] + ) + await asyncio.Event().wait() + finally: + closed.set() + + with patch.object(litellmchat, 'acompletion', new=AsyncMock(side_effect=lambda **kw: provider())): + output = requester.invoke_llm_stream(query=None, model=MockRuntimeModel(), messages=[]) + assert (await anext(output)).content == 'first' + await output.aclose() + assert closed.is_set() + + class TestBuildLiteLLMModelName: """Test _build_litellm_model_name method""" diff --git a/web/src/app/home/monitoring/components/TokenMonitoring.tsx b/web/src/app/home/monitoring/components/TokenMonitoring.tsx index 49ed0e2c0..4871d44f9 100644 --- a/web/src/app/home/monitoring/components/TokenMonitoring.tsx +++ b/web/src/app/home/monitoring/components/TokenMonitoring.tsx @@ -77,6 +77,8 @@ interface TokenStatistics { } interface TokenMonitoringProps { + mode?: 'all' | 'real' | 'debug'; + statusGroup?: string; botIds?: string[]; pipelineIds?: string[]; startTime?: string; @@ -103,6 +105,8 @@ const TOOLTIP_STYLE: React.CSSProperties = { }; export default function TokenMonitoring({ + mode, + statusGroup, botIds, pipelineIds, startTime, @@ -123,6 +127,9 @@ export default function TokenMonitoring({ setError(null); try { const result = await httpClient.getTokenStatistics({ + mode, + status: + statusGroup && statusGroup !== 'all' ? [statusGroup] : undefined, botId: botIds, pipelineId: pipelineIds, startTime, @@ -136,7 +143,16 @@ export default function TokenMonitoring({ setLoading(false); } // eslint-disable-next-line react-hooks/exhaustive-deps - }, [botIdsKey, pipelineIdsKey, startTime, endTime, bucket, refreshKey]); + }, [ + mode, + statusGroup, + botIdsKey, + pipelineIdsKey, + startTime, + endTime, + bucket, + refreshKey, + ]); useEffect(() => { fetchStats(); diff --git a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx index f0ad3c947..b31110b93 100644 --- a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx +++ b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx @@ -1,7 +1,7 @@ import { useEffect, useRef, useState } from 'react'; import { useTranslation } from 'react-i18next'; import { Link } from 'react-router-dom'; -import { ExternalLink } from 'lucide-react'; +import { RefreshCw } from 'lucide-react'; import { Sheet, SheetContent, @@ -13,14 +13,14 @@ import { Badge } from '@/components/ui/badge'; import { Button } from '@/components/ui/button'; import { Skeleton } from '@/components/ui/skeleton'; import { ScrollArea } from '@/components/ui/scroll-area'; -import { Separator } from '@/components/ui/separator'; -import { backendClient } from '@/app/infra/http'; +import { backendClient, useCurrentWorkspace } from '@/app/infra/http'; import type { - ExecutionCallRow, ExecutionDetail, + ExecutionDetailItem, ExecutionRow, + ExecutionSection, } from '@/app/infra/entities/api/monitoring-executions'; -import AgentExecutionTrace from '@/app/home/agents/components/AgentExecutionTrace'; +import { ExecutionValue as Value } from './ExecutionValue'; import { formatRunDuration } from '@/app/home/agents/components/processor-run-timing'; import { formatDateTime } from '../../utils/dateUtils'; import { @@ -28,329 +28,409 @@ import { executionProcessorLabel, executionStatusLabel, executionStatusVariant, - executionTargetLabel, } from './ExecutionTable'; -interface ExecutionDetailSheetProps { - row: ExecutionRow | null; - onClose: () => void; -} +export type ExecutionSelection = Pick & { + source: ExecutionRow['source'] | 'auto'; +}; -/** Walk parsed JSON, collecting every `"text"` value (OpenAI-style content). */ -function collectText(value: unknown): string { - if (typeof value === 'string') return value; - if (Array.isArray(value)) { - return value.map(collectText).filter(Boolean).join(' '); - } - if (value && typeof value === 'object') { - const record = value as Record; - if (typeof record.text === 'string') return record.text; - return Object.values(record).map(collectText).filter(Boolean).join(' '); - } - return ''; -} - -function messagePreview(raw: string | null | undefined): string { - if (!raw) return ''; - try { - return collectText(JSON.parse(raw)).trim() || raw; - } catch { - return raw; - } -} - -function callTitle(call: ExecutionCallRow, index: number): string { - const candidates = [ - call.model_name, - call.model, - call.name, - call.tool_name, - call.function, - call.error_type, - ]; - for (const candidate of candidates) { - if (typeof candidate === 'string' && candidate) return candidate; - } - return typeof call.id === 'string' && call.id ? call.id : `#${index + 1}`; -} +const mainSections: ExecutionSection[] = ['inputs', 'outputs', 'deliveries']; +const secondarySections: ExecutionSection[] = [ + 'related', + 'conversation', + 'llm_calls', + 'tool_calls', + 'errors', + 'events', +]; export default function ExecutionDetailSheet({ row, onClose, -}: ExecutionDetailSheetProps) { + onSelect, +}: { + row: ExecutionSelection | null; + onClose: () => void; + onSelect: (row: ExecutionSelection) => void; +}) { const { t } = useTranslation(); + const workspace = useCurrentWorkspace()?.workspace.uuid; const [detail, setDetail] = useState(null); const [loading, setLoading] = useState(false); const [error, setError] = useState(false); - const requestRef = useRef(0); - + const [sectionError, setSectionError] = useState( + null, + ); + const [busy, setBusy] = useState(null); + const [refresh, setRefresh] = useState(0); + const request = useRef(0); + const identity = `${workspace}:${row?.source}:${row?.id}`; + const [loadedIdentity, setLoadedIdentity] = useState(''); + const visible = loadedIdentity === identity ? detail : null; + const id = row?.id; + const source = row?.source; useEffect(() => { - if (!row) { - setDetail(null); - setError(false); + const version = ++request.current; + setDetail(null); + setError(false); + setSectionError(null); + setBusy(null); + if (!id || !source) { setLoading(false); return; } - const requestId = requestRef.current + 1; - requestRef.current = requestId; setLoading(true); - setError(false); - setDetail(null); backendClient - .getExecutionDetail(row.source, row.id) + .getExecutionDetail(source, id) .then((result) => { - if (requestRef.current !== requestId) return; + if (version !== request.current) return; setDetail(result); + setLoadedIdentity(identity); }) .catch(() => { - if (requestRef.current !== requestId) return; - setError(true); + if (version === request.current) setError(true); }) .finally(() => { - if (requestRef.current === requestId) setLoading(false); - }); - }, [row]); - - const isAgent = - detail != null ? detail.source === 'agent' : row?.source === 'agent'; - - const metadata: { label: string; value: string }[] = []; - if (row) { - metadata.push({ - label: t('monitoring.execution.detail.executionId'), - value: row.id, - }); - if (row.event_id) { - metadata.push({ - label: t('monitoring.execution.detail.eventId'), - value: row.event_id, + if (version === request.current) setLoading(false); }); + return () => { + request.current += 1; + }; + }, [id, source, workspace, identity, refresh]); + const status = visible?.row.status_group; + useEffect(() => { + if (!['running', 'queued'].includes(status ?? '')) return; + // Do not discard paged history with a background refresh. + if ( + Object.values(visible?.pages ?? {}).some((page) => page.next_offset > 100) + ) + return; + const timer = setInterval(() => setRefresh((value) => value + 1), 5000); + return () => clearInterval(timer); + }, [status, visible]); + const loadMore = async (section: ExecutionSection) => { + if (!visible || !row || busy) return; + const version = request.current; + setBusy(section); + setSectionError(null); + try { + const result = await backendClient.getExecutionDetail( + row.source, + row.id, + { section, offset: visible.pages[section]?.next_offset ?? 0 }, + ); + if (version !== request.current) return; + const next = result.pages[section]; + if (next) + setDetail( + (previous) => + previous && { + ...previous, + pages: { + ...previous.pages, + [section]: { + ...next, + items: [ + ...(previous.pages[section]?.items ?? []), + ...next.items, + ], + }, + }, + }, + ); + } catch { + if (version === request.current) setSectionError(section); + } finally { + if (version === request.current) setBusy(null); } - if (row.runner_id) { - metadata.push({ - label: t('monitoring.execution.detail.runner'), - value: row.runner_id, - }); + }; + const renderItem = (section: ExecutionSection, item: ExecutionDetailItem) => { + if (section === 'related') { + const related = item as unknown as ExecutionRow; + return ( + + ); } - if (row.queue_name) { - metadata.push({ - label: t('monitoring.execution.detail.queue'), - value: row.queue_name, - }); - } - if (row.created_at_ms != null) { - metadata.push({ - label: t('monitoring.execution.detail.created'), - value: formatDateTime(new Date(row.created_at_ms)), - }); - } - if (row.duration_ms != null) { - metadata.push({ - label: t('monitoring.execution.detail.duration'), - value: formatRunDuration(row.duration_ms), - }); - } - if (isAgent && row.usage?.total_tokens != null) { - metadata.push({ - label: t('monitoring.execution.detail.usage'), - value: row.usage.total_tokens.toLocaleString(), - }); - } - } - - const objectHref = row - ? row.target_kind === 'agent' && row.target_id - ? `/home/agents?id=${encodeURIComponent(row.target_id)}` - : row.target_kind === 'pipeline' && row.target_id - ? `/home/pipelines?id=${encodeURIComponent(row.target_id)}` + const content = [ + 'inputs', + 'outputs', + 'deliveries', + 'conversation', + ].includes(section); + const title = content + ? [ + item.actor_name || item.actor_id, + item.event_type, + item.role && + t(`monitoring.unified.roles.${item.role}`, { + defaultValue: item.role, + }), + ] + .filter(Boolean) + .join(' · ') + : String( + item.model_name || + item.tool_name || + item.error_type || + item.type || + item.id, + ); + return ( +
+
+ {title} + + {item.timestamp_ms + ? formatDateTime(new Date(item.timestamp_ms)) + : typeof item.timestamp === 'string' + ? item.timestamp + : ''} + + {item.status && ( + + {item.status} + + )} +
+ {content ? ( + <> + + + {item.origin === 'generated' && section === 'conversation' && ( + {t('monitoring.unified.outputs')} + )} + {item.delivery?.delivery_error ? ( +

+ {String(item.delivery.delivery_error)} +

+ ) : null} + {Array.isArray(item.metadata?.routes) && ( + + )} + + ) : ( + <> +
+ {typeof item.duration === 'number' && ( + {formatRunDuration(item.duration)} + )} + {typeof item.total_tokens === 'number' && ( + {item.total_tokens.toLocaleString()} Token + )} +
+ +
+ + {t('monitoring.unified.payload')} + +
+ +
+
+ + )} +
+ ); + }; + const renderSection = (section: ExecutionSection, primary = false) => { + const page = visible?.pages[section]; + if ( + !page || + (!primary && page.items.length === 0 && section !== 'conversation') + ) + return null; + const body = ( +
+ {section === 'conversation' && ( +

+ {t('monitoring.unified.contextHint')} +

+ )} + {page.items.length === 0 && ( +

+ {t('monitoring.unified.noRecords')} +

+ )} + {page.items.map((item, i) => ( +
{renderItem(section, item)}
+ ))} + {sectionError === section && ( +

+ {t('monitoring.execution.detail.loadError')} +

+ )} + {page.has_more && ( + + )} +
+ ); + const title = `${t(`monitoring.unified.${section}`)} (${page.items.length}${page.has_more ? '+' : ''})`; + return primary ? ( +
+

{title}

+ {body} +
+ ) : ( +
+ + {title} + + {body} +
+ ); + }; + const record = visible?.row; + const href = record?.target_id + ? record.target_kind === 'pipeline' + ? `/home/pipelines?id=${encodeURIComponent(record.target_id)}` + : ['agent', 'event_processor', 'processor'].includes(record.target_kind) + ? `/home/agents?id=${encodeURIComponent(record.target_id)}` : null : null; - - const renderCalls = (title: string, calls: ExecutionCallRow[]) => ( -
-

{title}

- {calls.length === 0 ? ( -

- {t('monitoring.execution.detail.noEvents')} -

- ) : ( -
- {calls.map((call, index) => { - const duration = - typeof call.duration === 'number' ? call.duration : null; - return ( -
-
-
{callTitle(call, index)}
- {call.error_message && ( -
- {call.error_message} -
- )} -
-
- {call.status && ( - - {call.status} - - )} - - {duration == null ? '—' : formatRunDuration(duration)} - -
-
- ); - })} -
- )} -
- ); - return ( { if (!open) onClose(); }} > - - - {t('monitoring.execution.detail.a11yDescription')} - + + + {t('monitoring.execution.detail.title')} + + {t('monitoring.execution.detail.a11yDescription')} + + -
- {row && ( +
+
+

+ {record + ? record.source === 'event' + ? t('monitoring.unified.unhandled') + : executionProcessorLabel(record) + : t('monitoring.execution.detail.title')} +

+ +
+ {loading && } + {error && ( +

+ {t('monitoring.execution.detail.loadError')} +

+ )} + {record && ( <> - - - {executionProcessorLabel(row)} - -

- {t('monitoring.execution.triggeredBy', { - trigger: executionTargetLabel(row, t), - })} -

-
- - {executionKindLabel(row, t)} +
+ + {executionStatusLabel(record, t)} + + + {executionKindLabel(record, t)} + + {record.debug && ( + + {t('monitoring.execution.debug')} - - {executionStatusLabel(row, t)} - - {row.debug && ( - - {t('monitoring.execution.debug')} - - )} - {objectHref && ( - - )} -
- - -
- {metadata.map((item) => ( -
-
- {item.label} -
-
- {item.value} -
-
- ))} + )} + {href && ( + + )}
- - - - {isAgent ? ( -
-

- {t('monitoring.execution.detail.timeline')} -

- {loading && ( -
- - -
- )} - {!loading && error && ( -

- {t('monitoring.execution.detail.loadError')} -

- )} - {!loading && - !error && - detail?.source === 'agent' && - (detail.events.length > 0 ? ( - - ) : ( -

- {t('monitoring.execution.detail.noEvents')} -

- ))} -
- ) : ( -
- {loading && ( -
- - -
- )} - {!loading && error && ( -

- {t('monitoring.execution.detail.loadError')} -

- )} - {!loading && !error && detail?.source === 'pipeline' && ( - <> -
-

- {t('monitoring.execution.detail.query')} -

-

- {messagePreview(detail.message.message_content) || - '—'} -

-
- {renderCalls( - t('monitoring.execution.detail.modelCalls'), - detail.llm_calls, - )} - {renderCalls( - t('monitoring.execution.detail.toolCalls'), - detail.tool_calls, - )} - {renderCalls( - t('monitoring.execution.detail.errors'), - detail.errors, - )} - - )} -
+

+ {[ + record.bot_name, + record.platform, + record.user_name || record.user_id, + ] + .filter(Boolean) + .join(' · ')} +

+ {record.status_reason && ( +

+ {record.status_reason} +

)} + {mainSections.map((section) => renderSection(section, true))} + {secondarySections.map((section) => renderSection(section))} +
+ + {t('monitoring.unified.metadata')} + +
+ {[ + [t('monitoring.execution.detail.executionId'), record.id], + [ + t('monitoring.execution.detail.eventId'), + record.event_id, + ], + [ + t('monitoring.unified.conversation'), + record.conversation_id || record.session_id, + ], + [ + t('monitoring.execution.detail.runner'), + record.runner_id, + ], + [ + t('monitoring.execution.detail.created'), + record.created_at_ms + ? formatDateTime(new Date(record.created_at_ms)) + : null, + ], + [ + t('monitoring.execution.detail.duration'), + record.duration_ms != null + ? formatRunDuration(record.duration_ms) + : null, + ], + ] + .filter(([, value]) => value) + .map(([label, value]) => ( +
+
{label}
+
{value}
+
+ ))} +
+
)}
diff --git a/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx b/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx index 901adef5d..7a447475a 100644 --- a/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx +++ b/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx @@ -45,17 +45,26 @@ export function executionStatusVariant( return 'outline'; } -const KIND_GROUPS = ['agent', 'pipeline', 'processor']; +const KIND_GROUPS = [ + 'agent', + 'pipeline', + 'processor', + 'event_processor', + 'event', +]; export function executionKindLabel(row: ExecutionRow, t: TFunction): string { if (KIND_GROUPS.includes(row.target_kind)) { + if (row.target_kind === 'event_processor') + return t('monitoring.execution.kind.processor'); + if (row.target_kind === 'event') return t('monitoring.unified.unhandled'); return t(`monitoring.execution.kind.${row.target_kind}`); } return row.target_kind; } export function executionTargetLabel(row: ExecutionRow, t: TFunction): string { - if (row.target_kind === 'agent') { + if (row.source !== 'pipeline') { return eventPatternLabel(row.title, t); } const title = row.title ?? ''; @@ -85,6 +94,7 @@ export default function ExecutionTable({ {t('monitoring.execution.columns.time')} {t('monitoring.execution.columns.target')} + {t('monitoring.unified.inputs')} {t('monitoring.execution.columns.status')} {t('monitoring.execution.columns.duration')} @@ -127,7 +137,9 @@ export default function ExecutionTable({ {executionKindLabel(row, t)} - {executionProcessorLabel(row)} + {row.source === 'event' + ? t('monitoring.unified.unhandled') + : executionProcessorLabel(row)}
@@ -137,6 +149,16 @@ export default function ExecutionTable({
+ +
+ {row.input_preview || '—'} +
+
+ {[row.bot_name, row.user_name || row.user_id] + .filter(Boolean) + .join(' · ')} +
+
diff --git a/web/src/app/home/monitoring/components/executions/ExecutionValue.tsx b/web/src/app/home/monitoring/components/executions/ExecutionValue.tsx new file mode 100644 index 000000000..72913462d --- /dev/null +++ b/web/src/app/home/monitoring/components/executions/ExecutionValue.tsx @@ -0,0 +1,178 @@ +import { MessageContentRenderer } from '../MessageContentRenderer'; + +// Keep arbitrary event properties, including false and zero. Empty transport +// fields are not messages. This also applies to historical serialized payloads. +export function hasExecutionValue(value: unknown): boolean { + if (value == null) return false; + if (typeof value === 'string') return value.trim().length > 0; + if (Array.isArray(value)) return value.some(hasExecutionValue); + if (typeof value === 'object') + return Object.values(value).some(hasExecutionValue); + return true; +} + +function textProjection(value: unknown): string { + if (typeof value === 'string') return value; + if (Array.isArray(value)) return value.map(textProjection).join(''); + if (value && typeof value === 'object') { + const item = value as Record; + if (item.type === 'text' || item.type === 'Plain') + return typeof item.text === 'string' ? item.text : ''; + } + return ''; +} + +const messageTypes = new Set([ + 'Plain', + 'At', + 'AtAll', + 'Image', + 'File', + 'Voice', + 'Quote', +]); +const componentFields: Record = { + Plain: ['text'], + At: ['target', 'display'], + AtAll: [], + Image: ['url', 'base64', 'imageId'], + File: ['name', 'id', 'url', 'size'], + Voice: ['url', 'base64', 'voiceId', 'length'], + Quote: ['origin'], +}; + +export function ExecutionValue({ value }: { value: unknown }) { + if (!hasExecutionValue(value)) return null; + if (typeof value === 'string') { + // Do not reinterpret ordinary strings such as "false", "0", or quoted + // user text. Only historical structured JSON needs decoding. + if (/^\s*[\[{]/.test(value)) { + try { + return ; + } catch { + /* Text. */ + } + } + return ( + 
+        {value}
+
+ ); + } + if (Array.isArray(value)) + return ( +
+ {value.map((item, index) => ( + + ))} +
+ ); + if (typeof value === 'object') { + const record = value as Record; + if (typeof record.type === 'string' && messageTypes.has(record.type)) { + const fields = ['type', ...componentFields[record.type]]; + return ( + <> + + !fields.includes(key)), + )} + /> + + ); + } + if (record.type === 'text') { + const rest = Object.fromEntries( + Object.entries(record).filter( + ([key, item]) => + key !== 'type' && key !== 'text' && hasExecutionValue(item), + ), + ); + return ( + <> + + + + ); + } + if (['image', 'image_url', 'image_base64'].includes(String(record.type))) { + const image = record.image_url; + const url = + typeof image === 'string' + ? image + : image && typeof image === 'object' + ? (image as { url?: string }).url + : typeof record.image_base64 === 'string' + ? record.image_base64.startsWith('data:') + ? record.image_base64 + : `data:image/png;base64,${record.image_base64}` + : null; + if (url) + return ( + <> + + + !['type', 'image_url', 'image_base64'].includes(key), + ), + )} + /> + + ); + } + if (['file', 'file_url', 'file_base64'].includes(String(record.type))) { + const name = record.file_name ?? record.name; + return ( + <> + + + !['type', 'file_name', 'name', 'file_base64'].includes(key), + ), + )} + /> + + ); + } + // Only equal text projections are aliases. Never deduplicate distinct + // records, or hide different text and structured content from one event. + const bodyKeys = ['chain', 'message_chain', 'contents', 'content']; + const bodyKey = bodyKeys.find((key) => hasExecutionValue(record[key])); + const body = bodyKey ? record[bodyKey] : undefined; + const duplicateText = + bodyKey && + typeof record.text === 'string' && + record.text === textProjection(body); + const entries = Object.entries(record).filter( + ([key, item]) => + hasExecutionValue(item) && !(key === 'text' && duplicateText), + ); + return ( +
+ {entries.map(([key, item]) => ( +
+ {!['text', ...bodyKeys, 'attachments'].includes(key) && ( +
{key}
+ )} + +
+ ))} +
+ ); + } + return {String(value)}; +} diff --git a/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx b/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx index 2470451f6..dcfcb4d8d 100644 --- a/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx +++ b/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx @@ -8,12 +8,9 @@ import { SelectValue, } from '@/components/ui/select'; import { Label } from '@/components/ui/label'; -import { backendClient } from '@/app/infra/http'; +import { backendClient, useCurrentWorkspace } from '@/app/infra/http'; import { TimeRangeOption } from '../../types/monitoring'; -import type { - ExecutionModeFilter, - ExecutionSourceFilter, -} from '@/app/infra/entities/api/monitoring-executions'; +import type { ExecutionModeFilter } from '@/app/infra/entities/api/monitoring-executions'; interface MonitoringFiltersProps { selectedBots: string[]; @@ -22,8 +19,6 @@ interface MonitoringFiltersProps { onBotsChange: (bots: string[]) => void; onPipelinesChange: (pipelines: string[]) => void; onTimeRangeChange: (timeRange: TimeRangeOption) => void; - source?: ExecutionSourceFilter; - onSourceChange?: (value: ExecutionSourceFilter) => void; mode?: ExecutionModeFilter; onModeChange?: (value: ExecutionModeFilter) => void; statusGroup?: string; @@ -39,14 +34,25 @@ const STATUS_GROUP_OPTIONS = [ 'ignored', ]; +const STATUS_COLORS: Record = { + completed: 'text-emerald-700 dark:text-emerald-400', + failed: 'text-red-600 dark:text-red-400', + running: 'text-blue-600 dark:text-blue-400', + queued: 'text-amber-700 dark:text-amber-400', + cancelled: 'text-slate-600 dark:text-slate-400', + ignored: 'text-muted-foreground', +}; + interface Bot { uuid: string; name: string; + iconURL: string; } -interface Pipeline { +interface Processor { uuid: string; name: string; + emoji?: string; } export default function MonitoringFilters({ @@ -56,21 +62,22 @@ export default function MonitoringFilters({ onBotsChange, onPipelinesChange, onTimeRangeChange, - source, - onSourceChange, mode, onModeChange, statusGroup, onStatusGroupChange, }: MonitoringFiltersProps) { const { t } = useTranslation(); + const workspace = useCurrentWorkspace()?.workspace.uuid; const [bots, setBots] = useState([]); - const [pipelines, setPipelines] = useState([]); + const [processors, setProcessors] = useState([]); const [loadingBots, setLoadingBots] = useState(false); - const [loadingPipelines, setLoadingPipelines] = useState(false); + const [loadingProcessors, setLoadingProcessors] = useState(false); // Fetch bots list useEffect(() => { + let active = true; + setBots([]); const fetchBots = async () => { setLoadingBots(true); try { @@ -78,41 +85,57 @@ export default function MonitoringFilters({ // Filter out bots without uuid and map to local Bot interface const validBots = (response.bots || []) .filter((bot): bot is typeof bot & { uuid: string } => !!bot.uuid) - .map((bot) => ({ uuid: bot.uuid, name: bot.name })); - setBots(validBots); + .map((bot) => ({ + uuid: bot.uuid, + name: bot.name, + iconURL: backendClient.getAdapterIconURL(bot.adapter), + })); + if (active) setBots(validBots); } catch (error) { console.error('Failed to fetch bots:', error); } finally { - setLoadingBots(false); + if (active) setLoadingBots(false); } }; fetchBots(); - }, []); + return () => { + active = false; + }; + }, [workspace]); - // Fetch pipelines list + // Fetch all processor kinds from the same catalog as the sidebar useEffect(() => { - const fetchPipelines = async () => { - setLoadingPipelines(true); + let active = true; + setProcessors([]); + const fetchProcessors = async () => { + setLoadingProcessors(true); try { - const response = await backendClient.getPipelines(); - // Filter out pipelines without uuid and map to local Pipeline interface - const validPipelines = (response.pipelines || []) + const response = await backendClient.getAgents(); + // Preserve the custom icon displayed to the left of the sidebar name. + const validProcessors = (response.agents || []) .filter( (pipeline): pipeline is typeof pipeline & { uuid: string } => !!pipeline.uuid, ) - .map((pipeline) => ({ uuid: pipeline.uuid, name: pipeline.name })); - setPipelines(validPipelines); + .map((pipeline) => ({ + uuid: pipeline.uuid, + name: pipeline.name, + emoji: pipeline.emoji, + })); + if (active) setProcessors(validProcessors); } catch (error) { - console.error('Failed to fetch pipelines:', error); + console.error('Failed to fetch processors:', error); } finally { - setLoadingPipelines(false); + if (active) setLoadingProcessors(false); } }; - fetchPipelines(); - }, []); + fetchProcessors(); + return () => { + active = false; + }; + }, [workspace]); const handleBotChange = (value: string) => { if (value === 'all') { @@ -135,41 +158,104 @@ export default function MonitoringFilters({ }; return ( -
- {/* Source Filter */} - {onSourceChange && ( -
- - + - - - - - - {t('monitoring.execution.filters.sourceAll')} + + + + + {t('monitoring.filters.allBots')} + + {bots.map((bot) => ( + + + + {bot.name} + - - {t('monitoring.execution.filters.sourceAgent')} + ))} + + +
+ + {/* Processor Filter */} +
+ + -
- )} + ))} + + +
{/* Mode Filter */} {onModeChange && ( -
-