diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index b04ec8394..9d3db00d2 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -194,6 +194,21 @@ class AgentRunOrchestrator: state_context = build_state_context(event, binding, descriptor) run_id = context['run_id'] + + # Monitoring attributes every call to the processor that ran it and to the + # record that owns it; a Runner has to claim both itself, since no + # Pipeline did. Runs before the Runner is invoked and before the run + # context is handed to it, so its own variables stay private. + from ...pipeline import monitoring_helper + + await monitoring_helper.prepare_monitoring_identity( + self.ap, + execution_query, + processor_type=binding.processor_type, + processor_id=binding.processor_id, + runner_id=binding.runner_id, + execution_record_id=run_id, + ) context['context']['available_apis']['reply_stream'] = hasattr(PluginToRuntimeAction, 'REPLY_STREAM') and any( tool.get('tool_name') == 'event_reply' and tool.get('tool_type') == 'platform' for tool in resources.get('tools', []) diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 01a901743..1d9656761 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/src/langbot/pkg/api/http/controller/groups/monitoring.py @@ -50,6 +50,47 @@ class MonitoringRouterGroup(group.RouterGroup): return self.success(data=metrics) + @self.route('/executions', methods=['GET'], permission=Permission.RESOURCE_VIEW) + async def get_executions(request_context: RequestContext) -> str: + """Unified execution list across agent runs and pipeline queries.""" + bot_ids = quart.request.args.getlist('botId') + pipeline_ids = quart.request.args.getlist('pipelineId') + agent_ids = quart.request.args.getlist('agentId') + statuses = quart.request.args.getlist('status') + source = quart.request.args.get('source', 'all') + mode = quart.request.args.get('mode', 'all') + start_time = parse_iso_datetime(quart.request.args.get('startTime')) + end_time = parse_iso_datetime(quart.request.args.get('endTime')) + + result = await self.ap.monitoring_service.get_executions( + request_context, + bot_ids=bot_ids if bot_ids else None, + pipeline_ids=pipeline_ids if pipeline_ids else None, + agent_ids=agent_ids if agent_ids else None, + statuses=statuses if statuses else None, + source=source, + mode=mode, + start_time=start_time, + end_time=end_time, + limit=quart.request.args.get('limit', 50), + offset=quart.request.args.get('offset', 0), + ) + + return self.success(data=result) + + @self.route( + '/executions//', + methods=['GET'], + permission=Permission.RESOURCE_VIEW, + ) + 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) + except ValueError as exc: + return self.fail(404, str(exc)) + return self.success(data=detail) + @self.route('/token-statistics', methods=['GET'], permission=Permission.RESOURCE_VIEW) async def get_token_statistics(request_context: RequestContext) -> str: """Get detailed token usage statistics (summary, per-model, timeseries).""" @@ -639,7 +680,7 @@ class MonitoringRouterGroup(group.RouterGroup): 'platform', ] else: - return self.error(message=f'Invalid export type: {export_type}', code=400) + return self.fail(400, f'Invalid export type: {export_type}') # Generate CSV content with UTF-8 BOM for Excel compatibility import io diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 0bae983a6..bb7407248 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -10,7 +10,10 @@ from sqlalchemy.dialects import postgresql as postgresql_dialect from sqlalchemy.dialects import sqlite as sqlite_dialect from ....core import app +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 ..authz import WorkspaceRequiredError from ..context import ExecutionContext from .tenant import TenantContext, require_workspace_uuid @@ -29,6 +32,76 @@ _HARD_MAX_MONITORING_OFFSET = 10000000 _DEFAULT_CLEANUP_BATCHES_PER_TABLE = 4 _HARD_MAX_CLEANUP_BATCHES_PER_TABLE = 100 +# Canonical execution status groups shared by the agent ledger and pipeline query +# records so both object types can be summarised with one vocabulary. +_AGENT_STATUS_GROUP = { + 'completed': 'completed', + 'failed': 'failed', + 'timeout': 'failed', + 'cancelled': 'cancelled', + 'created': 'queued', + 'queued': 'queued', + 'claimed': 'running', + 'running': 'running', +} +_PIPELINE_STATUS_GROUP = { + 'success': 'completed', + 'error': 'failed', + 'pending': 'running', + 'discarded': 'ignored', + 'not_matched': 'ignored', +} + + +def _status_group(status: str, source: str) -> str: + table = _AGENT_STATUS_GROUP if source == 'agent' else _PIPELINE_STATUS_GROUP + return table.get(status, status) + + +def _epoch_ms(value: datetime.datetime | None) -> int | None: + """Epoch milliseconds for a DB timestamp, treating naive values as UTC.""" + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=datetime.timezone.utc) + return int(round(value.timestamp() * 1000)) + + +def _percentile_ms(values: list[float], fraction: float) -> int | None: + """Nearest-rank percentile (ms) over a non-empty, prefix-sorted list.""" + if not values: + return None + rank = max(1, min(len(values), int(round(fraction * len(values))))) + return int(round(values[rank - 1])) + + +def _collect_text(node, out: list[str], depth: int = 0) -> None: + """Best-effort extraction of human text from a serialised message chain.""" + if depth > 6: + return + if isinstance(node, dict): + for key, value in node.items(): + if key == 'text' and isinstance(value, str): + out.append(value) + elif isinstance(value, (dict, list)): + _collect_text(value, out, depth + 1) + elif isinstance(node, list): + for item in node: + _collect_text(item, out, depth + 1) + + +def _message_preview(raw: str | None, limit: int = 160) -> str: + if not raw: + return '' + try: + parsed = json.loads(raw) + except (TypeError, ValueError): + return raw[:limit] + parts: list[str] = [] + _collect_text(parsed, parts) + text = ' '.join(part.strip() for part in parts if part.strip()) + return (text or raw)[:limit] + def _normalize_user_id(value: str | int | None) -> str | None: """Convert numeric platform IDs before binding a VARCHAR with asyncpg. @@ -885,6 +958,10 @@ class MonitoringService: active_sessions_result = await self.ap.persistence_mgr.execute_async(active_session_query) active_sessions = active_sessions_result.scalar() or 0 + # Execution-first KPI values (runs + pipeline queries, latency, token + # coverage) are served by `get_executions`, which applies the execution + # view filters. This endpoint stays the message/traffic rollup so the + # dashboard does not compute the same rollup twice per page load. return { 'total_messages': total_messages, 'llm_calls': llm_calls, @@ -894,6 +971,527 @@ class MonitoringService: 'active_sessions': active_sessions, } + # ========== Execution view ========== + + def _agent_run_conditions( + self, + workspace_uuid: str, + *, + bot_ids: list[str] | None = None, + agent_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] + if bot_ids: + conditions.append(AgentRun.bot_id.in_(bot_ids)) + if agent_ids: + conditions.append(AgentRun.agent_id.in_(agent_ids)) + if statuses is not None: + conditions.append(AgentRun.status.in_(statuses) if statuses else sqlalchemy.false()) + if start_time: + conditions.append(AgentRun.created_at >= start_time) + 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:%'), + ) + conditions.append(debug_condition if debug else sqlalchemy.not_(debug_condition)) + return conditions + + def _pipeline_execution_conditions( + self, + workspace_uuid: str, + *, + bot_ids: list[str] | None = None, + pipeline_ids: list[str] | None = None, + statuses: list[str] | None = None, + start_time: datetime.datetime | None = None, + end_time: datetime.datetime | None = None, + ) -> list: + MonitoringMessage = persistence_monitoring.MonitoringMessage + conditions = [ + MonitoringMessage.workspace_uuid == workspace_uuid, + # Assistant replies are outputs of a query, not an execution of their own. + sqlalchemy.or_(MonitoringMessage.role.is_(None), MonitoringMessage.role != 'assistant'), + ] + if bot_ids: + conditions.append(MonitoringMessage.bot_id.in_(bot_ids)) + if pipeline_ids: + conditions.append(MonitoringMessage.pipeline_id.in_(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) + return conditions + + @staticmethod + def _raw_agent_statuses(groups: list[str] | None) -> list[str] | None: + """Raw agent statuses for a group filter; empty means "no group applies".""" + if not groups: + return None + raw: list[str] = [] + for group in groups: + raw.extend(status for status, mapped in _AGENT_STATUS_GROUP.items() if mapped == group) + return raw + + @staticmethod + def _raw_pipeline_statuses(groups: list[str] | None) -> list[str] | None: + """Raw pipeline statuses for a group filter; empty means "no group applies".""" + if not groups: + return None + raw: list[str] = [] + for group in groups: + raw.extend(status for status, mapped in _PIPELINE_STATUS_GROUP.items() if mapped == group) + return raw + + @staticmethod + def _model_rows(result) -> list: + """Rows of one ``select(Model)`` executed through ``execute_async``. + + The Core keeps the connection result contract, where ``select(Model)`` + yields a *flat* row of the model's columns (``row.uuid``, ``row.status``) + rather than a one-item ORM row. ``.scalars()`` would therefore return the + first column - the id - instead of the row, so the rows are taken as-is. + """ + return list(result.all()) + + @staticmethod + def _deserialize_json(raw: str | None) -> dict: + if not raw: + return {} + try: + value = json.loads(raw) + except (TypeError, ValueError): + return {} + return value if isinstance(value, dict) else {} + + async def _agent_names(self, workspace_uuid: str, rows: list) -> dict[str, str]: + """The processor name of every Agent run in one page, in one query. + + 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} + if not agent_ids: + return {} + Agent = persistence_agent.Agent + result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(Agent.uuid, Agent.name).where( + sqlalchemy.and_( + Agent.workspace_uuid == workspace_uuid, + Agent.uuid.in_(agent_ids), + ) + ) + ) + return {row.uuid: row.name for row in self._model_rows(result)} + + def _serialize_agent_execution(self, row, agent_names: dict[str, str] | None = None) -> dict: + metadata = self._deserialize_json(row.metadata_json) + duration_ms = None + if row.started_at and row.finished_at: + duration_ms = int(round((row.finished_at - row.started_at).total_seconds() * 1000)) + 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:') + return { + 'source': 'agent', + 'id': row.run_id, + 'event_id': row.event_id, + '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_name': metadata.get('agent_name') + or metadata.get('target_name') + or (agent_names or {}).get(row.agent_id or ''), + 'bot_id': row.bot_id, + 'bot_name': metadata.get('bot_name'), + 'pipeline_id': None, + 'pipeline_name': None, + 'runner_id': row.runner_id, + 'conversation_id': row.conversation_id, + 'session_id': row.conversation_id, + 'created_at_ms': created_at_ms, + 'started_at_ms': started_at_ms, + 'finished_at_ms': finished_at_ms, + 'duration_ms': duration_ms, + 'usage': self._deserialize_json(row.usage_json) or None, + 'cost': self._deserialize_json(row.cost_json) or None, + 'queue_name': row.queue_name, + 'debug': is_debug, + 'has_error': row.status in {'failed', 'timeout'}, + } + + def _serialize_pipeline_execution(self, row) -> dict: + created_at_ms = _epoch_ms(row.timestamp) + return { + 'source': 'pipeline', + 'id': row.id, + 'event_id': None, + 'status': row.status, + 'status_group': _status_group(row.status, 'pipeline'), + 'title': _message_preview(row.message_content), + 'target_kind': 'pipeline', + 'target_id': row.pipeline_id, + 'target_name': row.pipeline_name, + 'bot_id': row.bot_id, + 'bot_name': row.bot_name, + 'pipeline_id': row.pipeline_id, + 'pipeline_name': row.pipeline_name, + 'runner_id': row.runner_name, + 'conversation_id': None, + 'session_id': row.session_id, + 'created_at_ms': created_at_ms, + 'started_at_ms': None, + 'finished_at_ms': None, + 'duration_ms': None, + 'usage': None, + 'cost': None, + 'queue_name': None, + 'debug': False, + 'has_error': row.status == 'error', + 'platform': row.platform, + 'user_id': row.user_id, + 'user_name': row.user_name, + } + + async def _get_token_coverage( + self, + workspace_uuid: str, + *, + bot_ids: list[str] | None = None, + pipeline_ids: list[str] | None = None, + start_time: datetime.datetime | None = None, + end_time: datetime.datetime | None = None, + ) -> dict: + LLMCall = persistence_monitoring.MonitoringLLMCall + conditions = [LLMCall.workspace_uuid == workspace_uuid] + if bot_ids: + conditions.append(LLMCall.bot_id.in_(bot_ids)) + if pipeline_ids: + conditions.append(LLMCall.pipeline_id.in_(pipeline_ids)) + if start_time: + conditions.append(LLMCall.timestamp >= start_time) + if end_time: + conditions.append(LLMCall.timestamp <= end_time) + result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select( + sqlalchemy.func.count(LLMCall.id), + sqlalchemy.func.coalesce(sqlalchemy.func.sum(LLMCall.total_tokens), 0), + sqlalchemy.func.sum(sqlalchemy.case((LLMCall.total_tokens > 0, 1), else_=0)), + ).where(sqlalchemy.and_(*conditions)) + ) + row = result.first() + calls, total, calls_with_usage = row if row else (0, 0, 0) + return { + 'total_tokens': int(total or 0), + 'calls': calls or 0, + 'calls_with_usage': calls_with_usage or 0, + } + + async def _get_execution_summary( + self, + workspace_uuid: str, + *, + bot_ids: list[str] | None = None, + pipeline_ids: list[str] | None = None, + start_time: datetime.datetime | None = None, + end_time: datetime.datetime | None = None, + ) -> 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 + ) + run_status_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(AgentRun.status, sqlalchemy.func.count(AgentRun.id)) + .where(sqlalchemy.and_(*run_conditions)) + .group_by(AgentRun.status) + ) + agent_counts = {row[0]: row[1] for row in run_status_result.all()} + agent_total = sum(agent_counts.values()) + + debug_count_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(sqlalchemy.func.count(AgentRun.id)).where( + sqlalchemy.and_( + *run_conditions, + sqlalchemy.or_( + AgentRun.binding_id.like('debug:%'), + AgentRun.conversation_id.like('debug:%'), + ), + ) + ) + ) + debug_count = debug_count_result.scalar() or 0 + + # Percentiles are computed in Python, so the sample must stay bounded: + # take the most recent completed runs of the window and label the count + # so the UI can state how many runs the numbers rest on. + timing_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(AgentRun.started_at, AgentRun.finished_at) + .where( + sqlalchemy.and_( + *run_conditions, + AgentRun.status == 'completed', + AgentRun.started_at.is_not(None), + AgentRun.finished_at.is_not(None), + ) + ) + .order_by(AgentRun.created_at.desc()) + .limit(self._detail_limit()) + ) + 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 + ) + pipeline_status_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select( + persistence_monitoring.MonitoringMessage.status, + sqlalchemy.func.count(persistence_monitoring.MonitoringMessage.id), + ) + .where(sqlalchemy.and_(*pipeline_conditions)) + .group_by(persistence_monitoring.MonitoringMessage.status) + ) + pipeline_counts = {row[0]: row[1] for row in pipeline_status_result.all()} + pipeline_total = sum(pipeline_counts.values()) + + # Pending interactions are gated by the same bot/window scope as the runs + # so "waiting input" cannot drift from the rest of the card. + waiting_conditions = [ + persistence_agent_interaction.AgentInteraction.workspace_id == workspace_uuid, + persistence_agent_interaction.AgentInteraction.status == 'pending', + ] + if bot_ids: + waiting_conditions.append(persistence_agent_interaction.AgentInteraction.bot_id.in_(bot_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_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(sqlalchemy.func.count(persistence_agent_interaction.AgentInteraction.id)).where( + sqlalchemy.and_(*waiting_conditions) + ) + ) + waiting = waiting_result.scalar() or 0 + + by_status: dict[str, int] = {} + for status, count in agent_counts.items(): + group = _status_group(status, 'agent') + by_status[group] = by_status.get(group, 0) + count + for status, count in pipeline_counts.items(): + group = _status_group(status, 'pipeline') + by_status[group] = by_status.get(group, 0) + count + + completed = by_status.get('completed', 0) + failed = by_status.get('failed', 0) + denominator = completed + failed + total = agent_total + pipeline_total + return { + 'total': total, + 'by_source': {'agent': agent_total, 'pipeline': pipeline_total}, + 'by_status': by_status, + 'completed': completed, + 'failed': failed, + 'running': by_status.get('running', 0), + 'queued': by_status.get('queued', 0), + 'cancelled': by_status.get('cancelled', 0), + 'ignored': by_status.get('ignored', 0), + 'waiting': waiting, + 'success_rate': round(completed / denominator * 100, 2) if denominator > 0 else None, + 'denominator': denominator, + 'p50_duration_ms': _percentile_ms(durations, 0.5), + 'p95_duration_ms': _percentile_ms(durations, 0.95), + 'duration_sample': len(durations), + 'debug': debug_count, + 'real': total - debug_count, + } + + async def get_executions( + self, + context: TenantContext, + *, + bot_ids: list[str] | None = None, + pipeline_ids: list[str] | None = None, + agent_ids: list[str] | None = None, + statuses: list[str] | None = None, + source: str = 'all', + mode: str = 'all', + start_time: datetime.datetime | None = None, + end_time: datetime.datetime | None = None, + limit: int = 50, + offset: int = 0, + ) -> dict: + """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'}: + 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( + workspace_uuid, + bot_ids=bot_ids, + agent_ids=agent_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) + ) + 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( + workspace_uuid, + bot_ids=bot_ids, + pipeline_ids=pipeline_ids, + statuses=self._raw_pipeline_statuses(statuses), + start_time=start_time, + end_time=end_time, + ) + 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) + ) + 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) + ) + ) + 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] + return { + 'items': page, + 'total': total, + 'limit': limit, + 'offset': offset, + 'has_more': total > offset + len(page), + 'summary': { + 'executions': await self._get_execution_summary( + workspace_uuid, + bot_ids=bot_ids, + pipeline_ids=pipeline_ids, + start_time=start_time, + end_time=end_time, + ), + 'tokens': await self._get_token_coverage( + workspace_uuid, + bot_ids=bot_ids, + pipeline_ids=pipeline_ids, + start_time=start_time, + end_time=end_time, + ), + }, + } + + 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, diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index 15ae38f47..c4c7fc8f1 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -18,6 +18,103 @@ if typing.TYPE_CHECKING: from .pool import get_query_execution_context +async def resolve_processor_name( + ap: app.Application, + workspace_uuid: str, + processor_type: str, + processor_id: str | None, + runner_id: str = '', +) -> str: + """Display name of the processor kind that handled a run. + + Monitoring attributed every call to a Pipeline, so an Agent or a plugin event + processor recorded "Unknown" for the entity that actually ran. Each kind is + resolved from the table that owns it, falling back to the identity the + binding carried. + """ + import sqlalchemy + + from ..entity.persistence import agent as persistence_agent + from ..entity.persistence import pipeline as persistence_pipeline + + identifier = str(processor_id or '').strip() + if identifier: + try: + model = persistence_pipeline.LegacyPipeline if processor_type == 'pipeline' else persistence_agent.Agent + result = await ap.persistence_mgr.execute_async( + sqlalchemy.select(model.name).where( + sqlalchemy.and_( + model.uuid == identifier, + model.workspace_uuid == workspace_uuid, + ) + ) + ) + name = str(result.scalar() or '').strip() + if name: + return name + except Exception as e: + ap.logger.warning(f'Failed to resolve monitoring processor name: {e}') + # A deleted Agent or an unregistered plugin processor still names the + # component that ran, never a bare "Unknown". + return identifier or runner_id or processor_type + + +async def prepare_monitoring_identity( + ap: app.Application, + query: pipeline_query.Query, + *, + processor_type: str, + processor_id: str | None, + runner_id: str = '', + execution_record_id: str = '', +) -> None: + """Pin the monitoring identity of a runner-owned execution. + + A Pipeline pins these variables when it starts a query; a Runner started by + an Agent or a plugin event processor has to pin them itself, otherwise every + LLM/tool call it makes is attributed to "Unknown". An enclosing Pipeline + already named the run, so its label is never overwritten. + + ``_monitoring_message_id`` is the execution record the calls belong to: the + Pipeline's message, or the Agent run's id. Both are the id the executions + list shows, which is what lets a call jump back to its record. + """ + try: + variables = getattr(query, 'variables', None) + if not isinstance(variables, dict): + return + if variables.get('_monitoring_pipeline_name'): + return + + bot_name = 'WebChat' + bot_uuid = str(getattr(query, 'bot_uuid', '') or '') + workspace_uuid = str(getattr(query, 'workspace_uuid', '') or '') + if bot_uuid and workspace_uuid: + try: + bot = await ap.bot_service.get_bot( + workspace_uuid, + bot_uuid, + include_secret=False, + ) + if bot: + bot_name = bot.get('name', bot_name) + except Exception as e: + ap.logger.warning(f'Failed to resolve monitoring bot name: {e}') + + variables['_monitoring_bot_name'] = bot_name + variables['_monitoring_pipeline_name'] = await resolve_processor_name( + ap, + workspace_uuid, + processor_type, + processor_id, + runner_id, + ) + if execution_record_id: + variables['_monitoring_message_id'] = execution_record_id + except Exception as e: + ap.logger.error(f'Failed to prepare monitoring identity: {e}') + + class MonitoringHelper: """Helper class for monitoring operations""" diff --git a/src/langbot/pkg/provider/tools/toolmgr.py b/src/langbot/pkg/provider/tools/toolmgr.py index 26329dad0..97e554f42 100644 --- a/src/langbot/pkg/provider/tools/toolmgr.py +++ b/src/langbot/pkg/provider/tools/toolmgr.py @@ -2,6 +2,7 @@ from __future__ import annotations import typing import time +import asyncio import inspect from typing import TYPE_CHECKING @@ -448,6 +449,19 @@ class ToolManager: start_time = time.perf_counter() try: result = await invoke() + except asyncio.CancelledError: + duration_ms = int((time.perf_counter() - start_time) * 1000) + await self._record_tool_call( + name=name, + source=source, + parameters=parameters, + query=query, + duration_ms=duration_ms, + status='error', + error_message='cancelled', + ) + self._trace_tool_call(source=source, name=name, query=query, outcome='cancelled', error='cancelled') + raise except Exception as e: duration_ms = int((time.perf_counter() - start_time) * 1000) await self._record_tool_call( @@ -459,6 +473,14 @@ class ToolManager: status='error', error_message=str(e), ) + outcome = 'timeout' if isinstance(e, TimeoutError) else 'failed' + self._trace_tool_call( + source=source, + name=name, + query=query, + outcome=outcome, + error=str(e) or type(e).__name__, + ) raise duration_ms = int((time.perf_counter() - start_time) * 1000) @@ -471,8 +493,60 @@ class ToolManager: status='success', result=result, ) + self._trace_tool_call(source=source, name=name, query=query, outcome='success') return result + def _trace_tool_call( + self, + *, + source: str, + name: str, + query: pipeline_query.Query, + outcome: str, + error: str = '', + ) -> None: + """Report one tool invocation as its own step of the execution chain. + + Every tool call a Run makes - sandbox, skill, plugin or MCP - is the + observable action of that Run, so the chain shows it under the step that + issued it instead of collapsing the Run into a single opaque node. + + An Agent's tool call is executed for the Runtime's RPC task, which never + inherited the chain context. The Run id the caller already carries is the + alias that task's record resolves the owning execution through. + """ + # Imported here: telemetry pulls the application, and the tool manager is + # constructed while the application is still assembling. + from ...telemetry.execution import record as record_execution + from ...telemetry.platform import processing_mode + + try: + record_execution( + self.ap, + get_query_execution_context(query), + execution_id=self._trace_execution_id(query), + family='tool', + operation=name, + adapter=source, + mode=processing_mode.get(), + outcome=outcome, + error=error, + ) + except Exception as e: + self.ap.logger.warning(f'Failed to record tool call trace: {e}') + + @staticmethod + def _trace_execution_id(query: pipeline_query.Query) -> str: + """The execution chain this call belongs to, when the caller knows it. + + ``''`` lets the emitter fall back to the task's own trace context, which + is what a Pipeline-lane tool call has. + """ + session = getattr(query, '_agent_run_session', None) + if not isinstance(session, dict): + return '' + return str(session.get('run_id') or '').strip() + async def execute_func_call( self, name: str, diff --git a/src/langbot/pkg/telemetry/execution.py b/src/langbot/pkg/telemetry/execution.py index 1e5a16a1c..8a72288f2 100644 --- a/src/langbot/pkg/telemetry/execution.py +++ b/src/langbot/pkg/telemetry/execution.py @@ -33,7 +33,7 @@ from .trace import TraceState MODES = frozenset({'pipeline', 'agent', 'event_processor', 'none'}) OUTCOMES = frozenset({'success', 'failed', 'cancelled', 'timeout', 'skipped', 'unknown'}) -FAMILIES = frozenset({'platform_event', 'event_route', 'pipeline', 'runner', 'platform_api'}) +FAMILIES = frozenset({'platform_event', 'event_route', 'pipeline', 'runner', 'platform_api', 'tool'}) ORIGINS = frozenset({'platform', 'webui', 'api'}) CLOSED_BY = frozenset({'event_done', 'pipeline_done', 'runner_done', 'timeout'}) CHAIN_FAILURES = frozenset({'failed', 'cancelled', 'timeout'}) @@ -255,9 +255,9 @@ class ExecutionCounters: state = trace_mod.current() if state is None: # Cross-task work has no inherited context; resolve the owning - # execution through the registry. It has no open step here, so - # its node becomes a root of that execution instead of being - # mirrored from another task's stack. + # execution through the registry. It owns no step here, so the + # record attaches to the innermost step the chain has open - + # the step that caused this work. lookup = str(execution_id or '').strip() or current_execution_id() if lookup: state = self._resolve(lookup) diff --git a/src/langbot/pkg/telemetry/trace.py b/src/langbot/pkg/telemetry/trace.py index fb4d18031..6ae8da2d1 100644 --- a/src/langbot/pkg/telemetry/trace.py +++ b/src/langbot/pkg/telemetry/trace.py @@ -11,12 +11,17 @@ small execution-scoped context the emitter needs: * the routing/run identity pinned around the block that produced a node; * the open-step stack, which lives in a ContextVar holding *one task's own immutable tuple*. Child tasks inherit a snapshot at creation; no shared - mutable stack and no cross-task mirror exists, so sibling tasks can never - corrupt each other's parent links. + mutable stack exists, so sibling tasks can never corrupt each other's parent + links. Cross-task callers (plugin/RPC work with no inherited context) resolve their execution through the owner's execution-id registry instead, which only -carries the execution state - never node payloads. +carries the execution state - never node payloads. Such a caller owns no step +of its own, yet its work is still caused by a step of the chain: the sandbox +tool a Runner invoked, the platform API a plugin called back over RPC. The +execution state therefore also keeps the ordered set of steps any task +currently has open, and a record emitted with no open step of its own attaches +to the innermost one of those instead of becoming a root of the chain. """ from __future__ import annotations @@ -77,6 +82,7 @@ class TraceState: 'pipeline_plugins', 'node_adapter', 'node_runner', + 'open_frames', ) def __init__(self, execution_id: str | None = None) -> None: @@ -110,6 +116,10 @@ class TraceState: # First adapter/runner seen on a node, used for the chain record. self.node_adapter = '' self.node_runner = '' + # Steps currently open anywhere in this chain, in the order they were + # opened. A record emitted by a task that owns no step of its own + # attaches to the innermost entry here. + self.open_frames: dict[str, NodeFrame] = {} def allocate_seq(self) -> int: """Mint the next chain-local node order.""" @@ -117,6 +127,17 @@ class TraceState: self.node_seq += 1 return seq + def open_lane(self) -> str: + """The innermost step open in this chain, whichever task opened it.""" + frames = list(self.open_frames) + return frames[-1] if frames else '' + + def open_step(self, frame: NodeFrame) -> None: + self.open_frames[frame.node_id] = frame + + def close_step(self, frame: NodeFrame) -> None: + self.open_frames.pop(frame.node_id, None) + def note_node( self, *, @@ -221,8 +242,10 @@ def resolve_node( A record stamped with a ``node`` the calling task has open reuses that step's identity, order and start time; anything else (a plain observation, - or a node opened by another task) is a fresh node under this task's - innermost open step. + or a node opened by another task) is a fresh node under the innermost step + open in this task. A task that inherited no step at all - plugin/RPC work + that only knows the execution - attaches to the innermost step the chain + has open, which is the step that caused the work. """ stack = _node.get() if node: @@ -231,7 +254,7 @@ def resolve_node( if frame.node_id == node: parent = stack[index - 1].node_id if index > 0 else '' return frame.node_id, frame.seq, frame.started_at, parent - parent = stack[-1].node_id if stack else '' + parent = stack[-1].node_id if stack else state.open_lane() return node or uuid4().hex, state.allocate_seq(), fallback_started_at, parent @@ -250,9 +273,11 @@ def stage_scope() -> typing.Iterator[str]: return frame = NodeFrame(uuid4().hex, state.allocate_seq(), _now()) token = _node.set(_node.get() + (frame,)) + state.open_step(frame) try: yield frame.node_id finally: + state.close_step(frame) try: _node.reset(token) except (ValueError, RuntimeError): diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py new file mode 100644 index 000000000..cd8d767ec --- /dev/null +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -0,0 +1,350 @@ +"""Execution-view tests for the monitoring service. + +Covers the unified agent-run + pipeline-query execution list, its rollup +summary, and single-execution traces. +""" + +from __future__ import annotations + +import datetime +from types import SimpleNamespace + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.orm import sessionmaker + +from langbot.pkg.api.http.service.monitoring import MonitoringService +from langbot.pkg.entity.persistence import agent_interaction as persistence_interaction +from langbot.pkg.entity.persistence import agent_run as persistence_agent_run +from langbot.pkg.entity.persistence import monitoring as persistence_monitoring +from langbot.pkg.entity.persistence import workspace as persistence_workspace # noqa: F401 +from langbot.pkg.entity.persistence.base import Base + +WORKSPACE = 'workspace-1' + + +class FakePersistenceManager: + """Runs statements against a real in-memory SQLite engine.""" + + def __init__(self, engine): + self._engine = engine + self._factory = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + def get_db_engine(self): + return self._engine + + async def execute_async(self, statement): + # Match PersistenceManager's connection-level result shape. + async with self._engine.connect() as connection: + return await connection.execute(statement) + + @staticmethod + def serialize_model(model, data): + out = {} + for column in model.__table__.columns: + value = getattr(data, column.name) + out[column.name] = value.isoformat() if isinstance(value, datetime.datetime) else value + return out + + +@pytest.fixture +async def engine(tmp_path): + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "monitoring_executions.db"}') + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield engine + await engine.dispose() + + +@pytest.fixture +def service(engine): + ap = SimpleNamespace( + persistence_mgr=FakePersistenceManager(engine), + instance_config=None, + ) + return MonitoringService(ap) + + +def _dt(hour: int, minute: int = 0, second: int = 0) -> datetime.datetime: + return datetime.datetime(2026, 1, 1, hour, minute, second) + + +async def _seed(engine) -> None: + factory = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + session.add_all( + [ + persistence_agent_run.AgentRun( + run_id='run-done', + event_id='evt-1', + agent_id='agent-1', + binding_id='agent_agent-1_default', + runner_id='runner-1', + workspace_id=WORKSPACE, + bot_id='bot-1', + status='completed', + created_at=_dt(10), + started_at=_dt(10, 0, 1), + finished_at=_dt(10, 0, 3), + usage_json='{"total_tokens": 42}', + ), + persistence_agent_run.AgentRun( + run_id='run-failed', + binding_id='agent_agent-1_default', + runner_id='runner-1', + workspace_id=WORKSPACE, + bot_id='bot-1', + status='failed', + created_at=_dt(10, 5), + ), + persistence_agent_run.AgentRun( + run_id='run-debug', + event_id='debug:agent-1:e1', + agent_id='agent-1', + binding_id='debug:agent-1:runner-1', + runner_id='runner-1', + workspace_id=WORKSPACE, + bot_id='bot-1', + status='completed', + created_at=_dt(10, 6), + started_at=_dt(10, 6, 0), + finished_at=_dt(10, 6, 1), + ), + persistence_agent_run.AgentRunEvent( + run_id='run-done', + sequence=1, + type='run.completed', + data_json='{"message": {"role": "assistant", "content": "hi"}}', + ), + persistence_interaction.AgentInteraction( + interaction_id='int-1', + run_id='run-done', + binding_id='agent_agent-1_default', + runner_id='runner-1', + processor_type='agent', + processor_id='agent-1', + workspace_id=WORKSPACE, + status='pending', + request_json='{}', + callback_token_hash='hash-1', + ), + ] + ) + for message in [ + dict( + id='msg-user', + timestamp=_dt(10, 2), + session_id='session-1', + message_content='{"chain": [{"type": "Plain", "text": "hello there"}]}', + status='success', + role='user', + ), + dict( + id='msg-assistant', + timestamp=_dt(10, 3), + session_id='session-1', + message_content='{"chain": [{"type": "Plain", "text": "reply"}]}', + status='success', + role='assistant', + ), + dict( + id='msg-discarded', + timestamp=_dt(10, 4), + session_id='session-2', + message_content='{"chain": [{"type": "Plain", "text": "ignored"}]}', + status='discarded', + role='user', + ), + ]: + session.add( + persistence_monitoring.MonitoringMessage( + workspace_uuid=WORKSPACE, + bot_id='bot-1', + bot_name='Bot One', + pipeline_id='pipeline-1', + pipeline_name='Pipeline One', + level='info', + **message, + ) + ) + session.add_all( + [ + persistence_monitoring.MonitoringLLMCall( + id='llm-1', + workspace_uuid=WORKSPACE, + timestamp=_dt(10, 2, 30), + model_name='gpt', + input_tokens=10, + output_tokens=20, + total_tokens=30, + duration=100, + status='success', + bot_id='bot-1', + bot_name='Bot One', + pipeline_id='pipeline-1', + pipeline_name='Pipeline One', + session_id='session-1', + message_id='msg-user', + ), + persistence_monitoring.MonitoringLLMCall( + id='llm-2', + workspace_uuid=WORKSPACE, + timestamp=_dt(10, 4, 30), + model_name='gpt', + input_tokens=0, + output_tokens=0, + total_tokens=0, + duration=50, + status='success', + bot_id='bot-1', + bot_name='Bot One', + pipeline_id='pipeline-1', + pipeline_name='Pipeline One', + session_id='session-2', + message_id='msg-discarded', + ), + ] + ) + session.add( + persistence_monitoring.MonitoringSession( + workspace_uuid=WORKSPACE, + bot_id='bot-1', + session_id='session-1', + bot_name='Bot One', + pipeline_id='pipeline-1', + pipeline_name='Pipeline One', + message_count=2, + start_time=_dt(10, 2), + last_activity=_dt(10, 5), + is_active=True, + ) + ) + await session.commit() + + +@pytest.mark.asyncio +async def test_executions_merge_and_order_both_sources(engine, service): + await _seed(engine) + + result = await service.get_executions(WORKSPACE, limit=50) + + assert result['total'] == 5 + ids = [(item['source'], item['id']) for item in result['items']] + # Assistant replies are outputs, not executions. + assert ('pipeline', 'msg-assistant') not in ids + # Newest first across both object types. + assert ids == [ + ('agent', 'run-debug'), + ('agent', 'run-failed'), + ('pipeline', 'msg-discarded'), + ('pipeline', 'msg-user'), + ('agent', 'run-done'), + ] + + by_id = {item['id']: item for item in result['items']} + assert result['summary']['executions']['total'] == 5 + assert result['summary']['executions']['denominator'] == 4 + assert result['summary']['tokens']['calls'] == 2 + assert result['summary']['tokens']['calls_with_usage'] == 1 + assert by_id['run-done']['status_group'] == 'completed' + assert by_id['run-failed']['status_group'] == 'failed' + assert by_id['run-debug']['debug'] is True + assert by_id['run-done']['debug'] is False + assert by_id['run-done']['usage'] == {'total_tokens': 42} + assert by_id['run-done']['duration_ms'] == 2000 + assert by_id['msg-user']['status_group'] == 'completed' + assert by_id['msg-discarded']['status_group'] == 'ignored' + assert by_id['msg-user']['title'] == 'hello there' + assert by_id['msg-user']['target_id'] == 'pipeline-1' + + +@pytest.mark.asyncio +async def test_executions_source_status_and_paging(engine, service): + await _seed(engine) + + only_agents = await service.get_executions(WORKSPACE, source='agent') + assert {item['source'] for item in only_agents['items']} == {'agent'} + assert only_agents['total'] == 3 + + pipeline_only = await service.get_executions(WORKSPACE, source='pipeline') + assert {item['source'] for item in pipeline_only['items']} == {'pipeline'} + assert pipeline_only['total'] == 2 + + # A single-object target selection narrows the union to that source. + by_pipeline = await service.get_executions(WORKSPACE, pipeline_ids=['pipeline-1']) + assert {item['source'] for item in by_pipeline['items']} == {'pipeline'} + + failed = await service.get_executions(WORKSPACE, statuses=['failed']) + assert [item['id'] for item in failed['items']] == ['run-failed'] + + # `ignored` only exists for pipeline queries: it must not widen the agent + # filter back to "every status". + ignored = await service.get_executions(WORKSPACE, statuses=['ignored']) + assert [item['id'] for item in ignored['items']] == ['msg-discarded'] + + debug_only = await service.get_executions(WORKSPACE, mode='debug') + assert [item['id'] for item in debug_only['items']] == ['run-debug'] + + first_page = await service.get_executions(WORKSPACE, limit=2) + assert len(first_page['items']) == 2 + assert first_page['has_more'] is True + second_page = await service.get_executions(WORKSPACE, limit=2, offset=2) + assert first_page['items'][0]['id'] != second_page['items'][0]['id'] + + +@pytest.mark.asyncio +async def test_execution_summary_uses_explicit_denominator(engine, service): + await _seed(engine) + + summary = (await service.get_executions(WORKSPACE))['summary'] + executions = summary['executions'] + + # completed = run-done + run-debug + msg-user; failed = run-failed. + assert executions['by_source'] == {'agent': 3, 'pipeline': 2} + assert executions['completed'] == 3 + assert executions['failed'] == 1 + assert executions['ignored'] == 1 + assert executions['denominator'] == 4 + assert executions['success_rate'] == 75.0 + assert executions['debug'] == 1 + assert executions['real'] == 4 + assert executions['waiting'] == 1 + assert executions['p50_duration_ms'] == 1000 + assert executions['p95_duration_ms'] == 2000 + # Only run-done and run-debug are completed runs with a start and a finish. + assert executions['duration_sample'] == 2 + + # Token coverage is explicit: llm-2 succeeded without usage. + assert summary['tokens'] == {'total_tokens': 30, 'calls': 2, 'calls_with_usage': 1} + + +@pytest.mark.asyncio +async def test_overview_metrics_keep_message_rollup(engine, service): + await _seed(engine) + + overview = await service.get_overview_metrics(WORKSPACE) + + assert overview['total_messages'] == 3 + assert overview['llm_calls'] == 2 + assert overview['model_calls'] == 2 + assert overview['active_sessions'] == 1 + + +@pytest.mark.asyncio +async def test_execution_detail_for_agent_and_pipeline(engine, service): + await _seed(engine) + + agent_detail = await service.get_execution_detail(WORKSPACE, 'agent', 'run-done') + assert agent_detail['run']['run_id'] == 'run-done' + assert agent_detail['events'][0]['type'] == 'run.completed' + + pipeline_detail = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user') + assert pipeline_detail['message']['id'] == 'msg-user' + assert [call['id'] for call in pipeline_detail['llm_calls']] == ['llm-1'] + assert pipeline_detail['tool_calls'] == [] + + with pytest.raises(ValueError): + await service.get_execution_detail(WORKSPACE, 'agent', 'missing-run') + + with pytest.raises(ValueError): + await service.get_execution_detail(WORKSPACE, 'pipeline', 'missing-message') diff --git a/tests/unit_tests/provider/test_tool_manager.py b/tests/unit_tests/provider/test_tool_manager.py index d404400a4..0e81ae5e2 100644 --- a/tests/unit_tests/provider/test_tool_manager.py +++ b/tests/unit_tests/provider/test_tool_manager.py @@ -7,6 +7,8 @@ Tests cover: from __future__ import annotations +import asyncio +import contextvars from importlib import import_module from types import SimpleNamespace from unittest.mock import AsyncMock, Mock @@ -380,6 +382,127 @@ class TestToolManagerExecuteFuncCall: assert 'MCP resource agent reads are disabled' in record.await_args.kwargs['error_message'] +class _RecordingTelemetrySender: + """Telemetry sender stand-in that keeps every payload it was handed.""" + + def __init__(self): + self.telemetry_config = {'url': 'https://space.example.test', 'execution_trace': 'all'} + self.sent: list[dict] = [] + + async def send(self, payload: dict) -> bool: + self.sent.append(payload) + return True + + +class TestToolCallTracing: + """Every tool call a Run makes is its own step of the execution chain.""" + + @pytest.fixture + def traced_app(self): + execution = import_module('langbot.pkg.telemetry.execution') + counters = execution.ExecutionCounters(_RecordingTelemetrySender()) + app = Mock() + app.logger = Mock() + app.telemetry = SimpleNamespace(execution=counters) + # Tool-call monitoring is asserted by the tests above; this path only + # needs the execution chain. + app.monitoring_service = None + + def _inert_loader(): + loader = Mock() + loader.has_tool = AsyncMock(return_value=False) + loader.invoke_tool = AsyncMock(return_value=None) + return loader + + app._native_loader = _inert_loader() + app._plugin_loader = _inert_loader() + app._mcp_loader = _inert_loader() + app._skill_loader = _inert_loader() + return app, counters + + @pytest.fixture + def sample_query(self): + query = Mock(spec=pipeline_query.Query) + query._execution_context = _CONTEXT + return query + + @staticmethod + def _manager(app): + """ToolManager wired to app-local inert loaders.""" + manager = get_toolmgr_module().ToolManager(app) + manager.native_tool_loader = app._native_loader + manager.plugin_tool_loader = app._plugin_loader + manager.mcp_tool_loader = app._mcp_loader + manager.skill_tool_loader = app._skill_loader + return manager + + @staticmethod + def _tool_nodes(counters): + return [record for record in counters.records if record['event_type'] == 'execution_node'] + + @pytest.mark.asyncio + async def test_a_tool_call_becomes_a_node_under_the_step_that_issued_it(self, traced_app, sample_query): + trace = import_module('langbot.pkg.telemetry.trace') + execution = import_module('langbot.pkg.telemetry.execution') + app, counters = traced_app + manager = self._manager(app) + + binding = execution.bind_trace(app, 'exec-tool') + # The Runtime's RPC task executes the call with no inherited trace + # context; the Run id on the query is what resolves the owning chain. + sample_query._agent_run_session = {'run_id': 'exec-tool'} + try: + with trace.stage_scope() as lane: + foreign = contextvars.Context() + task = asyncio.get_running_loop().create_task( + manager._invoke_tool_with_monitoring( + source='native', + name='exec', + parameters={'command': 'ls'}, + query=sample_query, + invoke=AsyncMock(return_value='sandbox output'), + ), + context=foreign, + ) + assert await task == 'sandbox output' + finally: + trace.unbind_root(binding) + + rows = self._tool_nodes(counters) + assert len(rows) == 1 + assert rows[0]['family'] == 'tool' + assert rows[0]['operation'] == 'exec' + assert rows[0]['adapter'] == 'native' + assert rows[0]['outcome'] == 'success' + assert rows[0]['parent_node_id'] == lane + + @pytest.mark.asyncio + async def test_a_failed_tool_call_carries_its_reason(self, traced_app, sample_query): + execution = import_module('langbot.pkg.telemetry.execution') + trace = import_module('langbot.pkg.telemetry.trace') + app, counters = traced_app + manager = self._manager(app) + + binding = execution.bind_trace(app, 'exec-tool-failed') + try: + with pytest.raises(RuntimeError): + await manager._invoke_tool_with_monitoring( + source='mcp', + name='read_resource', + parameters={}, + query=sample_query, + invoke=AsyncMock(side_effect=RuntimeError('server unreachable')), + ) + finally: + trace.unbind_root(binding) + + rows = self._tool_nodes(counters) + assert len(rows) == 1 + assert rows[0]['family'] == 'tool' + assert rows[0]['outcome'] == 'failed' + assert rows[0]['error'] == 'server unreachable' + + class TestToolManagerSourceResolution: @pytest.mark.asyncio @pytest.mark.parametrize('source', ['mcp', 'plugin']) diff --git a/tests/unit_tests/telemetry/test_trace.py b/tests/unit_tests/telemetry/test_trace.py index 95b9f06f1..381b0ee6b 100644 --- a/tests/unit_tests/telemetry/test_trace.py +++ b/tests/unit_tests/telemetry/test_trace.py @@ -344,10 +344,53 @@ class TestTraceNodeScope: lane_row, foreign_row = nodes(counters) assert lane_row['node_id'] == lane and lane_row['parent_node_id'] == '' and lane_row['root'] is True - # The cross-task node joins the same execution, but never mirrors another - # task's open-step stack: it is a root of the chain. + # The cross-task node joins the same execution and hangs off the step the + # chain had open - the step that caused the work - instead of mirroring + # any task's own stack. assert foreign_row['event_id'] == 'exec-cross' - assert foreign_row['parent_node_id'] == '' + assert foreign_row['parent_node_id'] == lane + assert foreign_row['root'] is False + + def test_cross_task_observation_attaches_to_the_innermost_open_step(self): + trace, execution = get_modules() + _, counters = make_counters(trace_config()) + binding = execution.bind_trace(make_ap(counters), 'exec-nested') + try: + with trace.stage_scope() as outer: + counters.record(CONTEXT, **{**STAGE, 'node': outer}) + with trace.stage_scope() as inner: + counters.record(CONTEXT, **{**STAGE, 'node': inner}) + foreign = contextvars.Context() + foreign.run(execution.set_execution_id, 'exec-nested') + foreign.run(counters.record, CONTEXT, **{**STAGE, 'family': 'tool', 'operation': 'exec'}) + finally: + trace.unbind_root(binding) + + outer_row, inner_row, tool_row = nodes(counters) + assert outer_row['parent_node_id'] == '' + assert inner_row['parent_node_id'] == outer + # The lane that caused the cross-task work is the innermost one open. + assert tool_row['parent_node_id'] == inner + + def test_cross_task_observation_after_the_lane_closed_is_a_root(self): + trace, execution = get_modules() + _, counters = make_counters(trace_config()) + binding = execution.bind_trace(make_ap(counters), 'exec-late') + try: + with trace.stage_scope() as lane: + counters.record(CONTEXT, **{**STAGE, 'node': lane}) + # A late callback arrives after the owning step returned: nothing is + # open anymore, so the node is a root of the chain rather than a + # child of a finished step. + foreign = contextvars.Context() + foreign.run(execution.set_execution_id, 'exec-late') + foreign.run(counters.record, CONTEXT, family='platform_api', operation='reply', mode='pipeline') + finally: + trace.unbind_root(binding) + + lane_row, late_row = nodes(counters) + assert late_row['parent_node_id'] == '' and late_row['root'] is False + assert lane_row['root'] is True def test_cross_task_observation_without_an_open_step_has_no_parent(self): trace, execution = get_modules() diff --git a/web/src/app/home/monitoring/components/ConversationTurnList.tsx b/web/src/app/home/monitoring/components/ConversationTurnList.tsx index afcaa1388..a1e61b03e 100644 --- a/web/src/app/home/monitoring/components/ConversationTurnList.tsx +++ b/web/src/app/home/monitoring/components/ConversationTurnList.tsx @@ -12,6 +12,7 @@ import { Wrench, } from 'lucide-react'; import { cn } from '@/lib/utils'; +import { Button } from '@/components/ui/button'; import { MessageContentRenderer } from './MessageContentRenderer'; import { ConversationTurn, @@ -507,10 +508,11 @@ export function ConversationTurnList({ key={call.id} className="border-t border-border/70 py-2 first:border-t-0 first:pt-0 last:pb-0" > - + {hasToolDetails && expandedToolCall && (
{exporting ? ( <> - - - - + {t('monitoring.export.exporting')} ) : ( diff --git a/web/src/app/home/monitoring/components/FeedbackCard.tsx b/web/src/app/home/monitoring/components/FeedbackCard.tsx index c2e93e52b..b5882b7f1 100644 --- a/web/src/app/home/monitoring/components/FeedbackCard.tsx +++ b/web/src/app/home/monitoring/components/FeedbackCard.tsx @@ -9,6 +9,7 @@ import { Heart, Smile, } from 'lucide-react'; +import { MetricCard } from './overview-cards/MetricCard'; interface FeedbackCardProps { title: string; @@ -23,6 +24,18 @@ interface FeedbackCardProps { loading?: boolean; } +// The variant only tints the icon tile; the card surface stays neutral so the +// feedback row matches the token-monitoring cards. +const variantAccents: Record< + NonNullable, + string +> = { + default: '#64748b', + success: '#10b981', + warning: '#f59e0b', + danger: '#ef4444', +}; + export function FeedbackCard({ title, value, @@ -32,89 +45,37 @@ export function FeedbackCard({ variant = 'default', loading = false, }: FeedbackCardProps) { - const variantStyles = { - default: 'bg-white dark:bg-[#2a2a2e] border-gray-200 dark:border-gray-700', - success: - 'bg-green-50 dark:bg-green-900/20 border-green-200 dark:border-green-800', - warning: - 'bg-yellow-50 dark:bg-yellow-900/20 border-yellow-200 dark:border-yellow-800', - danger: 'bg-red-50 dark:bg-red-900/20 border-red-200 dark:border-red-800', - }; - - const iconStyles = { - default: 'text-gray-500 dark:text-gray-400', - success: 'text-green-500 dark:text-green-400', - warning: 'text-yellow-500 dark:text-yellow-400', - danger: 'text-red-500 dark:text-red-400', - }; - const trendStyles = { up: 'text-green-500', - down: 'text-red-500', - neutral: 'text-gray-500', + down: 'text-destructive', + neutral: 'text-muted-foreground', }; - if (loading) { - return ( -
-
-
-
-
-
-
-
-
-
- ); - } - return ( -
-
-
-

- {title} -

-

- {value} -

- {subtitle && ( -

- {subtitle} -

- )} - {trend && ( -
- {trend.direction === 'up' && ( - - )} - {trend.direction === 'down' && ( - - )} - {trend.direction === 'neutral' && ( - - )} - - {trend.value > 0 ? '+' : ''} - {trend.value}% - -
- )} -
+ {trend && (
- {icon} + {trend.direction === 'up' && } + {trend.direction === 'down' && ( + + )} + {trend.direction === 'neutral' && } + + {trend.value > 0 ? '+' : ''} + {trend.value}% +
-
-
+ )} + ); } @@ -135,35 +96,37 @@ export function FeedbackStatsCards({ stats, loading }: FeedbackStatsProps) { { title: t('monitoring.feedback.totalFeedback'), value: stats?.totalFeedback ?? 0, - icon: , + icon: , variant: 'default' as const, }, { title: t('monitoring.feedback.totalLikes'), value: stats?.totalLikes ?? 0, - icon: , + icon: , variant: 'success' as const, }, { title: t('monitoring.feedback.totalDislikes'), value: stats?.totalDislikes ?? 0, - icon: , + icon: , variant: 'danger' as const, }, { title: t('monitoring.feedback.satisfactionRate'), value: stats ? `${stats.satisfactionRate}%` : '0%', - icon: , + icon: , + // The rate is the one feedback metric that is good or bad on its own, so + // its tile keeps encoding the verdict (as it did before the restyle). variant: (stats && stats.satisfactionRate >= 80 ? 'success' : stats && stats.satisfactionRate >= 50 ? 'warning' - : 'danger') as 'default' | 'success' | 'warning' | 'danger', + : 'danger') as 'success' | 'warning' | 'danger', }, ]; return ( -
+
{cards.map((card, index) => ( - -
- ); + return ; } if (!feedback || feedback.length === 0) { return ( -
- -

- {t('monitoring.feedback.noFeedback')} -

-

- {t('monitoring.feedback.noFeedbackDescription')} -

-
+ } + title={t('monitoring.feedback.noFeedback')} + hint={t('monitoring.feedback.noFeedbackDescription')} + /> ); } return (
{feedback.map((item) => ( -
{item.botName && ( <> - → - + → + {item.botName} )} {item.platform && ( - + {item.platform} - + )} {item.streamId && onViewMessage && ( - ))} -
+ { + if (value === 'hour' || value === 'day') setBucket(value); + }} + > + + {(['hour', 'day'] as const).map((b) => ( + + {t(`monitoring.tokens.bucket.${b}`)} + + ))} + +
@@ -372,93 +334,86 @@ export default function TokenMonitoring({
-
+ {/* Per-model breakdown */} -
+

{t('monitoring.tokens.byModel')}

-
- - - - - - - - - - - - - - {by_model.map((m) => { - const share = - summary.total_tokens > 0 - ? (m.total_tokens / summary.total_tokens) * 100 - : 0; - return ( - - - - - - - - - - ); - })} - -
- {t('monitoring.tokens.model')} - - {t('monitoring.tokens.calls')} - - {t('monitoring.tokens.inputTokens')} - - {t('monitoring.tokens.outputTokens')} - - {t('monitoring.tokens.totalTokens')} - - {t('monitoring.tokens.avgPerCall')} - - {t('monitoring.tokens.avgLatency')} -
-
- {m.model_name} -
-
-
-
-
- {m.calls} - {m.error_calls > 0 && ( - - {' '} - ({m.error_calls}✕) - - )} - - {formatNumber(m.input_tokens)} - - {formatNumber(m.output_tokens)} - - {formatNumber(m.total_tokens)} - - {formatNumber(m.avg_tokens_per_call)} - - {m.avg_duration_ms}ms -
-
-
+ + + + {t('monitoring.tokens.model')} + + {t('monitoring.tokens.calls')} + + + {t('monitoring.tokens.inputTokens')} + + + {t('monitoring.tokens.outputTokens')} + + + {t('monitoring.tokens.totalTokens')} + + + {t('monitoring.tokens.avgPerCall')} + + + {t('monitoring.tokens.avgLatency')} + + + + + {by_model.map((m) => { + const share = + summary.total_tokens > 0 + ? (m.total_tokens / summary.total_tokens) * 100 + : 0; + return ( + + +
+ {m.model_name} +
+
+
+
+ + + {m.calls} + {m.error_calls > 0 && ( + + {' '} + ({m.error_calls}✕) + + )} + + + {formatNumber(m.input_tokens)} + + + {formatNumber(m.output_tokens)} + + + {formatNumber(m.total_tokens)} + + + {formatNumber(m.avg_tokens_per_call)} + + + {m.avg_duration_ms}ms + + + ); + })} + +
+
); } diff --git a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx new file mode 100644 index 000000000..f0ad3c947 --- /dev/null +++ b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx @@ -0,0 +1,361 @@ +import { useEffect, useRef, useState } from 'react'; +import { useTranslation } from 'react-i18next'; +import { Link } from 'react-router-dom'; +import { ExternalLink } from 'lucide-react'; +import { + Sheet, + SheetContent, + SheetDescription, + SheetHeader, + SheetTitle, +} from '@/components/ui/sheet'; +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 type { + ExecutionCallRow, + ExecutionDetail, + ExecutionRow, +} from '@/app/infra/entities/api/monitoring-executions'; +import AgentExecutionTrace from '@/app/home/agents/components/AgentExecutionTrace'; +import { formatRunDuration } from '@/app/home/agents/components/processor-run-timing'; +import { formatDateTime } from '../../utils/dateUtils'; +import { + executionKindLabel, + executionProcessorLabel, + executionStatusLabel, + executionStatusVariant, + executionTargetLabel, +} from './ExecutionTable'; + +interface ExecutionDetailSheetProps { + row: ExecutionRow | null; + onClose: () => void; +} + +/** 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}`; +} + +export default function ExecutionDetailSheet({ + row, + onClose, +}: ExecutionDetailSheetProps) { + const { t } = useTranslation(); + const [detail, setDetail] = useState(null); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(false); + const requestRef = useRef(0); + + useEffect(() => { + if (!row) { + setDetail(null); + setError(false); + setLoading(false); + return; + } + const requestId = requestRef.current + 1; + requestRef.current = requestId; + setLoading(true); + setError(false); + setDetail(null); + backendClient + .getExecutionDetail(row.source, row.id) + .then((result) => { + if (requestRef.current !== requestId) return; + setDetail(result); + }) + .catch(() => { + if (requestRef.current !== requestId) return; + 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 (row.runner_id) { + metadata.push({ + label: t('monitoring.execution.detail.runner'), + value: row.runner_id, + }); + } + 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)}` + : 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')} + + +
+ {row && ( + <> + + + {executionProcessorLabel(row)} + +

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

+
+ + {executionKindLabel(row, t)} + + + {executionStatusLabel(row, t)} + + {row.debug && ( + + {t('monitoring.execution.debug')} + + )} + {objectHref && ( + + )} +
+ + +
+ {metadata.map((item) => ( +
+
+ {item.label} +
+
+ {item.value} +
+
+ ))} +
+ + + + {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, + )} + + )} +
+ )} + + )} +
+ + + + ); +} diff --git a/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx b/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx new file mode 100644 index 000000000..901adef5d --- /dev/null +++ b/web/src/app/home/monitoring/components/executions/ExecutionTable.tsx @@ -0,0 +1,169 @@ +import React from 'react'; +import { useTranslation } from 'react-i18next'; +import type { TFunction } from 'i18next'; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from '@/components/ui/table'; +import { Badge } from '@/components/ui/badge'; +import type { ExecutionRow } from '@/app/infra/entities/api/monitoring-executions'; +import { eventPatternLabel } from '@/app/home/components/event-patterns/event-pattern-groups'; +import { formatRunDuration } from '@/app/home/agents/components/processor-run-timing'; +import { formatDateTime } from '../../utils/dateUtils'; + +interface ExecutionTableProps { + rows: ExecutionRow[]; + selectedId?: string | null; + onSelect: (row: ExecutionRow) => void; +} + +const STATUS_GROUPS = [ + 'completed', + 'failed', + 'running', + 'queued', + 'cancelled', + 'ignored', +]; + +export function executionStatusLabel(row: ExecutionRow, t: TFunction): string { + if (STATUS_GROUPS.includes(row.status_group)) { + return t(`monitoring.execution.status.${row.status_group}`); + } + return row.status; +} + +export function executionStatusVariant( + statusGroup: string, +): 'default' | 'secondary' | 'destructive' | 'outline' { + if (statusGroup === 'failed') return 'destructive'; + if (statusGroup === 'ignored' || statusGroup === 'queued') return 'secondary'; + return 'outline'; +} + +const KIND_GROUPS = ['agent', 'pipeline', 'processor']; + +export function executionKindLabel(row: ExecutionRow, t: TFunction): string { + if (KIND_GROUPS.includes(row.target_kind)) { + 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') { + return eventPatternLabel(row.title, t); + } + const title = row.title ?? ''; + return title.length > 48 ? `${title.slice(0, 48)}…` : title; +} + +// The processor the row ran: the Agent or Pipeline that handled it. A run whose +// Agent was deleted still names the Runner that executed it. +export function executionProcessorLabel(row: ExecutionRow): string { + return row.target_name || row.runner_id || row.target_id || '—'; +} + +export default function ExecutionTable({ + rows, + selectedId, + onSelect, +}: ExecutionTableProps) { + const { t } = useTranslation(); + + return ( + // The page hands this table a fixed-height region: the table's own scroll + // container takes that height so the header stays sticky while rows scroll + // under it. +
+ + + + {t('monitoring.execution.columns.time')} + {t('monitoring.execution.columns.target')} + {t('monitoring.execution.columns.status')} + + {t('monitoring.execution.columns.duration')} + + + {t('monitoring.execution.columns.tokens')} + + + + + {rows.map((row) => { + const selected = selectedId === row.id; + return ( + onSelect(row)} + onKeyDown={( + event: React.KeyboardEvent, + ) => { + if (event.key === 'Enter' || event.key === ' ') { + event.preventDefault(); + onSelect(row); + } + }} + > + + {row.created_at_ms == null + ? '—' + : formatDateTime(new Date(row.created_at_ms))} + + +
+
+ + {executionKindLabel(row, t)} + + + {executionProcessorLabel(row)} + +
+ + {t('monitoring.execution.triggeredBy', { + trigger: executionTargetLabel(row, t), + })} + +
+ + +
+ + {executionStatusLabel(row, t)} + + {row.debug && ( + + {t('monitoring.execution.debug')} + + )} +
+ + + {row.duration_ms == null + ? '—' + : formatRunDuration(row.duration_ms)} + + + {row.usage?.total_tokens != null + ? row.usage.total_tokens.toLocaleString() + : '—'} + + + ); + })} + +
+
+ ); +} diff --git a/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx b/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx index 6cf204e14..2470451f6 100644 --- a/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx +++ b/web/src/app/home/monitoring/components/filters/MonitoringFilters.tsx @@ -7,8 +7,13 @@ import { SelectTrigger, SelectValue, } from '@/components/ui/select'; +import { Label } from '@/components/ui/label'; import { backendClient } from '@/app/infra/http'; import { TimeRangeOption } from '../../types/monitoring'; +import type { + ExecutionModeFilter, + ExecutionSourceFilter, +} from '@/app/infra/entities/api/monitoring-executions'; interface MonitoringFiltersProps { selectedBots: string[]; @@ -17,8 +22,23 @@ 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; + onStatusGroupChange?: (value: string) => void; } +const STATUS_GROUP_OPTIONS = [ + 'completed', + 'failed', + 'running', + 'queued', + 'cancelled', + 'ignored', +]; + interface Bot { uuid: string; name: string; @@ -36,6 +56,12 @@ export default function MonitoringFilters({ onBotsChange, onPipelinesChange, onTimeRangeChange, + source, + onSourceChange, + mode, + onModeChange, + statusGroup, + onStatusGroupChange, }: MonitoringFiltersProps) { const { t } = useTranslation(); const [bots, setBots] = useState([]); @@ -110,11 +136,98 @@ export default function MonitoringFilters({ return (
+ {/* Source Filter */} + {onSourceChange && ( +
+ + +
+ )} + + {/* Mode Filter */} + {onModeChange && ( +
+ + +
+ )} + + {/* Status Filter */} + {onStatusGroupChange && ( +
+ + +
+ )} + {/* Bot Filter */}
- -