diff --git a/.github/workflows/test-migrations.yml b/.github/workflows/test-migrations.yml index 086a29915..811939454 100644 --- a/.github/workflows/test-migrations.yml +++ b/.github/workflows/test-migrations.yml @@ -10,12 +10,16 @@ on: - 'src/langbot/pkg/persistence/**' - 'src/langbot/pkg/entity/persistence/**' - 'tests/integration/persistence/**' + - 'tests/unit_tests/api/service/test_monitoring_sessions.py' + - '.github/workflows/test-migrations.yml' pull_request: types: [opened, synchronize, reopened, ready_for_review] paths: - 'src/langbot/pkg/persistence/**' - 'src/langbot/pkg/entity/persistence/**' - 'tests/integration/persistence/**' + - 'tests/unit_tests/api/service/test_monitoring_sessions.py' + - '.github/workflows/test-migrations.yml' jobs: test-migrations-sqlite: @@ -80,6 +84,8 @@ jobs: run: >- uv run pytest tests/integration/persistence/test_migrations_postgres.py + tests/integration/persistence/test_monitoring_postgres.py + tests/unit_tests/api/service/test_monitoring_sessions.py::test_postgres_upgrade_rls_and_concurrent_bot_counts tests/integration/persistence/test_pgvector_postgres.py tests/integration/persistence/test_release_migration_postgres.py tests/integration/persistence/test_plugin_identity_migration.py diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 9a468ab3f..854ac0ec0 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/src/langbot/pkg/api/http/controller/groups/monitoring.py @@ -5,6 +5,7 @@ import quart from ...authz import Permission from ...context import RequestContext +from ...service.monitoring_traffic import get_traffic_series from .. import group @@ -377,6 +378,14 @@ class MonitoringRouterGroup(group.RouterGroup): return self.success( data={ + 'traffic': await get_traffic_series( + self.ap, + request_context, + bot_ids=bot_ids or None, + pipeline_ids=pipeline_ids or None, + start_time=start_time, + end_time=end_time, + ), 'overview': overview, 'messages': messages, 'llmCalls': llm_calls, @@ -405,6 +414,7 @@ class MonitoringRouterGroup(group.RouterGroup): session_id, start_time=start_time, end_time=end_time, + bot_id=quart.request.args.get('botId'), ) # Always return success with the analysis data diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 474a5c1d5..c90cec6d1 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -29,6 +29,19 @@ _DEFAULT_CLEANUP_BATCHES_PER_TABLE = 4 _HARD_MAX_CLEANUP_BATCHES_PER_TABLE = 100 +def _normalize_user_id(value: str | int | None) -> str | None: + """Convert numeric platform IDs before binding a VARCHAR with asyncpg. + + Opaque string IDs (including whitespace and leading zeros) and missing + IDs must remain unchanged. Do not silently stringify unsupported objects. + """ + if value is None or isinstance(value, str): + return value + if isinstance(value, int) and not isinstance(value, bool): + return str(value) + raise TypeError('user_id must be a string, integer, or None') + + def _workspace_transaction(method): """Run an explicit service entrypoint in one Workspace transaction.""" @@ -281,19 +294,21 @@ class MonitoringService: for _batch_number in range(max_batches): async def delete_batch() -> tuple[int, int]: + key_columns = list(model_cls.__table__.primary_key.columns) select_result = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(pk_column) + sqlalchemy.select(*key_columns) .where(model_cls.workspace_uuid == workspace_uuid, ts_column < cutoff) .limit(batch_size) ) - pk_values = list(select_result.scalars().all()) + pk_values = [tuple(row) for row in select_result.all()] if not pk_values: return 0, 0 delete_result = await self.ap.persistence_mgr.execute_async( sqlalchemy.delete(model_cls).where( model_cls.workspace_uuid == workspace_uuid, - pk_column.in_(pk_values), + sqlalchemy.tuple_(*key_columns).in_(pk_values), + ts_column < cutoff, ) ) return len(pk_values), int(delete_result.rowcount or 0) @@ -415,7 +430,7 @@ class MonitoringService: status: str = 'success', level: str = 'info', platform: str | None = None, - user_id: str | None = None, + user_id: str | int | None = None, user_name: str | None = None, runner_name: str | None = None, variables: str | None = None, @@ -437,7 +452,7 @@ class MonitoringService: 'status': status, 'level': level, 'platform': platform, - 'user_id': user_id, + 'user_id': _normalize_user_id(user_id), 'user_name': user_name, 'runner_name': runner_name, 'variables': variables, @@ -610,7 +625,7 @@ class MonitoringService: pipeline_id: str, pipeline_name: str, platform: str | None = None, - user_id: str | None = None, + user_id: str | int | None = None, user_name: str | None = None, ) -> None: """Record a new session""" @@ -622,17 +637,29 @@ class MonitoringService: 'bot_name': bot_name, 'pipeline_id': pipeline_id, 'pipeline_name': pipeline_name, - 'message_count': 0, + 'message_count': 1, 'start_time': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), 'last_activity': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), 'is_active': True, 'platform': platform, - 'user_id': user_id, + 'user_id': _normalize_user_id(user_id), 'user_name': user_name, } + model = persistence_monitoring.MonitoringSession + dialect = self.ap.persistence_mgr.get_db_engine().dialect.name + insert = postgresql_dialect.insert if dialect == 'postgresql' else sqlite_dialect.insert + statement = insert(model).values(session_data) await self.ap.persistence_mgr.execute_async( - sqlalchemy.insert(persistence_monitoring.MonitoringSession).values(session_data) + statement.on_conflict_do_update( + index_elements=['workspace_uuid', 'bot_id', 'session_id'], + set_={ + 'message_count': model.message_count + 1, + 'last_activity': statement.excluded.last_activity, + 'pipeline_id': statement.excluded.pipeline_id, + 'pipeline_name': statement.excluded.pipeline_name, + }, + ) ) @_workspace_transaction @@ -642,6 +669,7 @@ class MonitoringService: session_id: str, pipeline_id: str | None = None, pipeline_name: str | None = None, + bot_id: str | None = None, ) -> bool: """Update session last activity time and increment message count. @@ -651,6 +679,9 @@ class MonitoringService: True if session was found and updated, False if session doesn't exist. """ workspace_uuid = self._require_write_context(context) + bot_id = bot_id if bot_id is not None else context.bot_uuid + if not bot_id: + raise ValueError('Session activity requires a bot_id') update_values = { 'last_activity': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), 'message_count': persistence_monitoring.MonitoringSession.message_count + 1, @@ -667,6 +698,7 @@ class MonitoringService: .where( persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringSession.session_id == session_id, + persistence_monitoring.MonitoringSession.bot_id == bot_id, ) .values(update_values) ) @@ -769,13 +801,13 @@ class MonitoringService: message_conditions.append(persistence_monitoring.MonitoringMessage.timestamp >= start_time) llm_conditions.append(persistence_monitoring.MonitoringLLMCall.timestamp >= start_time) embedding_conditions.append(persistence_monitoring.MonitoringEmbeddingCall.timestamp >= start_time) - session_conditions.append(persistence_monitoring.MonitoringSession.start_time >= start_time) + session_conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time) if end_time: message_conditions.append(persistence_monitoring.MonitoringMessage.timestamp <= end_time) llm_conditions.append(persistence_monitoring.MonitoringLLMCall.timestamp <= end_time) embedding_conditions.append(persistence_monitoring.MonitoringEmbeddingCall.timestamp <= end_time) - session_conditions.append(persistence_monitoring.MonitoringSession.start_time <= end_time) + session_conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time) # Total messages message_query = sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringMessage.id)) @@ -1272,9 +1304,9 @@ class MonitoringService: if pipeline_ids: conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids)) if start_time: - conditions.append(persistence_monitoring.MonitoringSession.start_time >= start_time) + conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time) if end_time: - conditions.append(persistence_monitoring.MonitoringSession.start_time <= end_time) + conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time) if user_query and user_query.strip(): user_pattern = f'%{user_query.strip()}%' conditions.append( @@ -1376,6 +1408,7 @@ class MonitoringService: session_id: str, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, + bot_id: str | None = None, ) -> dict: """Get bounded session details with full statistics computed in SQL.""" workspace_uuid = require_workspace_uuid(context) @@ -1385,8 +1418,13 @@ class MonitoringService: persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringSession.session_id == session_id, ) - session_result = await self.ap.persistence_mgr.execute_async(session_query) - session_row = session_result.first() + if bot_id is not None: + session_query = session_query.where(persistence_monitoring.MonitoringSession.bot_id == bot_id) + session_result = await self.ap.persistence_mgr.execute_async(session_query.limit(2)) + session_rows = session_result.all() + if len(session_rows) > 1: + return {'session_id': session_id, 'found': False, 'ambiguous': True} + session_row = session_rows[0] if session_rows else None if not session_row: return { @@ -1395,6 +1433,7 @@ class MonitoringService: } session = session_row[0] if isinstance(session_row, tuple) else session_row + bot_id = session.bot_id message_stats_result = await self.ap.persistence_mgr.execute_async( sqlalchemy.select( @@ -1422,6 +1461,7 @@ class MonitoringService: ).where( persistence_monitoring.MonitoringMessage.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringMessage.session_id == session_id, + persistence_monitoring.MonitoringMessage.bot_id == bot_id, ) ) message_stats = message_stats_result.one() @@ -1460,6 +1500,7 @@ class MonitoringService: ).where( persistence_monitoring.MonitoringLLMCall.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringLLMCall.session_id == session_id, + persistence_monitoring.MonitoringLLMCall.bot_id == bot_id, ) ) llm_stats = llm_stats_result.one() @@ -1486,12 +1527,14 @@ class MonitoringService: ).where( persistence_monitoring.MonitoringToolCall.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringToolCall.session_id == session_id, + persistence_monitoring.MonitoringToolCall.bot_id == bot_id, ) ) tool_stats = tool_stats_result.one() tool_conditions = [ persistence_monitoring.MonitoringToolCall.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringToolCall.session_id == session_id, + persistence_monitoring.MonitoringToolCall.bot_id == bot_id, ] if start_time is not None: tool_conditions.append(persistence_monitoring.MonitoringToolCall.timestamp >= start_time) @@ -1520,6 +1563,7 @@ class MonitoringService: .where( persistence_monitoring.MonitoringError.workspace_uuid == workspace_uuid, persistence_monitoring.MonitoringError.session_id == session_id, + persistence_monitoring.MonitoringError.bot_id == bot_id, ) .order_by(persistence_monitoring.MonitoringError.timestamp.desc()) .limit(detail_limit + 1) @@ -2004,9 +2048,9 @@ class MonitoringService: if pipeline_ids: conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids)) if start_time: - conditions.append(persistence_monitoring.MonitoringSession.start_time >= start_time) + conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time) if end_time: - conditions.append(persistence_monitoring.MonitoringSession.start_time <= end_time) + conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time) query = sqlalchemy.select(persistence_monitoring.MonitoringSession).order_by( persistence_monitoring.MonitoringSession.last_activity.desc() @@ -2040,6 +2084,7 @@ class MonitoringService: # ========== Feedback Methods ========== + @_workspace_transaction async def record_feedback( self, context: ExecutionContext, @@ -2054,7 +2099,7 @@ class MonitoringService: session_id: str | None = None, message_id: str | None = None, stream_id: str | None = None, - user_id: str | None = None, + user_id: str | int | None = None, platform: str | None = None, ) -> str | None: """Record user feedback (like/dislike) from AI Bot conversation. @@ -2110,7 +2155,7 @@ class MonitoringService: 'session_id': session_id, 'message_id': message_id, 'stream_id': stream_id, - 'user_id': user_id, + 'user_id': _normalize_user_id(user_id), 'platform': platform, } dialect_name = self.ap.persistence_mgr.get_db_engine().dialect.name diff --git a/src/langbot/pkg/api/http/service/monitoring_traffic.py b/src/langbot/pkg/api/http/service/monitoring_traffic.py new file mode 100644 index 000000000..702d34e01 --- /dev/null +++ b/src/langbot/pkg/api/http/service/monitoring_traffic.py @@ -0,0 +1,83 @@ +"""Bounded traffic aggregation, independent of record-list pagination.""" + +from __future__ import annotations + +import datetime +import typing + +import sqlalchemy + +from ....entity.persistence.monitoring import MonitoringLLMCall, MonitoringMessage +from .tenant import TenantContext, require_workspace_uuid + +if typing.TYPE_CHECKING: + from ....core.app import Application + +MAX_TRAFFIC_POINTS = 1000 + + +async def get_traffic_series( + ap: Application, + context: TenantContext, + *, + 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: + """Count all matching records in UTC buckets, returning at most 1000 points.""" + workspace_uuid = require_workspace_uuid(context) + 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' + points: dict[datetime.datetime, dict[str, int]] = {} + truncated = False + for model, field in ((MonitoringMessage, 'messages'), (MonitoringLLMCall, 'llm_calls')): + timestamp = model.timestamp + if postgres: + time_bucket = sqlalchemy.func.date_trunc(bucket, timestamp) + else: + 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] + if bot_ids: + conditions.append(model.bot_id.in_(bot_ids)) + if pipeline_ids: + conditions.append(model.pipeline_id.in_(pipeline_ids)) + if start_time is not None: + conditions.append(timestamp >= start_time) + if end_time is not None: + conditions.append(timestamp <= end_time) + statement = ( + sqlalchemy.select(time_bucket.label('bucket'), sqlalchemy.func.count(model.id).label('count')) + .where(*conditions) + .group_by(time_bucket) + .order_by(time_bucket) + .limit(MAX_TRAFFIC_POINTS + 1) + ) + result = await ap.persistence_mgr.execute_async(statement) + rows = result.all() + truncated = truncated or len(rows) > MAX_TRAFFIC_POINTS + for timestamp_value, count in rows[:MAX_TRAFFIC_POINTS]: + key = ( + datetime.datetime.fromisoformat(timestamp_value) + if isinstance(timestamp_value, str) + else timestamp_value + ) + points.setdefault(key, {'messages': 0, 'llm_calls': 0})[field] = int(count) + + def floor(value: datetime.datetime) -> datetime.datetime: + return value.replace(minute=0, second=0, microsecond=0, **({'hour': 0} if bucket == 'day' else {})) + + first = floor(start_time) if start_time is not None else min(points, default=None) + last = floor(end_time) if end_time is not None else max(points, default=None) + series = [] + if first is not None and last is not None: + cursor = first + while cursor <= last and len(series) < MAX_TRAFFIC_POINTS: + series.append( + {'timestamp': cursor.isoformat() + 'Z', **points.get(cursor, {'messages': 0, 'llm_calls': 0})} + ) + cursor += step + truncated = truncated or cursor <= last + return {'bucket': bucket, 'points': series, 'truncated': truncated} diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index 35ebe161a..1d2e8ed83 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -111,8 +111,8 @@ class MonitoringSession(Base): sqlalchemy.ForeignKey('workspaces.uuid', ondelete='CASCADE'), primary_key=True, ) + bot_id = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True, index=True) session_id = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True) - bot_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True) bot_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) pipeline_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True) pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) diff --git a/src/langbot/pkg/persistence/alembic/versions/0023_bot_scoped_sessions.py b/src/langbot/pkg/persistence/alembic/versions/0023_bot_scoped_sessions.py new file mode 100644 index 000000000..4e8a03bfe --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0023_bot_scoped_sessions.py @@ -0,0 +1,104 @@ +"""Scope monitoring sessions by bot without changing runtime session IDs. + +Revision ID: 0023_bot_scoped_sessions +Revises: 0022_codex_credentials +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql, sqlite + +revision = '0023_bot_scoped_sessions' +down_revision = '0022_codex_credentials' +branch_labels = None +depends_on = None + +_TABLE = 'monitoring_sessions' +_KEY = ['workspace_uuid', 'bot_id', 'session_id'] + + +def upgrade() -> None: + conn = op.get_bind() + inspector = sa.inspect(conn) + if _TABLE not in inspector.get_table_names(): + return + pk = inspector.get_pk_constraint(_TABLE) + if pk['constrained_columns'] == _KEY: + return + # PostgreSQL alters in place, retaining indexes, grants, policies and RLS. + # SQLite batch reflection retains all existing indexes and foreign keys. + with op.batch_alter_table(_TABLE, naming_convention={'pk': 'pk_%(table_name)s'}) as batch: + batch.drop_constraint(pk['name'] or f'pk_{_TABLE}', type_='primary') + batch.create_primary_key(f'pk_{_TABLE}', _KEY) + + metadata = sa.MetaData() + sessions = sa.Table(_TABLE, metadata, autoload_with=conn) + messages = sa.Table('monitoring_messages', metadata, autoload_with=conn) + m = messages.c + collisions = ( + sa.select(m.workspace_uuid, m.session_id) + .group_by(m.workspace_uuid, m.session_id) + .having(sa.func.count(sa.distinct(m.bot_id)) > 1) + .subquery() + ) + partition = [m.workspace_uuid, m.bot_id, m.session_id] + # Repair only demonstrable collisions. Retention may have removed earlier + # evidence; these summaries describe surviving messages, never invented text. + ranked = ( + sa.select( + *[m[name] for name in _KEY], + m.bot_name, + m.pipeline_id, + m.pipeline_name, + m.platform, + m.user_id, + m.user_name, + sa.func.sum(sa.case((sa.or_(m.role == 'user', m.role.is_(None)), 1), else_=0)) + .over(partition_by=partition) + .label('message_count'), + sa.func.min(m.timestamp).over(partition_by=partition).label('start_time'), + sa.func.max(m.timestamp).over(partition_by=partition).label('last_activity'), + sa.func.row_number().over(partition_by=partition, order_by=[m.timestamp.desc(), m.id.desc()]).label('rank'), + ) + .join( + collisions, + sa.and_(m.workspace_uuid == collisions.c.workspace_uuid, m.session_id == collisions.c.session_id), + ) + .subquery() + ) + columns = _KEY + [ + 'bot_name', + 'pipeline_id', + 'pipeline_name', + 'platform', + 'user_id', + 'user_name', + 'message_count', + 'start_time', + 'last_activity', + 'is_active', + ] + select = sa.select(*[ranked.c[name] for name in columns[:-1]], sa.literal(True)).where(ranked.c.rank == 1) + insert = postgresql.insert if conn.dialect.name == 'postgresql' else sqlite.insert + statement = insert(sessions).from_select(columns, select) + conn.execute( + statement.on_conflict_do_update( + index_elements=_KEY, + set_={name: statement.excluded[name] for name in columns if name not in _KEY and name != 'is_active'}, + ) + ) + + +def downgrade() -> None: + conn = op.get_bind() + if _TABLE not in sa.inspect(conn).get_table_names(): + return + collisions = conn.execute( + sa.text('SELECT 1 FROM monitoring_sessions GROUP BY workspace_uuid, session_id HAVING COUNT(*) > 1 LIMIT 1') + ).first() + if collisions: + raise RuntimeError('Cannot downgrade bot-scoped sessions without losing colliding bot records') + pk = sa.inspect(conn).get_pk_constraint(_TABLE) + with op.batch_alter_table(_TABLE, naming_convention={'pk': 'pk_%(table_name)s'}) as batch: + batch.drop_constraint(pk['name'] or f'pk_{_TABLE}', type_='primary') + batch.create_primary_key(f'pk_{_TABLE}', ['workspace_uuid', 'session_id']) diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index b361b1067..d28397f60 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/src/langbot/pkg/persistence/tenant_uow.py @@ -207,6 +207,8 @@ _SYNC_PROXY_CAPABILITY: contextvars.ContextVar[_ScopedSessionGuardState | None] _ALLOWED_SCOPED_BUILTIN_FUNCTION_TYPES = { 'coalesce': sqlalchemy.sql.functions.coalesce, 'count': sqlalchemy.sql.functions.count, + 'min': sqlalchemy.sql.functions.min, + 'max': sqlalchemy.sql.functions.max, 'now': sqlalchemy.sql.functions.now, 'sum': sqlalchemy.sql.functions.sum, } diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index 0728b0c83..1bab4bda4 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -79,6 +79,7 @@ class MonitoringHelper: session_updated = await ap.monitoring_service.update_session_activity( get_query_execution_context(query), session_id, + bot_id=bot_id, pipeline_id=pipeline_id, pipeline_name=pipeline_name, ) diff --git a/tests/integration/api/test_monitoring.py b/tests/integration/api/test_monitoring.py index cf4608e65..0422d25e2 100644 --- a/tests/integration/api/test_monitoring.py +++ b/tests/integration/api/test_monitoring.py @@ -9,7 +9,7 @@ Run: uv run pytest tests/integration/api/test_monitoring.py -q from __future__ import annotations import pytest -from unittest.mock import MagicMock, AsyncMock, Mock +from unittest.mock import MagicMock, AsyncMock, Mock, patch from types import SimpleNamespace from tests.factories import FakeApp @@ -280,13 +280,20 @@ class TestMonitoringAllDataEndpoint: @pytest.mark.asyncio async def test_get_all_data_success(self, quart_test_client): """GET /api/v1/monitoring/data returns all data.""" - response = await quart_test_client.get( - '/api/v1/monitoring/data', headers={'Authorization': 'Bearer test_token'} - ) + traffic = {'series': [], 'truncated': False} + with patch( + 'langbot.pkg.api.http.controller.groups.monitoring.get_traffic_series', + new=AsyncMock(return_value=traffic), + ) as get_traffic: + response = await quart_test_client.get( + '/api/v1/monitoring/data', headers={'Authorization': 'Bearer test_token'} + ) + get_traffic.assert_awaited_once() assert response.status_code == 200 data = await response.get_json() assert 'overview' in data['data'] + assert data['data']['traffic'] == traffic @pytest.mark.usefixtures('mock_circular_import_chain') diff --git a/tests/integration/persistence/resource_migration_support.py b/tests/integration/persistence/resource_migration_support.py index afb05ae15..c82742acf 100644 --- a/tests/integration/persistence/resource_migration_support.py +++ b/tests/integration/persistence/resource_migration_support.py @@ -193,6 +193,22 @@ async def create_legacy_resource_schema(engine, *, instance_uuid: str) -> None: sa.Column('message_id', sa.String(255), nullable=True), ) + # Include historical monitoring columns consumed by later migrations. + for table_name in ('monitoring_messages', 'monitoring_sessions'): + table = monitoring_tables[table_name] + for name, value in (('bot_name', 'bot'), ('pipeline_id', 'pipeline-1'), ('pipeline_name', 'pipeline')): + table.append_column(sa.Column(name, sa.String(255), nullable=False, default=value)) + for name in ('platform', 'user_id', 'user_name'): + table.append_column(sa.Column(name, sa.String(255))) + if table_name == 'monitoring_messages': + table.append_column(sa.Column('bot_id', sa.String(255), nullable=False, default='bot-1')) + table.append_column(sa.Column('role', sa.String(50))) + else: + table.append_column(sa.Column('message_count', sa.Integer, nullable=False, default=1)) + table.append_column( + sa.Column('start_time', sa.DateTime, nullable=False, default=datetime.datetime(2026, 1, 1)) + ) + now = datetime.datetime(2026, 1, 1) async with engine.begin() as conn: await conn.run_sync(metadata.create_all) diff --git a/tests/integration/persistence/test_migrations.py b/tests/integration/persistence/test_migrations.py index 1c1674a4b..9304ade87 100644 --- a/tests/integration/persistence/test_migrations.py +++ b/tests/integration/persistence/test_migrations.py @@ -17,6 +17,7 @@ from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.persistence import mgr as persistence_mgr # noqa: F401 -- register all ORM tables from langbot.pkg.persistence.alembic_runner import ( run_alembic_downgrade, run_alembic_upgrade, @@ -108,7 +109,6 @@ class TestSQLiteMigrationUpgrade: await run_alembic_upgrade(sqlite_engine, 'head') assert await get_alembic_current(sqlite_engine) == _get_script_head() - assert _get_script_head() == '0022_codex_credentials' @pytest.mark.asyncio async def test_upgrade_from_reasoning_config_head_to_merged_head(self, sqlite_engine): @@ -119,7 +119,7 @@ class TestSQLiteMigrationUpgrade: await run_alembic_stamp(sqlite_engine, '0018_llm_reasoning_config') await run_alembic_upgrade(sqlite_engine, 'head') - assert await get_alembic_current(sqlite_engine) == '0022_codex_credentials' + assert await get_alembic_current(sqlite_engine) == _get_script_head() @pytest.mark.asyncio async def test_upgrade_from_baseline_to_head(self, sqlite_engine): @@ -280,6 +280,15 @@ class TestSQLiteMigrationUpgrade: class TestSQLiteMigrationFreshDatabase: """Tests for fresh database workflow.""" + @pytest.mark.asyncio + async def test_bot_scoped_sessions_skips_absent_table(self, sqlite_engine): + """A partial schema needs no session key migration in either direction.""" + await run_alembic_stamp(sqlite_engine, '0022_codex_credentials') + await run_alembic_upgrade(sqlite_engine, '0023_bot_scoped_sessions') + assert await get_alembic_current(sqlite_engine) == '0023_bot_scoped_sessions' + await run_alembic_downgrade(sqlite_engine, '0022_codex_credentials') + assert await get_alembic_current(sqlite_engine) == '0022_codex_credentials' + @pytest.mark.asyncio async def test_fresh_db_upgrade_from_scratch(self, tmp_path): """ diff --git a/tests/integration/persistence/test_monitoring_postgres.py b/tests/integration/persistence/test_monitoring_postgres.py new file mode 100644 index 000000000..ebfee7a7e --- /dev/null +++ b/tests/integration/persistence/test_monitoring_postgres.py @@ -0,0 +1,354 @@ +"""Monitoring regressions through asyncpg, Cloud UoW guards, and migrated RLS. + +TEST_POSTGRES_URL must identify a disposable PostgreSQL/pgvector test server +with permission to create databases and roles. Each run owns a fresh database; +no existing tables are dropped. Without that URL these tests are skipped. +""" + +from __future__ import annotations + +import logging +import os +import uuid +from types import SimpleNamespace + +import pytest +import pytest_asyncio +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.api.http.context import ExecutionContext +from langbot.pkg.api.http.service.monitoring import MonitoringService +from langbot.pkg.entity.persistence import monitoring as models +from langbot.pkg.entity.persistence.workspace import Workspace +from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode +from langbot.pkg.persistence.tenant_uow import TenantScopeRequiredError +from langbot.pkg.pipeline.monitoring_helper import MonitoringHelper + +pytestmark = [pytest.mark.integration, pytest.mark.slow, pytest.mark.asyncio(loop_scope='module')] + +WORKSPACE_A = '00000000-0000-0000-0000-00000000000a' +WORKSPACE_B = '00000000-0000-0000-0000-00000000000b' +RESOURCE = dict(bot_id='same-bot', bot_name='Bot', pipeline_id='same-pipeline', pipeline_name='Pipeline') +MONITORING_TABLES = tuple( + table for table in models.MonitoringMessage.metadata.sorted_tables if table.name.startswith('monitoring_') +) + + +def _context(workspace_uuid): + return ExecutionContext( + instance_uuid='monitoring-postgres-test', + workspace_uuid=workspace_uuid, + placement_generation=1, + bot_uuid=RESOURCE['bot_id'], + pipeline_uuid=RESOURCE['pipeline_id'], + ) + + +def _application(url): + return SimpleNamespace( + instance_config=SimpleNamespace( + data={ + 'database': { + 'use': 'postgresql', + 'postgresql': { + 'host': url.host, + 'port': url.port, + 'user': url.username, + 'password': url.password, + 'database': url.database, + }, + } + } + ), + logger=logging.getLogger('monitoring-postgres-test'), + ) + + +@pytest_asyncio.fixture(scope='module', loop_scope='module') +async def cloud_database(): + url = os.environ.get('TEST_POSTGRES_URL') + if not url: + pytest.skip('TEST_POSTGRES_URL not set') + admin_url = sa.engine.make_url(url) + admin = create_async_engine(admin_url, isolation_level='AUTOCOMMIT') + suffix = uuid.uuid4().hex[:12] + database_name = f'lb_monitoring_{suffix}' + runtime_role = f'lb_monitoring_{suffix}' + password = f'Test{uuid.uuid4().hex}' + database_created = role_created = False + release_manager = runtime_manager = None + quote = admin.dialect.identifier_preparer.quote + from langbot.pkg.persistence import mgr as mgr_module + from langbot.pkg.persistence.databases.postgresql import PostgreSQLDatabaseManager + from langbot.pkg.utils import constants + + with pytest.MonkeyPatch.context() as patch: + patch.setattr(mgr_module.database, 'preregistered_managers', [PostgreSQLDatabaseManager]) + patch.setattr(constants, 'instance_id', 'monitoring-postgres-test') + try: + async with admin.connect() as conn: + await conn.execute(sa.text(f'CREATE DATABASE {quote(database_name)}')) + database_created = True + await conn.execute( + sa.text(f"CREATE ROLE {quote(runtime_role)} LOGIN NOSUPERUSER NOBYPASSRLS PASSWORD '{password}'") + ) + role_created = True + release_app = _application(admin_url.set(database=database_name)) + release_manager = PersistenceManager(release_app, mode=PersistenceMode.RELEASE_MIGRATION) + release_app.persistence_mgr = release_manager + await release_manager.initialize() + async with release_manager.get_db_engine().begin() as conn: + for workspace in (WORKSPACE_A, WORKSPACE_B): + await conn.execute( + sa.insert(Workspace).values( + uuid=workspace, + instance_uuid='monitoring-postgres-test', + name=workspace, + slug=workspace, + source='cloud_projection', + ) + ) + tables = release_manager._runtime_business_table_names() + quoted_tables = ', '.join(f'public.{quote(name)}' for name in tables) + await conn.execute( + sa.text(f'GRANT CONNECT ON DATABASE {quote(database_name)} TO {quote(runtime_role)}') + ) + await conn.execute(sa.text(f'GRANT USAGE ON SCHEMA public TO {quote(runtime_role)}')) + await conn.execute( + sa.text(f'GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE {quoted_tables} TO {quote(runtime_role)}') + ) + await conn.execute(sa.text(f'GRANT SELECT ON public.alembic_version TO {quote(runtime_role)}')) + sequences = await release_manager._runtime_business_sequence_names(conn, tables) + if sequences: + names = ', '.join(f'public.{quote(name)}' for name in sequences) + await conn.execute(sa.text(f'GRANT USAGE, SELECT ON SEQUENCE {names} TO {quote(runtime_role)}')) + runtime_app = _application(admin_url.set(database=database_name, username=runtime_role, password=password)) + runtime_manager = PersistenceManager(runtime_app, mode=PersistenceMode.CLOUD_RUNTIME) + runtime_app.persistence_mgr = runtime_manager + await runtime_manager.initialize() + runtime_app.monitoring_service = MonitoringService(runtime_app) + yield runtime_app, release_manager.get_db_engine() + finally: + if runtime_manager is not None: + await runtime_manager.shutdown() + if release_manager is not None: + await release_manager.shutdown() + async with admin.connect() as conn: + if database_created: + await conn.execute(sa.text(f'DROP DATABASE {quote(database_name)} WITH (FORCE)')) + if role_created: + await conn.execute(sa.text(f'DROP ROLE {quote(runtime_role)}')) + await admin.dispose() + + +@pytest_asyncio.fixture(loop_scope='module') +async def service(cloud_database): + application, admin = cloud_database + async with admin.begin() as conn: + for table in MONITORING_TABLES: + await conn.execute(sa.delete(table)) + application.instance_config.data.pop('monitoring', None) + return application.monitoring_service + + +async def _read(service, method, context, *args, **kwargs): + # HTTP auth binds a tenant scope; exercise that same guard for service reads. + async with service.ap.persistence_mgr.tenant_scope(context.workspace_uuid): + return await getattr(service, method)(context, *args, **kwargs) + + +def _query(context, sender_id): + return SimpleNamespace( + _execution_context=context, + launcher_type='person', + launcher_id='same-user', + sender_id=sender_id, + message_chain=SimpleNamespace(model_dump=lambda: [{'type': 'Plain', 'text': 'hello'}]), + resp_message_chain=[SimpleNamespace(model_dump=lambda: [{'type': 'Plain', 'text': 'reply'}])], + message_event=SimpleNamespace(sender=SimpleNamespace(nickname='Alice')), + variables={'public': 'value', '_private': 'hidden'}, + ) + + +@pytest.mark.parametrize('user_id', [123456789, -100123456789, 0, None, '', '00123', ' opaque用户 ']) +@pytest.mark.parametrize('record_type', ['message', 'session', 'feedback']) +async def test_optional_user_ids_round_trip_through_asyncpg(service, user_id, record_type): + context = _context(WORKSPACE_A) + expected = str(user_id) if isinstance(user_id, int) else user_id + if record_type == 'message': + record_id = await service.record_message( + context, + **RESOURCE, + message_content='hello', + session_id='same-session', + user_id=user_id, + ) + details = await _read(service, 'get_message_details', context, record_id) + assert details['message']['user_id'] == expected + elif record_type == 'session': + await service.record_session_start(context, **RESOURCE, session_id='same-session', user_id=user_id) + rows, total = await _read(service, 'get_sessions', context) + assert total == 1 + assert rows[0]['user_id'] == expected + else: + await service.record_feedback(context, feedback_id='same-feedback', feedback_type=1, user_id=user_id) + rows, total = await _read(service, 'get_feedback_list', context) + assert total == 1 + assert rows[0]['user_id'] == expected + + +@pytest.mark.parametrize('user_id', [123456789, -100123456789]) +async def test_query_lifecycle_persists_messages_session_and_llm_link(service, user_id, caplog): + context = _context(WORKSPACE_A) + query = _query(context, user_id) + message_id = await MonitoringHelper.record_query_start(service.ap, query, **RESOURCE) + assert message_id, caplog.text + await MonitoringHelper.record_llm_call( + service.ap, + query, + **RESOURCE, + model_name='model', + input_tokens=3, + output_tokens=5, + duration_ms=25, + message_id=message_id, + ) + await MonitoringHelper.record_query_success(service.ap, message_id, query) + await MonitoringHelper.record_query_response(service.ap, query, **RESOURCE) + rows, total = await _read(service, 'get_messages', context) + assert total == 2 + assert {row['role'] for row in rows} == {'user', 'assistant'} + assert {row['user_id'] for row in rows} == {str(user_id)} + details = await _read(service, 'get_message_details', context, message_id) + assert details['message']['status'] == 'success' + assert details['message']['variables'] == '{"public": "value"}' + assert details['llm_calls'][0]['message_id'] == message_id + assert details['llm_stats']['total_tokens'] == 8 + sessions, total = await _read(service, 'get_sessions', context) + assert total == 1 + assert sessions[0]['session_id'] == 'person_same-user' + assert sessions[0]['user_id'] == str(user_id) + assert not [record for record in caplog.records if record.levelno >= logging.ERROR] + + +@pytest.mark.parametrize('user_id', [123, -123]) +async def test_query_error_persists_error_message_and_linked_log(service, user_id, caplog): + context = _context(WORKSPACE_A) + message_id = await MonitoringHelper.record_query_error( + service.ap, + _query(context, user_id), + **RESOURCE, + error=ValueError('failed query'), + ) + assert message_id, caplog.text + details = await _read(service, 'get_message_details', context, message_id) + assert details['message']['user_id'] == str(user_id) + assert details['message']['status'] == 'error' + assert details['errors'][0]['message_id'] == message_id + assert details['errors'][0]['error_type'] == 'ValueError' + + +@pytest.mark.parametrize('user_id', [True, 1.5, b'123', ['123']]) +@pytest.mark.parametrize('record_type', ['message', 'session', 'feedback']) +async def test_unsupported_user_ids_fail_at_the_write_boundary(service, user_id, record_type): + context = _context(WORKSPACE_A) + with pytest.raises(TypeError, match='user_id must be a string, integer, or None'): + if record_type == 'message': + await service.record_message( + context, + **RESOURCE, + message_content='hello', + session_id='session', + user_id=user_id, + ) + elif record_type == 'session': + await service.record_session_start(context, **RESOURCE, session_id='session', user_id=user_id) + else: + await service.record_feedback(context, feedback_id='feedback', feedback_type=1, user_id=user_id) + async with service.ap.persistence_mgr.tenant_scope(WORKSPACE_A): + for model in (models.MonitoringMessage, models.MonitoringSession, models.MonitoringFeedback): + count = await service.ap.persistence_mgr.execute_async(sa.select(sa.func.count()).select_from(model)) + assert count.scalar_one() == 0 + + +async def test_session_analysis_aggregates_under_cloud_sql_guard(service): + context = _context(WORKSPACE_A) + await service.record_session_start(context, **RESOURCE, session_id='same-session') + await service.record_message(context, **RESOURCE, session_id='same-session', message_content='hello') + result = await _read(service, 'get_session_analysis', context, 'same-session') + assert result['found'] is True + assert result['message_stats'] == {'total': 1, 'success': 1, 'error': 0, 'pending': 0} + assert result['llm_stats']['total_calls'] == 0 + assert result['tool_stats']['total_calls'] == 0 + assert result['session_duration_seconds'] == 0 + + +async def test_rls_is_enforced_without_application_workspace_predicates(service, cloud_database): + _, admin = cloud_database + for workspace in (WORKSPACE_A, WORKSPACE_B): + await service.record_message( + _context(workspace), **RESOURCE, session_id='same-session', message_content=workspace + ) + async with admin.connect() as conn: + states = ( + await conn.execute( + sa.text( + 'SELECT relname, relrowsecurity, relforcerowsecurity FROM pg_class ' + "WHERE relname LIKE 'monitoring_%' AND relkind = 'r'" + ) + ) + ).all() + assert len(states) == len(MONITORING_TABLES) + assert all(enabled and forced for _, enabled, forced in states) + engine = service.ap.persistence_mgr.get_db_engine() + async with engine.connect() as conn: + role = ( + await conn.execute(sa.text('SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user')) + ).one() + assert role == (False, False) + assert (await conn.execute(sa.select(models.MonitoringMessage.id))).all() == [] + for workspace in (WORKSPACE_A, WORKSPACE_B): + async with service.ap.persistence_mgr.tenant_uow(workspace): + rows = ( + await service.ap.persistence_mgr.execute_async(sa.select(models.MonitoringMessage.workspace_uuid)) + ).all() + assert rows == [(workspace,)] + with pytest.raises(TenantScopeRequiredError): + await service.ap.persistence_mgr.execute_async(sa.select(models.MonitoringMessage.id)) + + +async def test_traffic_series_aggregates_all_rows_under_cloud_rls(service): + import datetime + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + context = _context(WORKSPACE_A) + for workspace, count in ((WORKSPACE_A, 61), (WORKSPACE_B, 2)): + async with service.ap.persistence_mgr.tenant_scope(workspace): + await service.ap.persistence_mgr.execute_async( + sa.insert(models.MonitoringMessage).values( + [ + dict( + workspace_uuid=workspace, + id=f'{workspace}-m-{i}', + **RESOURCE, + session_id='shared', + message_content='test', + status='success', + level='info', + timestamp=datetime.datetime(2026, 9, 11, 1, 30), + ) + for i in range(count) + ] + ) + ) + async with service.ap.persistence_mgr.tenant_uow(WORKSPACE_A): + result = await get_traffic_series( + service.ap, + context, + bot_ids=[RESOURCE['bot_id']], + start_time=datetime.datetime(2026, 9, 11), + end_time=datetime.datetime(2026, 9, 12), + ) + assert result['truncated'] is False + assert sum(point['messages'] for point in result['points']) == 61 diff --git a/tests/integration/persistence/test_resource_tenancy_migration.py b/tests/integration/persistence/test_resource_tenancy_migration.py index 8bc3797a1..932a78070 100644 --- a/tests/integration/persistence/test_resource_tenancy_migration.py +++ b/tests/integration/persistence/test_resource_tenancy_migration.py @@ -142,7 +142,7 @@ async def test_legacy_sqlite_resources_are_backfilled_and_contracted(tmp_path): assert pk_columns == { 'binary_storages': ('workspace_uuid', 'unique_key'), 'plugin_settings': ('workspace_uuid', 'plugin_author', 'plugin_name'), - 'monitoring_sessions': ('workspace_uuid', 'session_id'), + 'monitoring_sessions': ('workspace_uuid', 'bot_id', 'session_id'), } pipeline_run_foreign_keys = await _inspect( @@ -237,8 +237,10 @@ async def test_sqlite_scoped_keys_allow_cross_workspace_but_reject_same_workspac await conn.execute( sa.text( 'INSERT INTO monitoring_sessions ' - '(workspace_uuid, session_id, bot_id, last_activity, is_active) ' - "VALUES (:workspace_uuid, 'session-1', 'bot-2', CURRENT_TIMESTAMP, 1)" + '(workspace_uuid, session_id, bot_id, bot_name, pipeline_id, pipeline_name, ' + 'start_time, last_activity, message_count, is_active) ' + "VALUES (:workspace_uuid, 'session-1', 'bot-2', 'bot', 'pipeline-2', 'pipeline', " + 'CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, 1, 1)' ), {'workspace_uuid': second_workspace_uuid}, ) diff --git a/tests/unit_tests/api/service/test_monitoring_identifiers.py b/tests/unit_tests/api/service/test_monitoring_identifiers.py new file mode 100644 index 000000000..c8a3b7792 --- /dev/null +++ b/tests/unit_tests/api/service/test_monitoring_identifiers.py @@ -0,0 +1,19 @@ +"""Identifier normalization must not rely on SQLite's permissive codecs.""" + +import pytest + +from langbot.pkg.api.http.service import monitoring + + +@pytest.mark.parametrize( + ('value', 'expected'), + [(None, None), ('', ''), ('00123', '00123'), (' 用户 ', ' 用户 '), (123, '123'), (-123, '-123'), (0, '0')], +) +def test_normalize_user_id_preserves_opaque_strings(value, expected): + assert monitoring._normalize_user_id(value) == expected + + +@pytest.mark.parametrize('value', [True, False, 1.5, b'123', ['123'], {'id': 123}]) +def test_normalize_user_id_rejects_unsupported_types(value): + with pytest.raises(TypeError, match='user_id must be a string, integer, or None'): + monitoring._normalize_user_id(value) diff --git a/tests/unit_tests/api/service/test_monitoring_sessions.py b/tests/unit_tests/api/service/test_monitoring_sessions.py new file mode 100644 index 000000000..7451c9124 --- /dev/null +++ b/tests/unit_tests/api/service/test_monitoring_sessions.py @@ -0,0 +1,220 @@ +"""Bot-scoped session regressions exercised against real SQL databases.""" + +import datetime as dt +import logging +from types import SimpleNamespace + +import pytest +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.api.http.context import ExecutionContext +from langbot.pkg.api.http.service.monitoring import MonitoringService +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence import monitoring as models +from langbot.pkg.persistence.mgr import PersistenceManager +from langbot.pkg.pipeline.monitoring_helper import MonitoringHelper + +from tests.integration.persistence.test_monitoring_postgres import cloud_database # noqa: F401 + +pytestmark = pytest.mark.asyncio + + +@pytest.mark.asyncio(loop_scope='module') +async def test_postgres_upgrade_rls_and_concurrent_bot_counts(cloud_database): # noqa: F811 + import asyncio + import importlib + from alembic.migration import MigrationContext + from alembic.operations import Operations + from tests.integration.persistence.test_monitoring_postgres import WORKSPACE_A, _context, _read + + ap, admin = cloud_database + service = ap.monitoring_service + ctx = _context(WORKSPACE_A) + await service.record_session_start(ctx, session_id='person_42', **resource('a')) + for bot in ['a', 'b']: + await service.record_message(ctx, session_id='person_42', message_content=bot, **resource(bot)) + async with admin.begin() as conn: + + def migrate(connection): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0023_bot_scoped_sessions') + with Operations.context(MigrationContext.configure(connection)): + migration.downgrade() + migration.upgrade() + rls = connection.execute( + sa.text("SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE relname='monitoring_sessions'") + ).one() + assert tuple(rls) == (True, True) + assert ( + connection.execute( + sa.text("SELECT count(*) FROM pg_policies WHERE tablename='monitoring_sessions'") + ).scalar_one() + == 1 + ) + + await conn.run_sync(migrate) + rows, total = await _read(service, 'get_sessions', ctx) + assert total == 2 + assert {r['bot_id']: r['message_count'] for r in rows} == {'a': 1, 'b': 1} + await asyncio.gather(*[service.record_session_start(ctx, session_id='race', **resource('a')) for _ in range(10)]) + result = await _read(service, 'get_session_analysis', ctx, 'race', bot_id='a') + assert result['session']['message_count'] == 10 + assert not (await _read(service, 'get_session_analysis', ctx, 'person_42'))['found'] + assert (await _read(service, 'get_session_analysis', ctx, 'person_42', bot_id='b'))['message_stats']['total'] == 1 + + +async def test_migration_reconstructs_collisions_and_preserves_indexes(service): + import importlib + from alembic.migration import MigrationContext + from alembic.operations import Operations + + engine = service.ap.persistence_mgr.get_db_engine() + async with engine.begin() as conn: + + def upgrade(connection): + table = models.MonitoringSession.__table__ + table.drop(connection) + metadata = sa.MetaData() + legacy = table.to_metadata(metadata) + legacy.primary_key._columns.remove(legacy.c.bot_id) + legacy.c.bot_id.primary_key = False + # Resolve the unchanged Workspace FK in copied metadata. + Base.metadata.tables['workspaces'].to_metadata(metadata) + legacy.create(connection) + now = dt.datetime(2026, 1, 1) + connection.execute( + sa.insert(legacy).values( + workspace_uuid='workspace', + session_id='person_42', + **resource('a'), + message_count=99, + start_time=now, + last_activity=now, + is_active=True, + ) + ) + for bot in ['a', 'b']: + connection.execute( + sa.insert(models.MonitoringMessage).values( + id=bot, + workspace_uuid='workspace', + timestamp=now, + **resource(bot), + session_id='person_42', + message_content=bot, + role='user', + status='success', + level='info', + ) + ) + indexes = {i['name'] for i in sa.inspect(connection).get_indexes('monitoring_sessions')} + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0023_bot_scoped_sessions') + with Operations.context(MigrationContext.configure(connection)): + migration.upgrade() + migration.upgrade() # Fresh/already-upgraded schema is safe. + assert sa.inspect(connection).get_pk_constraint('monitoring_sessions')['constrained_columns'] == [ + 'workspace_uuid', + 'bot_id', + 'session_id', + ] + assert indexes <= {i['name'] for i in sa.inspect(connection).get_indexes('monitoring_sessions')} + + await conn.run_sync(upgrade) + rows, total = await service.get_sessions(context()) + assert total == 2 + assert {r['bot_id']: r['message_count'] for r in rows} == {'a': 1, 'b': 1} + assert {r['pipeline_id'] for r in rows} == {'a', 'b'} + + +def context(bot=None): + return ExecutionContext(instance_uuid='test', workspace_uuid='workspace', placement_generation=1, bot_uuid=bot) + + +def resource(bot): + return dict(bot_id=bot, bot_name=bot, pipeline_id=bot, pipeline_name=bot) + + +@pytest.fixture +async def service(): + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + class Persistence: + serialize_model = PersistenceManager.serialize_model + + def get_db_engine(self): + return engine + + async def execute_async(self, stmt): + async with engine.begin() as conn: + return await conn.execute(stmt) + + ap = SimpleNamespace(persistence_mgr=Persistence(), logger=logging.getLogger(__name__)) + ap.monitoring_service = MonitoringService(ap) + yield ap.monitoring_service + await engine.dispose() + + +async def test_helper_first_message_count_and_two_bot_isolation(service): + for bot in ['a', 'b', 'a']: + query = SimpleNamespace( + _execution_context=context(bot), + launcher_type='person', + launcher_id=42, + sender_id=42, + message_chain=SimpleNamespace(model_dump=lambda: []), + ) + assert await MonitoringHelper.record_query_start(service.ap, query, **resource(bot)) + rows, total = await service.get_sessions(context()) + assert total == 2 + assert {r['bot_id']: r['message_count'] for r in rows} == {'a': 2, 'b': 1} + assert {r['pipeline_id'] for r in rows} == {'a', 'b'} + assert {r['session_id'] for r in rows} == {'person_42'} + + +async def test_analysis_fails_closed_and_scopes_statistics(service): + for bot in ['a', 'b']: + await service.record_session_start(context(bot), session_id='person_42', **resource(bot)) + await service.record_message(context(bot), session_id='person_42', message_content=bot, **resource(bot)) + assert (await service.get_session_analysis(context(), 'person_42'))['found'] is False + result = await service.get_session_analysis(context(), 'person_42', bot_id='b') + assert result['message_stats']['total'] == 1 + assert result['session']['bot_id'] == 'b' + + +async def test_activity_requires_bot_and_upsert_counts_racing_first_queries(service): + for _ in range(2): + await service.record_session_start(context('a'), session_id='person_42', **resource('a')) + with pytest.raises(ValueError, match='bot'): + await service.update_session_activity(context(), 'person_42') + assert await service.update_session_activity(context('a'), 'person_42') + assert not await service.update_session_activity(context('b'), 'person_42') + rows, _ = await service.get_sessions(context()) + assert rows[0]['message_count'] == 3 + + +async def test_old_active_sessions_are_listed_exported_and_not_cleaned(service): + for bot in ['a', 'b']: + await service.record_session_start(context(bot), session_id='person_42', **resource(bot)) + old = dt.datetime(2000, 1, 1) + await service.ap.persistence_mgr.execute_async(sa.update(models.MonitoringSession).values(start_time=old)) + await service.ap.persistence_mgr.execute_async( + sa.update(models.MonitoringSession).where(models.MonitoringSession.bot_id == 'a').values(last_activity=old) + ) + since = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None) - dt.timedelta(days=1) + rows, total = await service.get_sessions(context(), start_time=since) + assert total == 1 and rows[0]['bot_id'] == 'b' + assert len(await service.export_sessions(context(), start_time=since)) == 1 + count = await service._delete_expired_in_batches( + context(), + models.MonitoringSession, + models.MonitoringSession.last_activity, + models.MonitoringSession.session_id, + since, + 1, + 2, + ) + assert count == 1 + rows, total = await service.get_sessions(context()) + assert total == 1 and rows[0]['bot_id'] == 'b' diff --git a/tests/unit_tests/api/service/test_monitoring_traffic.py b/tests/unit_tests/api/service/test_monitoring_traffic.py new file mode 100644 index 000000000..527c8bea8 --- /dev/null +++ b/tests/unit_tests/api/service/test_monitoring_traffic.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import datetime +from types import SimpleNamespace + +import pytest +import sqlalchemy +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.api.http.context import ExecutionContext +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence.monitoring import MonitoringLLMCall, MonitoringMessage +from langbot.pkg.entity.persistence.workspace import Workspace + +pytestmark = pytest.mark.asyncio + +A = '00000000-0000-0000-0000-00000000000a' +B = '00000000-0000-0000-0000-00000000000b' +START = datetime.datetime(2026, 1, 1) + + +@pytest.fixture +async def traffic_app(): + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + await connection.execute( + sqlalchemy.insert(Workspace), + [ + {'uuid': wid, 'instance_uuid': 'instance', 'name': wid, 'slug': wid, 'source': 'cloud_projection'} + for wid in (A, B) + ], + ) + for wid, bot, count in [(A, 'bot-a', 60), (A, 'bot-b', 7), (B, 'bot-a', 9)]: + common = { + 'workspace_uuid': wid, + 'timestamp': START, + 'bot_id': bot, + 'bot_name': bot, + 'pipeline_id': 'pipeline', + 'pipeline_name': 'Pipeline', + 'session_id': 'person_42', + 'status': 'success', + } + await connection.execute( + sqlalchemy.insert(MonitoringMessage), + [ + dict(common, id=f'{wid}-{bot}-{i}', message_content='test fixture', level='info', role='user') + for i in range(count) + ], + ) + await connection.execute( + sqlalchemy.insert(MonitoringLLMCall), + [ + dict( + common, + id=f'{wid}-{bot}-{i}', + model_name='fixture-model', + input_tokens=1, + output_tokens=1, + total_tokens=2, + duration=1, + ) + for i in range(count) + ], + ) + + class Persistence: + def get_db_engine(self): + return engine + + async def execute_async(self, statement): + async with engine.connect() as connection: + return await connection.execute(statement) + + yield SimpleNamespace(persistence_mgr=Persistence()) + await engine.dispose() + + +async def test_traffic_counts_all_rows_not_just_latest_page(traffic_app): + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + context = ExecutionContext(instance_uuid='instance', workspace_uuid=A, placement_generation=1) + result = await get_traffic_series( + traffic_app, context, bot_ids=['bot-a'], start_time=START, end_time=START + datetime.timedelta(hours=2) + ) + assert result['bucket'] == 'hour' + assert result['truncated'] is False + assert sum(point['messages'] for point in result['points']) == 60 + assert sum(point['llm_calls'] for point in result['points']) == 60 + assert len(result['points']) == 3 + assert result['points'][1]['messages'] == result['points'][1]['llm_calls'] == 0 + assert result['points'][0]['timestamp'] == '2026-01-01T00:00:00Z' + + +async def test_traffic_workspace_pipeline_and_empty_filters(traffic_app): + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + context = ExecutionContext(instance_uuid='instance', workspace_uuid=B, placement_generation=1) + kwargs = dict(start_time=START, end_time=START + datetime.timedelta(hours=2)) + result = await get_traffic_series(traffic_app, context, **kwargs) + assert sum(point['messages'] for point in result['points']) == 9 + empty = await get_traffic_series(traffic_app, context, pipeline_ids=['missing'], **kwargs) + assert sum(point['messages'] for point in empty['points']) == 0 + assert sum(point['llm_calls'] for point in empty['points']) == 0 + + +async def test_traffic_bounds_large_ranges_and_marks_truncation(traffic_app): + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + context = ExecutionContext(instance_uuid='instance', workspace_uuid=A, placement_generation=1) + result = await get_traffic_series( + traffic_app, context, start_time=START, end_time=START + datetime.timedelta(days=5000) + ) + assert result['bucket'] == 'day' + assert result['truncated'] is True + assert len(result['points']) == 1000 + + +async def test_traffic_fails_closed_without_workspace(traffic_app): + from langbot.pkg.api.http.authz import WorkspaceRequiredError + from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series + + with pytest.raises(WorkspaceRequiredError): + await get_traffic_series(traffic_app, None) diff --git a/tests/unit_tests/persistence/test_tenant_uow.py b/tests/unit_tests/persistence/test_tenant_uow.py index 8fd5e79f8..0c374f201 100644 --- a/tests/unit_tests/persistence/test_tenant_uow.py +++ b/tests/unit_tests/persistence/test_tenant_uow.py @@ -958,6 +958,8 @@ async def test_scoped_session_rejects_raw_or_unapproved_sql( [ sa.select(sa.literal('set_config(')), sa.select(sa.func.count()), + sa.select(sa.func.min(sa.column('timestamp'))), + sa.select(sa.func.max(sa.column('timestamp'))), sa.select(sa.func.coalesce(sa.func.sum(sa.literal(1)), sa.literal(0))), sa.select( sa.func.now(), diff --git a/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx b/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx index 5b31e2e9b..2505c6e6b 100644 --- a/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx +++ b/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx @@ -157,6 +157,9 @@ const BotSessionMonitor = forwardRef< const [messagePage, setMessagePage] = useState(0); const [loadingSessions, setLoadingSessions] = useState(false); const [loadingMessages, setLoadingMessages] = useState(false); + const [sessionError, setSessionError] = useState(false); + const [messageError, setMessageError] = useState(false); + const [analysisError, setAnalysisError] = useState(false); const [copiedUserId, setCopiedUserId] = useState(false); const [feedbackMap, setFeedbackMap] = useState< Record @@ -236,6 +239,8 @@ const BotSessionMonitor = forwardRef< const loadSessions = useCallback(async () => { const requestId = ++sessionRequestIdRef.current; setLoadingSessions(true); + setSessionError(false); + setSessions([]); try { const response = await httpClient.getBotSessions(botId, { limit: SESSION_PAGE_SIZE, @@ -254,6 +259,7 @@ const BotSessionMonitor = forwardRef< } catch (error) { if (requestId === sessionRequestIdRef.current) { console.error('Failed to load sessions:', error); + setSessionError(true); } } finally { if (requestId === sessionRequestIdRef.current) { @@ -274,12 +280,18 @@ const BotSessionMonitor = forwardRef< async (sessionId: string, page: number) => { const requestId = ++messageRequestIdRef.current; setLoadingMessages(true); + setMessageError(false); + setAnalysisError(false); + setMessages([]); + setToolCalls([]); + setFeedbackMap({}); setExpandedToolCallIds({}); try { const messagesRes = await httpClient.getSessionMessages( sessionId, MESSAGE_PAGE_SIZE, page * MESSAGE_PAGE_SIZE, + botId, ); if (requestId !== messageRequestIdRef.current) return; const sorted = (messagesRes.messages ?? []).sort( @@ -290,22 +302,19 @@ const BotSessionMonitor = forwardRef< setMessageTotal(messagesRes.total ?? 0); try { - const analysisParams = new URLSearchParams(); - if (sorted.length > 0) { - analysisParams.set('startTime', sorted[0].timestamp); - analysisParams.set('endTime', sorted[sorted.length - 1].timestamp); - } - const analysisRes = await httpClient.get<{ + const analysisRes = await httpClient.getSessionAnalysis<{ tool_calls?: SessionToolCall[]; - }>( - `/api/v1/monitoring/sessions/${encodeURIComponent(sessionId)}/analysis?${analysisParams.toString()}`, - ); + }>(sessionId, botId, { + startTime: sorted[0]?.timestamp, + endTime: sorted[sorted.length - 1]?.timestamp, + }); if (requestId !== messageRequestIdRef.current) return; setToolCalls(analysisRes?.tool_calls ?? []); } catch (analysisError) { if (requestId !== messageRequestIdRef.current) return; console.error('Failed to load session tool calls:', analysisError); setToolCalls([]); + setAnalysisError(true); } // Collect user message IDs for feedback matching @@ -337,6 +346,7 @@ const BotSessionMonitor = forwardRef< } catch (error) { if (requestId === messageRequestIdRef.current) { console.error('Failed to load session messages:', error); + setMessageError(true); } } finally { if (requestId === messageRequestIdRef.current) { @@ -349,6 +359,9 @@ const BotSessionMonitor = forwardRef< useEffect(() => { loadSessions(); + return () => { + sessionRequestIdRef.current += 1; + }; }, [loadSessions]); useEffect(() => { @@ -362,12 +375,17 @@ const BotSessionMonitor = forwardRef< } else { messageRequestIdRef.current += 1; setLoadingMessages(false); + setMessageError(false); + setAnalysisError(false); setMessages([]); setMessageTotal(0); setToolCalls([]); setExpandedToolCallIds({}); setFeedbackMap({}); } + return () => { + messageRequestIdRef.current += 1; + }; }, [selectedSessionId, messagePage, loadMessages]); useEffect(() => { @@ -728,6 +746,20 @@ const BotSessionMonitor = forwardRef<
{t('bots.sessionMonitor.loading')}
+ ) : sessionError ? ( +
+

{t('monitoring.loadError')}

+ +
) : sessions.length === 0 ? (
{t('bots.sessionMonitor.noSessions')} @@ -898,10 +930,46 @@ const BotSessionMonitor = forwardRef< className="flex-1 px-4 py-4 overflow-y-auto min-h-0" >
+ {analysisError && !loadingMessages && ( +
+

+ {t('monitoring.toolCalls.title')}:{' '} + {t('monitoring.loadError')} +

+ +
+ )} {loadingMessages ? (
{t('bots.sessionMonitor.loading')}
+ ) : messageError ? ( +
+

{t('monitoring.loadError')}

+ +
) : timelineItems.length === 0 ? (
{t('bots.sessionMonitor.noMessages')} diff --git a/web/src/app/home/monitoring/components/overview-cards/OverviewCards.tsx b/web/src/app/home/monitoring/components/overview-cards/OverviewCards.tsx index f37d1523f..cb1e9c1d8 100644 --- a/web/src/app/home/monitoring/components/overview-cards/OverviewCards.tsx +++ b/web/src/app/home/monitoring/components/overview-cards/OverviewCards.tsx @@ -4,24 +4,18 @@ import { MessageSquare, Sparkles, Check, Users } from 'lucide-react'; import MetricCard from './MetricCard'; import SystemStatusCard from './SystemStatusCards'; import TrafficChart from './TrafficChart'; -import { - OverviewMetrics, - MonitoringMessage, - LLMCall, -} from '../../types/monitoring'; +import { OverviewMetrics, MonitoringData } from '../../types/monitoring'; interface OverviewCardsProps { metrics: OverviewMetrics | null; - messages?: MonitoringMessage[]; - llmCalls?: LLMCall[]; + traffic?: MonitoringData['traffic']; loading?: boolean; refreshKey?: number; } export default function OverviewCards({ metrics, - messages = [], - llmCalls = [], + traffic, loading, refreshKey, }: OverviewCardsProps) { @@ -100,7 +94,7 @@ export default function OverviewCards({
{/* Traffic Chart */} - +
); } diff --git a/web/src/app/home/monitoring/components/overview-cards/TrafficChart.tsx b/web/src/app/home/monitoring/components/overview-cards/TrafficChart.tsx index 674c140e2..fd825299a 100644 --- a/web/src/app/home/monitoring/components/overview-cards/TrafficChart.tsx +++ b/web/src/app/home/monitoring/components/overview-cards/TrafficChart.tsx @@ -11,119 +11,33 @@ import { ResponsiveContainer, Legend, } from 'recharts'; -import { MonitoringMessage, LLMCall } from '../../types/monitoring'; +import { MonitoringData } from '../../types/monitoring'; interface TrafficChartProps { - messages: MonitoringMessage[]; - llmCalls: LLMCall[]; + traffic?: MonitoringData['traffic']; loading?: boolean; } -interface ChartDataPoint { - time: string; - timestamp: number; - messages: number; - llmCalls: number; -} - -export default function TrafficChart({ - messages, - llmCalls, - loading, -}: TrafficChartProps) { +export default function TrafficChart({ traffic, loading }: TrafficChartProps) { const { t } = useTranslation(); - - const chartData = useMemo(() => { - const safeMessages = Array.isArray(messages) ? messages : []; - const safeLlmCalls = Array.isArray(llmCalls) ? llmCalls : []; - if (!safeMessages.length && !safeLlmCalls.length) { - return []; - } - - // Combine all timestamps and find the range - const allTimestamps = [ - ...safeMessages.map((m) => m.timestamp.getTime()), - ...safeLlmCalls.map((c) => c.timestamp.getTime()), - ]; - - if (allTimestamps.length === 0) return []; - - const minTime = Math.min(...allTimestamps); - const maxTime = Math.max(...allTimestamps); - const timeRange = maxTime - minTime; - - // Determine bucket size based on time range - let bucketSize: number; - let formatTime: (date: Date) => string; - - if (timeRange <= 60 * 60 * 1000) { - // <= 1 hour: 5-minute buckets - bucketSize = 5 * 60 * 1000; - formatTime = (date) => - date.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); - } else if (timeRange <= 6 * 60 * 60 * 1000) { - // <= 6 hours: 15-minute buckets - bucketSize = 15 * 60 * 1000; - formatTime = (date) => - date.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); - } else if (timeRange <= 24 * 60 * 60 * 1000) { - // <= 24 hours: 1-hour buckets - bucketSize = 60 * 60 * 1000; - formatTime = (date) => - date.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); - } else if (timeRange <= 7 * 24 * 60 * 60 * 1000) { - // <= 7 days: 4-hour buckets - bucketSize = 4 * 60 * 60 * 1000; - formatTime = (date) => - `${date.toLocaleDateString([], { - month: 'short', - day: 'numeric', - })} ${date.toLocaleTimeString([], { hour: '2-digit' })}`; - } else { - // > 7 days: 1-day buckets - bucketSize = 24 * 60 * 60 * 1000; - formatTime = (date) => - date.toLocaleDateString([], { month: 'short', day: 'numeric' }); - } - - // Create buckets - const buckets: Map = new Map(); - const startBucket = Math.floor(minTime / bucketSize) * bucketSize; - const endBucket = Math.ceil(maxTime / bucketSize) * bucketSize; - - for (let bucket = startBucket; bucket <= endBucket; bucket += bucketSize) { - buckets.set(bucket, { - time: formatTime(new Date(bucket)), - timestamp: bucket, - messages: 0, - llmCalls: 0, - }); - } - - // Count messages per bucket - safeMessages.forEach((msg) => { - const bucket = - Math.floor(msg.timestamp.getTime() / bucketSize) * bucketSize; - const point = buckets.get(bucket); - if (point) { - point.messages++; - } - }); - - // Count LLM calls per bucket - safeLlmCalls.forEach((call) => { - const bucket = - Math.floor(call.timestamp.getTime() / bucketSize) * bucketSize; - const point = buckets.get(bucket); - if (point) { - point.llmCalls++; - } - }); - - return Array.from(buckets.values()).sort( - (a, b) => a.timestamp - b.timestamp, - ); - }, [messages, llmCalls]); + const chartData = useMemo( + () => + (traffic?.points ?? []).map((point) => ({ + ...point, + time: point.timestamp.toLocaleString( + [], + traffic?.bucket === 'day' + ? { month: 'short', day: 'numeric' } + : { + month: 'short', + day: 'numeric', + hour: '2-digit', + minute: '2-digit', + }, + ), + })), + [traffic], + ); if (loading) { return ( @@ -150,7 +64,13 @@ export default function TrafficChart({
-
{t('monitoring.trafficChart.noData')}
+
+ {t( + traffic + ? 'monitoring.trafficChart.noData' + : 'monitoring.trafficChart.unavailable', + )} +
); @@ -161,6 +81,11 @@ export default function TrafficChart({

{t('monitoring.trafficChart.title')}

+ {traffic?.truncated && ( +

+ {t('monitoring.trafficChart.truncated')} +

+ )}
(null); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); + const workspaceUuid = useCurrentWorkspace()?.workspace.uuid; + const requestIdRef = useRef(0); + const scope = JSON.stringify([workspaceUuid, filterState]); + const [requestScope, setRequestScope] = useState(null); // Memoize filter parameters to prevent unnecessary re-renders const selectedBotsStr = useMemo( @@ -72,6 +77,12 @@ export function useMonitoringData(filterState: FilterState) { // Fetch data based on filters const fetchData = useCallback(async () => { + const requestId = ++requestIdRef.current; + const isCurrent = () => + requestId === requestIdRef.current && + getCurrentWorkspaceSnapshot()?.workspace.uuid === workspaceUuid; + setRequestScope(scope); + setData(null); setLoading(true); setError(null); @@ -91,6 +102,7 @@ export function useMonitoringData(filterState: FilterState) { endTime, limit: 50, }); + if (!isCurrent()) return; const overview = response?.overview ?? { total_messages: 0, @@ -127,6 +139,17 @@ export function useMonitoringData(filterState: FilterState) { // Transform the response to match MonitoringData interface const transformedData: MonitoringData = { + traffic: response.traffic + ? { + bucket: response.traffic.bucket, + truncated: response.traffic.truncated, + points: response.traffic.points.map((point) => ({ + timestamp: parseUTCTimestamp(point.timestamp), + messages: point.messages, + llmCalls: point.llm_calls, + })), + } + : undefined, overview: { totalMessages: overview.total_messages, llmCalls: overview.llm_calls, @@ -396,22 +419,33 @@ export function useMonitoringData(filterState: FilterState) { setData(transformedData); } catch (err) { + if (!isCurrent()) return; setError(err as Error); console.error('Failed to fetch monitoring data:', err); } finally { - setLoading(false); + if (isCurrent()) setLoading(false); } - }, [getTimeRange, filterState.selectedBots, filterState.selectedPipelines]); + }, [ + getTimeRange, + filterState.selectedBots, + filterState.selectedPipelines, + scope, + workspaceUuid, + ]); // Fetch data when filter state changes useEffect(() => { fetchData(); + return () => { + requestIdRef.current += 1; + }; // eslint-disable-next-line react-hooks/exhaustive-deps }, [ selectedBotsStr, selectedPipelinesStr, filterState.timeRange, customDateRangeStr, + workspaceUuid, ]); // Manual refetch function @@ -420,9 +454,9 @@ export function useMonitoringData(filterState: FilterState) { }; return { - data, - loading, - error, + data: requestScope === scope ? data : null, + loading: requestScope !== scope || loading, + error: requestScope === scope ? error : null, refetch, }; } diff --git a/web/src/app/home/monitoring/page.tsx b/web/src/app/home/monitoring/page.tsx index 0d05a4c1a..a28d8978d 100644 --- a/web/src/app/home/monitoring/page.tsx +++ b/web/src/app/home/monitoring/page.tsx @@ -32,7 +32,7 @@ function MonitoringPageContent() { currentWorkspace?.permissions.includes('data.export') ?? false; const { filterState, setSelectedBots, setSelectedPipelines, setTimeRange } = useMonitoringFilters(); - const { data, loading, refetch } = useMonitoringData(filterState); + const { data, loading, error, refetch } = useMonitoringData(filterState); // Counter to force feedbackTimeRange recomputation on manual refresh const [feedbackRefreshKey, setFeedbackRefreshKey] = useState(0); @@ -174,492 +174,556 @@ function MonitoringPageContent() {
{/* Content Area */} -
- {/* Overview Section */} - + {error ? ( +
+

{t('monitoring.loadError')}

+ +
+ ) : ( +
+ {/* Overview Section */} + - {/* Tabs Section */} -
- -
- - - {t('monitoring.tabs.messages')} - - - {t('monitoring.tabs.modelCalls')} - - - {t('monitoring.tabs.tokens')} - - - {t('monitoring.tabs.feedback')} - - - {t('monitoring.tabs.errors')} - - + {/* Tabs Section */} + {!loading && data && ( +
+ {data.totalCount.messages > data.messages.length && ( +

+ {t('monitoring.partialMessages', { + shown: data.messages.length, + total: data.totalCount.messages, + })} +

+ )} + {data.totalCount.llmCalls + data.totalCount.embeddingCalls > + data.modelCalls.length && ( +

+ {t('monitoring.partialModelCalls', { + shown: data.modelCalls.length, + total: + data.totalCount.llmCalls + data.totalCount.embeddingCalls, + })} +

+ )} + {(data.totalCount.toolCalls ?? 0) > data.toolCalls.length && ( +

+ {t('monitoring.partialToolCalls', { + shown: data.toolCalls.length, + total: data.totalCount.toolCalls, + })} +

+ )} + {data.totalCount.errors > data.errors.length && ( +

+ {t('monitoring.partialErrors', { + shown: data.errors.length, + total: data.totalCount.errors, + })} +

+ )}
- - -
- {loading && ( -
- -
- )} - - {!loading && data && conversationTurns.length > 0 && ( - - )} - - {!loading && (!data || conversationTurns.length === 0) && ( -
- -
- {t('monitoring.messageList.noMessages')} -
-
- )} + )} +
+ +
+ + + {t('monitoring.tabs.messages')} + + + {t('monitoring.tabs.modelCalls')} + + + {t('monitoring.tabs.tokens')} + + + {t('monitoring.tabs.feedback')} + + + {t('monitoring.tabs.errors')} + +
- - -
- {loading && ( -
- -
- )} - - {!loading && - data && - data.modelCalls && - data.modelCalls.length > 0 && ( -
- {data.modelCalls.map((call) => ( -
-
-
- {/* Query ID - only show if messageId exists */} - {call.messageId && ( -
- - Query ID: {call.messageId} - - -
- )} -
- {/* Model Type Badge */} - - {call.modelType === 'llm' - ? t('monitoring.modelCalls.llmModel') - : t('monitoring.modelCalls.embeddingModel')} - - {/* Call Type Badge for Embedding */} - {call.modelType === 'embedding' && - call.callType && ( - - {call.callType === 'retrieve' - ? t( - 'monitoring.modelCalls.retrieveCall', - ) - : t( - 'monitoring.modelCalls.embeddingCall', - )} - - )} - {/* Status Badge */} - - {call.status} - -
- {/* Model Name */} -
- {call.modelName} -
- {/* Context Info - only for LLM calls */} - {call.modelType === 'llm' && - call.botName && - call.pipelineName && ( -
- {call.botName} → {call.pipelineName} -
- )} - {/* Token Info */} -
-
- {call.modelType === 'llm' && call.tokens && ( - <> - - {t('monitoring.llmCalls.inputTokens')}:{' '} - {call.tokens.input} - - - {t('monitoring.llmCalls.outputTokens')}:{' '} - {call.tokens.output} - - - {t('monitoring.llmCalls.totalTokens')}:{' '} - {call.tokens.total} - - - )} - {call.modelType === 'embedding' && ( - <> - - {t( - 'monitoring.embeddingCalls.promptTokens', - )} - : {call.promptTokens} - - - {t( - 'monitoring.embeddingCalls.totalTokens', - )} - : {call.totalTokens} - - - {t( - 'monitoring.embeddingCalls.inputCount', - )} - : {call.inputCount} - - - )} - - {t('monitoring.llmCalls.duration')}:{' '} - {call.duration}ms - - {call.cost && ( - - {t('monitoring.llmCalls.cost')}: $ - {call.cost.toFixed(4)} - - )} -
- {/* Knowledge Base Info for Embedding */} - {call.modelType === 'embedding' && - call.knowledgeBaseId && ( -
- {t( - 'monitoring.embeddingCalls.knowledgeBase', - )} - : {call.knowledgeBaseId} -
- )} - {/* Query Text for Embedding Retrieve */} - {call.modelType === 'embedding' && - call.queryText && ( -
- - {t( - 'monitoring.embeddingCalls.queryText', - )} - :{' '} - - - {call.queryText.length > 100 - ? call.queryText.substring(0, 100) + - '...' - : call.queryText} - -
- )} -
- {call.errorMessage && ( -
- Error: {call.errorMessage} -
- )} -
- - {call.timestamp.toLocaleString()} - -
-
- ))} + +
+ {loading && ( +
+
)} - {!loading && - (!data || - !data.modelCalls || - data.modelCalls.length === 0) && ( + {!loading && data && conversationTurns.length > 0 && ( + + )} + + {!loading && (!data || conversationTurns.length === 0) && (
- +
- {t('monitoring.modelCalls.noData')} + {t('monitoring.messageList.noMessages')}
)} -
- +
+ - - 0 - ? filterState.selectedBots - : undefined - } - pipelineIds={ - filterState.selectedPipelines.length > 0 - ? filterState.selectedPipelines - : undefined - } - startTime={feedbackTimeRange.startTime} - endTime={feedbackTimeRange.endTime} - refreshKey={feedbackRefreshKey} - /> - - - -
- {loading && ( -
- -
- )} - - {!loading && ( - <> - {/* Feedback Stats Cards */} -
- + +
+ {loading && ( +
+
+ )} - {/* Feedback List */} -

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

- - - )} -
- - - -
- {loading && ( -
- -
- )} - - {!loading && data && data.errors && data.errors.length > 0 && ( -
- {data.errors.map((error) => ( -
- {/* Error Header - Always Visible */} -
toggleErrorExpand(error.id)} - > -
-
- {/* Expand Icon */} -
- {expandedErrorId === error.id ? ( - - ) : ( - - )} -
- - {/* Error Info */} + {!loading && + data && + data.modelCalls && + data.modelCalls.length > 0 && ( +
+ {data.modelCalls.map((call) => ( +
+
- {/* Query ID */} -
- - Query ID: {error.messageId || '-'} - - {error.messageId && ( + {/* Query ID - only show if messageId exists */} + {call.messageId && ( +
+ + Query ID: {call.messageId} + - )} -
+
+ )}
- - {error.errorType} + {/* Model Type Badge */} + + {call.modelType === 'llm' + ? t('monitoring.modelCalls.llmModel') + : t( + 'monitoring.modelCalls.embeddingModel', + )} - → - - {error.botName} - - → - - {error.pipelineName} + {/* Call Type Badge for Embedding */} + {call.modelType === 'embedding' && + call.callType && ( + + {call.callType === 'retrieve' + ? t( + 'monitoring.modelCalls.retrieveCall', + ) + : t( + 'monitoring.modelCalls.embeddingCall', + )} + + )} + {/* Status Badge */} + + {call.status}
-

- {error.errorMessage} -

+ {/* Model Name */} +
+ {call.modelName} +
+ {/* Context Info - only for LLM calls */} + {call.modelType === 'llm' && + call.botName && + call.pipelineName && ( +
+ {call.botName} → {call.pipelineName} +
+ )} + {/* Token Info */} +
+
+ {call.modelType === 'llm' && + call.tokens && ( + <> + + {t( + 'monitoring.llmCalls.inputTokens', + )} + : {call.tokens.input} + + + {t( + 'monitoring.llmCalls.outputTokens', + )} + : {call.tokens.output} + + + {t( + 'monitoring.llmCalls.totalTokens', + )} + : {call.tokens.total} + + + )} + {call.modelType === 'embedding' && ( + <> + + {t( + 'monitoring.embeddingCalls.promptTokens', + )} + : {call.promptTokens} + + + {t( + 'monitoring.embeddingCalls.totalTokens', + )} + : {call.totalTokens} + + + {t( + 'monitoring.embeddingCalls.inputCount', + )} + : {call.inputCount} + + + )} + + {t('monitoring.llmCalls.duration')}:{' '} + {call.duration}ms + + {call.cost && ( + + {t('monitoring.llmCalls.cost')}: $ + {call.cost.toFixed(4)} + + )} +
+ {/* Knowledge Base Info for Embedding */} + {call.modelType === 'embedding' && + call.knowledgeBaseId && ( +
+ {t( + 'monitoring.embeddingCalls.knowledgeBase', + )} + : {call.knowledgeBaseId} +
+ )} + {/* Query Text for Embedding Retrieve */} + {call.modelType === 'embedding' && + call.queryText && ( +
+ + {t( + 'monitoring.embeddingCalls.queryText', + )} + :{' '} + + + {call.queryText.length > 100 + ? call.queryText.substring(0, 100) + + '...' + : call.queryText} + +
+ )} +
+ {call.errorMessage && ( +
+ Error: {call.errorMessage} +
+ )}
-
- - {/* Timestamp */} -
- - {error.timestamp.toLocaleString()} + + {call.timestamp.toLocaleString()}
-
+ ))} +
+ )} - {/* Expanded Details */} - {expandedErrorId === error.id && ( -
-
- {/* Error Details */} -
-

- {t('monitoring.errors.errorMessage')} -

-
- {error.errorMessage} + {!loading && + (!data || + !data.modelCalls || + data.modelCalls.length === 0) && ( +
+ +
+ {t('monitoring.modelCalls.noData')} +
+
+ )} +
+ + + + 0 + ? filterState.selectedBots + : undefined + } + pipelineIds={ + filterState.selectedPipelines.length > 0 + ? filterState.selectedPipelines + : undefined + } + startTime={feedbackTimeRange.startTime} + endTime={feedbackTimeRange.endTime} + refreshKey={feedbackRefreshKey} + /> + + + +
+ {loading && ( +
+ +
+ )} + + {!loading && ( + <> + {/* Feedback Stats Cards */} +
+ +
+ + {/* Feedback List */} +

+ {t('monitoring.feedback.feedbackList')} +

+ + + )} +
+ + + +
+ {loading && ( +
+ +
+ )} + + {!loading && + data && + data.errors && + data.errors.length > 0 && ( +
+ {data.errors.map((error) => ( +
+ {/* Error Header - Always Visible */} +
toggleErrorExpand(error.id)} + > +
+
+ {/* Expand Icon */} +
+ {expandedErrorId === error.id ? ( + + ) : ( + + )} +
+ + {/* Error Info */} +
+ {/* Query ID */} +
+ + Query ID: {error.messageId || '-'} + + {error.messageId && ( + + )} +
+
+ + {error.errorType} + + → + + {error.botName} + + → + + {error.pipelineName} + +
+

+ {error.errorMessage} +

+
+
+ + {/* Timestamp */} +
+ + {error.timestamp.toLocaleString()} +
+
- {/* Context Info */} -
-

- {t('monitoring.messageList.viewDetails')} -

-
-
-
- {t('monitoring.messageList.bot')} -
-
- {error.botName} + {/* Expanded Details */} + {expandedErrorId === error.id && ( +
+
+ {/* Error Details */} +
+

+ {t('monitoring.errors.errorMessage')} +

+
+ {error.errorMessage}
-
-
- {t('monitoring.messageList.pipeline')} -
-
- {error.pipelineName} + + {/* Context Info */} +
+

+ {t('monitoring.messageList.viewDetails')} +

+
+
+
+ {t('monitoring.messageList.bot')} +
+
+ {error.botName} +
+
+
+
+ {t('monitoring.messageList.pipeline')} +
+
+ {error.pipelineName} +
+
+ {error.sessionId && ( +
+
+ {t('monitoring.sessions.sessionId')} +
+
+ {error.sessionId} +
+
+ )}
- {error.sessionId && ( -
-
- {t('monitoring.sessions.sessionId')} -
-
- {error.sessionId} -
+ + {/* Stack Trace */} + {error.stackTrace && ( +
+

+ {t('monitoring.errors.stackTrace')} +

+ 
+                                        {error.stackTrace}
+
)}
- - {/* Stack Trace */} - {error.stackTrace && ( -
-

- {t('monitoring.errors.stackTrace')} -

- 
-                                    {error.stackTrace}
-
-
- )} -
+ )}
- )} + ))}
- ))} -
- )} + )} - {!loading && - (!data || !data.errors || data.errors.length === 0) && ( -
- -
- {t('monitoring.errors.noErrors')} + {!loading && + (!data || !data.errors || data.errors.length === 0) && ( +
+ +
+ {t('monitoring.errors.noErrors')} +
-
- )} -
- - + )} +
+ + +
-
+ )}
); } diff --git a/web/src/app/home/monitoring/types/monitoring.ts b/web/src/app/home/monitoring/types/monitoring.ts index 06e6cc677..8a1f4dc53 100644 --- a/web/src/app/home/monitoring/types/monitoring.ts +++ b/web/src/app/home/monitoring/types/monitoring.ts @@ -217,6 +217,11 @@ export interface FeedbackStats { } export interface MonitoringData { + traffic?: { + bucket: 'hour' | 'day'; + points: Array<{ timestamp: Date; messages: number; llmCalls: number }>; + truncated: boolean; + }; overview: OverviewMetrics; messages: MonitoringMessage[]; llmCalls: LLMCall[]; diff --git a/web/src/app/home/monitoring/utils/conversationTurns.ts b/web/src/app/home/monitoring/utils/conversationTurns.ts index 658540900..b650855c0 100644 --- a/web/src/app/home/monitoring/utils/conversationTurns.ts +++ b/web/src/app/home/monitoring/utils/conversationTurns.ts @@ -155,17 +155,18 @@ function findTurnBySessionTime( sessionTurns: Map, sessionId: string | undefined, timestamp: Date, + botId: string, ): ConversationTurn | undefined { if (!sessionId) { return undefined; } - const turns = sessionTurns.get(sessionId); + const turns = sessionTurns.get(JSON.stringify([botId, sessionId])); if (!turns?.length) { return undefined; } - let nearest = turns[0]; + let nearest: ConversationTurn | undefined; const targetTime = timestamp.getTime(); for (const turn of turns) { @@ -203,15 +204,16 @@ export function buildConversationTurns( for (const message of visibleMessages) { const role = normalizeRole(message, activityMessageIds); - const previousTurn = lastTurnBySession.get(message.sessionId); + const sessionKey = JSON.stringify([message.botId, message.sessionId]); + const previousTurn = lastTurnBySession.get(sessionKey); const shouldStartTurn = role === 'user' || !previousTurn; const turn = shouldStartTurn ? createTurn(message) : previousTurn; if (shouldStartTurn) { - const turns = sessionTurns.get(message.sessionId) ?? []; + const turns = sessionTurns.get(sessionKey) ?? []; turns.push(turn); - sessionTurns.set(message.sessionId, turns); - lastTurnBySession.set(message.sessionId, turn); + sessionTurns.set(sessionKey, turns); + lastTurnBySession.set(sessionKey, turn); } addMessageToTurn(turn, message, role); @@ -221,9 +223,14 @@ export function buildConversationTurns( const allTurns = Array.from(sessionTurns.values()).flat(); for (const call of llmCalls) { - const turn = - (call.messageId ? messageIdToTurn.get(call.messageId) : undefined) ?? - findTurnBySessionTime(sessionTurns, call.sessionId, call.timestamp); + const turn = call.messageId + ? messageIdToTurn.get(call.messageId) + : findTurnBySessionTime( + sessionTurns, + call.sessionId, + call.timestamp, + call.botId, + ); if (!turn) { continue; @@ -243,9 +250,14 @@ export function buildConversationTurns( } for (const call of toolCalls) { - const turn = - (call.messageId ? messageIdToTurn.get(call.messageId) : undefined) ?? - findTurnBySessionTime(sessionTurns, call.sessionId, call.timestamp); + const turn = call.messageId + ? messageIdToTurn.get(call.messageId) + : findTurnBySessionTime( + sessionTurns, + call.sessionId, + call.timestamp, + call.botId, + ); if (!turn) { continue; @@ -262,9 +274,14 @@ export function buildConversationTurns( } for (const error of errors) { - const turn = - (error.messageId ? messageIdToTurn.get(error.messageId) : undefined) ?? - findTurnBySessionTime(sessionTurns, error.sessionId, error.timestamp); + const turn = error.messageId + ? messageIdToTurn.get(error.messageId) + : findTurnBySessionTime( + sessionTurns, + error.sessionId, + error.timestamp, + error.botId, + ); if (!turn) { continue; diff --git a/web/src/app/infra/http/BackendClient.ts b/web/src/app/infra/http/BackendClient.ts index 4e742259d..426c9ab83 100644 --- a/web/src/app/infra/http/BackendClient.ts +++ b/web/src/app/infra/http/BackendClient.ts @@ -563,10 +563,24 @@ export class BackendClient extends BaseHttpClient { return this.get(`/api/v1/monitoring/sessions?${queryParams.toString()}`); } + public getSessionAnalysis( + sessionId: string, + botId: string, + options: { startTime?: string; endTime?: string } = {}, + ): Promise { + const queryParams = new URLSearchParams({ botId }); + if (options.startTime) queryParams.set('startTime', options.startTime); + if (options.endTime) queryParams.set('endTime', options.endTime); + return this.get( + `/api/v1/monitoring/sessions/${encodeURIComponent(sessionId)}/analysis?${queryParams.toString()}`, + ); + } + public getSessionMessages( sessionId: string, limit: number = 200, offset: number = 0, + botId?: string, ): Promise<{ messages: Array<{ id: string; @@ -590,6 +604,7 @@ export class BackendClient extends BaseHttpClient { }> { const queryParams = new URLSearchParams(); queryParams.append('sessionId', sessionId); + if (botId) queryParams.append('botId', botId); queryParams.append('limit', limit.toString()); queryParams.append('offset', offset.toString()); return this.get(`/api/v1/monitoring/messages?${queryParams.toString()}`); @@ -1496,6 +1511,11 @@ export class BackendClient extends BaseHttpClient { endTime?: string; limit?: number; }): Promise<{ + traffic?: { + bucket: 'hour' | 'day'; + points: Array<{ timestamp: string; messages: number; llm_calls: number }>; + truncated: boolean; + }; overview: { total_messages: number; llm_calls: number; diff --git a/web/src/i18n/locales/en-US.ts b/web/src/i18n/locales/en-US.ts index e5192ecad..bdab194bc 100644 --- a/web/src/i18n/locales/en-US.ts +++ b/web/src/i18n/locales/en-US.ts @@ -1644,7 +1644,16 @@ const enUS = { queryVariables: { title: 'Query Variables', }, + loadError: 'Failed to load monitoring data', + partialMessages: + 'Showing {{shown}} of {{total}} messages. Conversation traces may be incomplete.', + partialModelCalls: 'Showing {{shown}} of {{total}} model calls.', + partialToolCalls: + 'Showing {{shown}} of {{total}} tool calls. Conversation traces may be incomplete.', + partialErrors: 'Showing {{shown}} of {{total}} errors.', trafficChart: { + unavailable: 'Traffic aggregation unavailable', + truncated: 'Traffic range truncated. Choose a shorter time range.', title: 'Traffic Overview', messages: 'Messages', llmCalls: 'LLM Calls', diff --git a/web/src/i18n/locales/es-ES.ts b/web/src/i18n/locales/es-ES.ts index 9f8f97d99..42a2da720 100644 --- a/web/src/i18n/locales/es-ES.ts +++ b/web/src/i18n/locales/es-ES.ts @@ -1602,7 +1602,17 @@ const esES = { queryVariables: { title: 'Variables de consulta', }, + loadError: 'No se pudieron cargar los datos de monitoreo', + partialMessages: + 'Se muestran {{shown}} de {{total}} mensajes. Las trazas de conversación pueden estar incompletas.', + partialModelCalls: 'Se muestran {{shown}} de {{total}} llamadas al modelo.', + partialToolCalls: + 'Se muestran {{shown}} de {{total}} llamadas a herramientas. Las trazas de conversación pueden estar incompletas.', + partialErrors: 'Se muestran {{shown}} de {{total}} errores.', trafficChart: { + unavailable: 'Agregación de tráfico no disponible', + truncated: + 'Rango de tráfico truncado. Selecciona un intervalo más corto.', title: 'Resumen de tráfico', messages: 'Mensajes', llmCalls: 'Llamadas LLM', diff --git a/web/src/i18n/locales/ja-JP.ts b/web/src/i18n/locales/ja-JP.ts index 37fe6f6b2..ec0f06fdb 100644 --- a/web/src/i18n/locales/ja-JP.ts +++ b/web/src/i18n/locales/ja-JP.ts @@ -1653,7 +1653,17 @@ const jaJP = { queryVariables: { title: 'クエリ変数', }, + loadError: 'モニタリングデータを読み込めませんでした', + partialMessages: + '全 {{total}} 件中 {{shown}} 件のメッセージを表示。会話トレースは不完全な場合があります。', + partialModelCalls: '全 {{total}} 件中 {{shown}} 件のモデル呼び出しを表示。', + partialToolCalls: + '全 {{total}} 件中 {{shown}} 件のツール呼び出しを表示。会話トレースは不完全な場合があります。', + partialErrors: '全 {{total}} 件中 {{shown}} 件のエラーを表示。', trafficChart: { + unavailable: 'トラフィック集計を利用できません', + truncated: + 'トラフィック範囲が切り詰められています。短い期間を選択してください。', title: 'トラフィック概要', messages: 'メッセージ', llmCalls: 'LLM呼び出し', diff --git a/web/src/i18n/locales/ru-RU.ts b/web/src/i18n/locales/ru-RU.ts index 261c8c9e3..eb707530a 100644 --- a/web/src/i18n/locales/ru-RU.ts +++ b/web/src/i18n/locales/ru-RU.ts @@ -1574,7 +1574,16 @@ const ruRU = { queryVariables: { title: 'Переменные запроса', }, + loadError: 'Не удалось загрузить данные мониторинга', + partialMessages: + 'Показано {{shown}} из {{total}} сообщений. Трассировки диалогов могут быть неполными.', + partialModelCalls: 'Показано {{shown}} из {{total}} вызовов модели.', + partialToolCalls: + 'Показано {{shown}} из {{total}} вызовов инструментов. Трассировки диалогов могут быть неполными.', + partialErrors: 'Показано {{shown}} из {{total}} ошибок.', trafficChart: { + unavailable: 'Агрегированные данные трафика недоступны', + truncated: 'Диапазон трафика обрезан. Выберите более короткий период.', title: 'Обзор трафика', messages: 'Сообщения', llmCalls: 'Вызовы LLM', diff --git a/web/src/i18n/locales/th-TH.ts b/web/src/i18n/locales/th-TH.ts index 2621d770a..c0a552a9e 100644 --- a/web/src/i18n/locales/th-TH.ts +++ b/web/src/i18n/locales/th-TH.ts @@ -1543,7 +1543,17 @@ const thTH = { queryVariables: { title: 'ตัวแปรคำค้นหา', }, + loadError: 'โหลดข้อมูลการตรวจสอบไม่สำเร็จ', + partialMessages: + 'แสดง {{shown}} จาก {{total}} ข้อความ ประวัติการสนทนาอาจไม่ครบถ้วน', + partialModelCalls: 'แสดง {{shown}} จาก {{total}} การเรียกโมเดล', + partialToolCalls: + 'แสดง {{shown}} จาก {{total}} การเรียกเครื่องมือ ประวัติการสนทนาอาจไม่ครบถ้วน', + partialErrors: 'แสดง {{shown}} จาก {{total}} ข้อผิดพลาด', trafficChart: { + unavailable: 'ไม่มีข้อมูลสรุปปริมาณการใช้งาน', + truncated: + 'ช่วงข้อมูลปริมาณการใช้งานถูกตัดทอน โปรดเลือกช่วงเวลาที่สั้นลง', title: 'ภาพรวมปริมาณการใช้งาน', messages: 'ข้อความ', llmCalls: 'การเรียก LLM', diff --git a/web/src/i18n/locales/vi-VN.ts b/web/src/i18n/locales/vi-VN.ts index 6e81af16f..a8b9f73a9 100644 --- a/web/src/i18n/locales/vi-VN.ts +++ b/web/src/i18n/locales/vi-VN.ts @@ -1567,7 +1567,17 @@ const viVN = { queryVariables: { title: 'Biến truy vấn', }, + loadError: 'Không thể tải dữ liệu giám sát', + partialMessages: + 'Hiển thị {{shown}} trên {{total}} tin nhắn. Dấu vết hội thoại có thể không đầy đủ.', + partialModelCalls: 'Hiển thị {{shown}} trên {{total}} lượt gọi mô hình.', + partialToolCalls: + 'Hiển thị {{shown}} trên {{total}} lượt gọi công cụ. Dấu vết hội thoại có thể không đầy đủ.', + partialErrors: 'Hiển thị {{shown}} trên {{total}} lỗi.', trafficChart: { + unavailable: 'Không có dữ liệu tổng hợp lưu lượng', + truncated: + 'Phạm vi lưu lượng bị cắt ngắn. Hãy chọn khoảng thời gian ngắn hơn.', title: 'Tổng quan lưu lượng', messages: 'Tin nhắn', llmCalls: 'Cuộc gọi LLM', diff --git a/web/src/i18n/locales/zh-Hans.ts b/web/src/i18n/locales/zh-Hans.ts index e91090f4d..8fbfc73b8 100644 --- a/web/src/i18n/locales/zh-Hans.ts +++ b/web/src/i18n/locales/zh-Hans.ts @@ -1572,7 +1572,16 @@ const zhHans = { queryVariables: { title: '查询变量', }, + loadError: '监控数据加载失败', + partialMessages: + '显示 {{total}} 条消息中的 {{shown}} 条，对话轨迹可能不完整。', + partialModelCalls: '显示 {{total}} 次模型调用中的 {{shown}} 次。', + partialToolCalls: + '显示 {{total}} 次工具调用中的 {{shown}} 次，对话轨迹可能不完整。', + partialErrors: '显示 {{total}} 条错误中的 {{shown}} 条。', trafficChart: { + unavailable: '流量聚合数据不可用', + truncated: '流量时间范围已截断，请选择更短的时间范围。', title: '流量概览', messages: '消息数', llmCalls: 'LLM调用', diff --git a/web/src/i18n/locales/zh-Hant.ts b/web/src/i18n/locales/zh-Hant.ts index 1920a8ba4..75e3e6a4f 100644 --- a/web/src/i18n/locales/zh-Hant.ts +++ b/web/src/i18n/locales/zh-Hant.ts @@ -1495,7 +1495,16 @@ const zhHant = { queryVariables: { title: '查詢變數', }, + loadError: '監控資料載入失敗', + partialMessages: + '顯示 {{total}} 則訊息中的 {{shown}} 則，對話軌跡可能不完整。', + partialModelCalls: '顯示 {{total}} 次模型呼叫中的 {{shown}} 次。', + partialToolCalls: + '顯示 {{total}} 次工具呼叫中的 {{shown}} 次，對話軌跡可能不完整。', + partialErrors: '顯示 {{total}} 筆錯誤中的 {{shown}} 筆。', trafficChart: { + unavailable: '流量彙總資料無法使用', + truncated: '流量時間範圍已截斷，請選擇較短的時間範圍。', title: '流量概覽', messages: '訊息', llmCalls: 'LLM呼叫', diff --git a/web/tests/e2e/bot-session-tool-timeline.spec.ts b/web/tests/e2e/bot-session-tool-timeline.spec.ts index 7214f1780..5f580f4a0 100644 --- a/web/tests/e2e/bot-session-tool-timeline.spec.ts +++ b/web/tests/e2e/bot-session-tool-timeline.spec.ts @@ -66,7 +66,381 @@ function toolCall( }; } +test.describe('bot session request recovery', () => { + for (const failure of [ + 'initial list', + 'list page', + 'session switch', + 'message page', + 'analysis', + ]) { + test(`${failure} failure is visible and retry recovers`, async ({ + page, + }) => { + await installLangBotApiMocks(page, { authenticated: true }); + let failing = true; + await page.route('**/api/v1/monitoring/**', async (route) => { + const url = new URL(route.request().url()); + const offset = Number(url.searchParams.get('offset') || 0); + const second = url.searchParams.get('sessionId') === 'person-second'; + const list = url.pathname.endsWith('/sessions'); + const message = url.pathname.endsWith('/messages'); + const analysis = url.pathname.endsWith('/analysis'); + if (!list && !message && !analysis) return route.fallback(); + const fail = + failing && + ((list && failure === 'initial list') || + (list && failure === 'list page' && offset > 0) || + (message && failure === 'session switch' && second) || + (message && failure === 'message page' && offset > 0) || + (analysis && failure === 'analysis')); + if (fail) + return route.fulfill({ + status: 500, + json: { code: 500, message: 'fixture failure' }, + }); + const data = list + ? { + sessions: [sessionId, 'person-second'].map((id, i) => ({ + session_id: id, + bot_id: botId, + bot_name: botName, + pipeline_id: pipelineId, + pipeline_name: pipelineName, + message_count: 51, + start_time: at(0), + last_activity: at(4), + is_active: true, + user_name: offset ? `Page two ${i}` : `Recovery user ${i}`, + })), + total: 21, + } + : message + ? { + messages: [ + sessionMessage( + 'recovery-message', + 'user', + 0, + second + ? 'Second session message' + : offset + ? 'Second page message' + : 'Successful message', + ), + ], + total: 51, + } + : { + tool_calls: [ + toolCall('recovery-tool', 1, 'recovered_tool', 40), + ], + }; + return route.fulfill({ json: { code: 0, data } }); + }); + await page.goto(`/home/bots?id=${botId}`); + await page.getByRole('tab', { name: /Sessions/ }).click(); + if (failure === 'list page') { + await page.getByRole('button', { name: 'Next', exact: true }).click(); + } else if (failure !== 'initial list') { + await page.getByRole('button', { name: /Recovery user 0/ }).click(); + if (failure !== 'analysis') { + await expect( + page.getByText('Successful message', { exact: true }), + ).toBeVisible(); + if (failure === 'session switch') + await page.getByRole('button', { name: /Recovery user 1/ }).click(); + else + await page + .getByRole('button', { name: 'Next', exact: true }) + .last() + .click(); + } + } + await expect(page.getByRole('alert')).toBeVisible(); + await expect( + page.getByText('No sessions found', { exact: true }), + ).toHaveCount(0); + if (failure === 'analysis') { + await expect(page.getByRole('alert')).toContainText(/Tool/i); + await expect( + page.getByText('Successful message', { exact: true }), + ).toBeVisible(); + } else { + await expect( + page.getByText('Successful message', { exact: true }), + ).toHaveCount(0); + } + if (failure === 'list page') + await expect( + page.getByRole('button', { name: /Recovery user 0/ }), + ).toHaveCount(0); + failing = false; + await page + .getByRole('alert') + .getByRole('button', { name: 'Retry', exact: true }) + .click(); + await expect(page.getByRole('alert')).toHaveCount(0); + if (failure === 'initial list' || failure === 'list page') { + await expect( + page.getByRole('button', { + name: failure === 'list page' ? /Page two 0/ : /Recovery user 0/, + }), + ).toBeVisible(); + } else { + await expect( + page.getByText( + failure === 'session switch' + ? 'Second session message' + : failure === 'message page' + ? 'Second page message' + : 'Successful message', + { exact: true }, + ), + ).toBeVisible(); + await expect( + page.getByText('recovered_tool', { exact: true }), + ).toBeVisible(); + } + }); + } +}); + +test.describe('bot session request races', () => { + for (const kind of ['messages', 'analysis', 'sessions']) { + for (const status of [200, 500]) { + test(`ignores stale ${kind} ${status} after switching`, async ({ + page, + }) => { + await installLangBotApiMocks(page, { authenticated: true }); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let held = false; + let released = false; + await page.route('**/api/v1/monitoring/**', async (route) => { + const url = new URL(route.request().url()); + const list = url.pathname.endsWith('/sessions'); + const message = url.pathname.endsWith('/messages'); + const analysis = url.pathname.endsWith('/analysis'); + if (!list && !message && !analysis) return route.fallback(); + const old = + kind === 'sessions' + ? url.searchParams.get('userQuery') === 'old' + : message + ? url.searchParams.get('sessionId') === sessionId + : url.pathname.includes(sessionId); + const isHeld = old && url.pathname.endsWith(`/${kind}`); + if (isHeld) { + held = true; + await gate; + if (status === 500) { + await route.fulfill({ status: 500, json: { code: 500 } }); + released = true; + return; + } + } + const data = list + ? { + sessions: [sessionId, 'person-new'].map((id, i) => ({ + session_id: id, + bot_id: botId, + bot_name: botName, + pipeline_id: pipelineId, + pipeline_name: pipelineName, + message_count: 1, + start_time: at(0), + last_activity: at(4), + is_active: true, + user_name: isHeld ? 'Stale list' : `Race user ${i}`, + })), + total: 2, + } + : message + ? { + messages: [ + sessionMessage( + 'race-message', + 'user', + 0, + old ? 'Old message' : 'Current message', + ), + ], + total: 1, + } + : { + tool_calls: [ + toolCall( + 'race-tool', + 1, + old ? 'old_tool' : 'current_tool', + 40, + ), + ], + }; + await route.fulfill({ json: { code: 0, data } }); + if (isHeld) released = true; + }); + await page.goto(`/home/bots?id=${botId}`); + await page.getByRole('tab', { name: /Sessions/ }).click(); + if (kind === 'sessions') { + await page + .getByRole('textbox', { name: 'User ID or name' }) + .fill('old'); + await page + .getByRole('textbox', { name: 'User ID or name' }) + .press('Enter'); + } else await page.getByRole('button', { name: /Race user 0/ }).click(); + await expect.poll(() => held).toBe(true); + if (kind === 'sessions') { + await page + .getByRole('textbox', { name: 'User ID or name' }) + .fill('new'); + await page + .getByRole('textbox', { name: 'User ID or name' }) + .press('Enter'); + await expect( + page.getByRole('button', { name: /Race user 0/ }), + ).toBeVisible(); + } else { + await page.getByRole('button', { name: /Race user 1/ }).click(); + await expect( + page.getByText('Current message', { exact: true }), + ).toBeVisible(); + } + release(); + await expect.poll(() => released).toBe(true); + // Allow the released HTTP response and React's queued update to settle. + await page.waitForTimeout(200); + await expect(page.getByRole('alert')).toHaveCount(0); + await expect(page.getByText('Stale list', { exact: true })).toHaveCount( + 0, + ); + if (kind !== 'sessions') { + await expect( + page.getByText('Current message', { exact: true }), + ).toBeVisible(); + await expect( + page.getByText('current_tool', { exact: true }), + ).toBeVisible(); + await expect( + page.getByText('Old message', { exact: true }), + ).toHaveCount(0); + await expect(page.getByText('old_tool', { exact: true })).toHaveCount( + 0, + ); + } + }); + } + } +}); + test.describe('bot session monitor tool timeline', () => { + test('isolates messages and analysis for two bots sharing a raw session id', async ({ + page, + }) => { + await installLangBotApiMocks(page, { authenticated: true }); + const requests: Array<{ bot: string; path: string }> = []; + await page.route('**/api/v1/monitoring/**', async (route) => { + const url = new URL(route.request().url()); + const selectedBot = url.searchParams.get('botId'); + if ( + !url.pathname.endsWith('/sessions') && + !url.pathname.endsWith('/messages') && + !url.pathname.endsWith('/analysis') + ) { + return route.fallback(); + } + expect(['bot-shared-a', 'bot-shared-b']).toContain(selectedBot); + expect(route.request().headers().authorization).toBe( + 'Bearer playwright-token', + ); + expect(route.request().headers()['x-workspace-id']).toBe( + 'workspace-playwright', + ); + requests.push({ bot: selectedBot!, path: url.pathname }); + const shared = { + session_id: sessionId, + bot_id: selectedBot, + bot_name: selectedBot, + pipeline_id: pipelineId, + pipeline_name: pipelineName, + message_count: 1, + start_time: at(0), + last_activity: at(4), + is_active: true, + platform: 'person', + user_id: 'shared-user', + user_name: 'Shared User', + }; + const data = url.pathname.endsWith('/sessions') + ? { sessions: [shared], total: 1 } + : url.pathname.endsWith('/messages') + ? { + messages: [ + { + ...sessionMessage( + 'shared-message', + 'user', + 0, + `Message for ${selectedBot}`, + ), + bot_id: selectedBot, + }, + ], + total: 1, + } + : { + session_id: sessionId, + found: true, + tool_calls: [ + { + ...toolCall('shared-tool', 1, `tool_${selectedBot}`, 40), + bot_id: selectedBot, + }, + ], + }; + if (url.pathname.endsWith('/messages')) + expect(url.searchParams.get('sessionId')).toBe(sessionId); + if (url.pathname.endsWith('/analysis')) + expect(decodeURIComponent(url.pathname)).toContain( + `/sessions/${sessionId}/analysis`, + ); + await route.fulfill({ json: { code: 0, data } }); + }); + for (const selectedBot of ['bot-shared-a', 'bot-shared-b']) { + await page.goto(`/home/bots?id=${selectedBot}`); + await page.getByRole('tab', { name: /Sessions/ }).click(); + await page.getByRole('button', { name: /Shared User/ }).click(); + await expect( + page.getByText(`Message for ${selectedBot}`, { exact: true }), + ).toBeVisible(); + await expect( + page.getByText(`tool_${selectedBot}`, { exact: true }), + ).toBeVisible(); + const otherBot = + selectedBot === 'bot-shared-a' ? 'bot-shared-b' : 'bot-shared-a'; + await expect( + page.getByText(`Message for ${otherBot}`, { exact: true }), + ).toHaveCount(0); + await expect( + page.getByText(`tool_${otherBot}`, { exact: true }), + ).toHaveCount(0); + expect( + requests.some( + (request) => + request.bot === selectedBot && request.path.endsWith('/messages'), + ), + ).toBe(true); + expect( + requests.some( + (request) => + request.bot === selectedBot && request.path.endsWith('/analysis'), + ), + ).toBe(true); + } + }); test('renders tool calls as left-side agent events interleaved with messages', async ({ page, }) => { @@ -117,11 +491,41 @@ test.describe('bot session monitor tool timeline', () => { }, }); + const monitoringRequests: import('@playwright/test').Request[] = []; + page.on('request', (request) => { + if (request.url().includes('/api/v1/monitoring/')) + monitoringRequests.push(request); + }); await page.goto(`/home/bots?id=${botId}`); await page.getByRole('tab', { name: /Sessions/ }).click(); await page.getByRole('button', { name: /Timeline User/ }).click(); await expect(page.getByText('Need a timeline check')).toBeVisible(); + await expect + .poll(() => + monitoringRequests.some((request) => + request.url().includes('/analysis?'), + ), + ) + .toBe(true); + for (const request of monitoringRequests.filter((request) => + /\/messages\?|\/analysis\?/.test(request.url()), + )) { + const url = new URL(request.url()); + expect(url.searchParams.get('botId')).toBe(botId); + if (url.pathname.endsWith('/analysis')) { + expect(url.searchParams.get('startTime')).toBe(at(0)); + expect(url.searchParams.get('endTime')).toBe(at(4)); + } + expect(request.headers().authorization).toBe('Bearer playwright-token'); + expect(request.headers()['x-workspace-id']).toBe('workspace-playwright'); + if (url.pathname.endsWith('/messages')) + expect(url.searchParams.get('sessionId')).toBe(sessionId); + else + expect(decodeURIComponent(url.pathname)).toContain( + `/sessions/${sessionId}/analysis`, + ); + } await expect( page.getByText('repo_file_read', { exact: true }), ).toBeVisible(); diff --git a/web/tests/e2e/monitoring-turns.spec.ts b/web/tests/e2e/monitoring-turns.spec.ts index 89c98554f..2abc76eb6 100644 --- a/web/tests/e2e/monitoring-turns.spec.ts +++ b/web/tests/e2e/monitoring-turns.spec.ts @@ -1,4 +1,4 @@ -import { expect, test } from '@playwright/test'; +import { expect, test, Route } from '@playwright/test'; import { installLangBotApiMocks } from './fixtures/langbot-api'; import { buildConversationTurns } from '../../src/app/home/monitoring/utils/conversationTurns'; @@ -271,7 +271,198 @@ function rawMonitoringData() { }; } +async function respond(route: Route, label: string) { + const data = rawMonitoringData(); + data.messages = [rawMessage(message(label, 'user', 10, label))]; + await route.fulfill({ json: { code: 0, data } }); +} + +test.describe('monitoring request contracts', () => { + test('shows failures instead of empty success and retries with auth and Workspace headers', async ({ + page, + }) => { + await installLangBotApiMocks(page, { authenticated: true }); + let failing = true; + await page.route('**/api/v1/monitoring/data?*', async (route) => { + expect(route.request().headers().authorization).toBe( + 'Bearer playwright-token', + ); + expect(route.request().headers()['x-workspace-id']).toBe( + 'workspace-playwright', + ); + if (failing) + await route.fulfill({ + status: 500, + json: { code: 500, msg: 'fixture database unavailable' }, + }); + else await respond(route, 'Recovered monitoring'); + }); + await page.goto('/home/monitoring'); + await expect(page.getByRole('alert')).toContainText( + 'Failed to load monitoring data', + ); + await expect(page.getByText('No message records')).toHaveCount(0); + failing = false; + await page.getByRole('button', { name: 'Retry', exact: true }).click(); + await expect( + page.getByText('Recovered monitoring', { exact: true }), + ).toBeVisible(); + await expect(page.getByRole('alert')).toHaveCount(0); + }); + + test('latest filter request wins over delayed data and delayed failures', async ({ + page, + }) => { + await installLangBotApiMocks(page, { authenticated: true }); + const pending: Route[] = []; + await page.route('**/api/v1/monitoring/data?*', (route) => { + pending.push(route); + }); + await page.goto('/home/monitoring'); + await expect.poll(() => pending.length).toBe(2); + await page.getByRole('combobox').last().click(); + await page.getByRole('option', { name: /Last 7 days/i }).click(); + await expect.poll(() => pending.length).toBe(3); + await respond(pending[2], 'Latest filter data'); + await expect( + page.getByText('Latest filter data', { exact: true }), + ).toBeVisible(); + await respond(pending[0], 'Obsolete filter data'); + await respond(pending[1], 'Obsolete filter data'); + await page.evaluate( + () => + new Promise((resolve) => + requestAnimationFrame(() => requestAnimationFrame(() => resolve())), + ), + ); + await expect( + page.getByText('Latest filter data', { exact: true }), + ).toBeVisible(); + await page + .getByRole('button', { name: 'Refresh Data', exact: true }) + .click(); + await expect.poll(() => pending.length).toBe(4); + await expect( + page.getByText('Obsolete filter data', { exact: true }), + ).toHaveCount(0); + await page.getByRole('combobox').last().click(); + await page.getByRole('option', { name: /Last 24 hours/i }).click(); + await expect.poll(() => pending.length).toBe(5); + await respond(pending[4], 'Current result'); + await expect( + page.getByText('Current result', { exact: true }), + ).toBeVisible(); + await pending[3].fulfill({ + status: 500, + json: { code: 500, msg: 'old failure' }, + }); + await expect( + page.getByText('Current result', { exact: true }), + ).toBeVisible(); + await expect(page.getByRole('alert')).toHaveCount(0); + }); + + test('uses aggregate traffic rather than the sparse record page and discloses truncation', async ({ + page, + }) => { + const data = rawMonitoringData(); + data.totalCount.messages = 125; + await installLangBotApiMocks(page, { + authenticated: true, + monitoringData: { + ...data, + traffic: { + bucket: 'hour', + truncated: true, + points: [ + { timestamp: time(0).toISOString(), messages: 125, llm_calls: 77 }, + { timestamp: time(1).toISOString(), messages: 0, llm_calls: 0 }, + ], + }, + }, + }); + await page.goto('/home/monitoring'); + await expect( + page.getByText( + 'Showing 7 of 125 messages. Conversation traces may be incomplete.', + ), + ).toBeVisible(); + await expect( + page.getByText('Traffic range truncated. Choose a shorter time range.'), + ).toBeVisible(); + const chart = page.locator('.recharts-wrapper'); + await expect(chart).toHaveCount(1); + await chart + .locator(':scope > .recharts-surface') + .hover({ position: { x: 70, y: 100 } }); + await expect(chart.locator('.recharts-tooltip-wrapper')).toContainText( + '125', + ); + await expect(chart.locator('.recharts-tooltip-wrapper')).toContainText( + '77', + ); + }); + + test('does not invent traffic totals when aggregation is unavailable', async ({ + page, + }) => { + await installLangBotApiMocks(page, { + authenticated: true, + monitoringData: rawMonitoringData(), + }); + await page.goto('/home/monitoring'); + await expect( + page.getByText('Traffic aggregation unavailable'), + ).toBeVisible(); + await expect(page.locator('.recharts-wrapper')).toHaveCount(0); + }); +}); + test.describe('monitoring conversation turn grouping', () => { + test('does not reassign explicitly linked activity outside the visible page', () => { + const turns = buildConversationTurns( + [message('visible', 'user', 10, 'Visible turn')], + [llmCall('older-call', 11, 'off-page', 10, 5, 40)], + [errorLog('older-error', 11, 'off-page')], + [toolCall('older-tool', 11, 'off-page', 'search', 40)], + ); + expect(turns[0].llmCalls).toEqual([]); + expect(turns[0].toolCalls).toEqual([]); + expect(turns[0].errors).toEqual([]); + }); + + test('does not assign unlinked activity before the first visible turn', () => { + const turns = buildConversationTurns( + [message('visible', 'user', 10, 'Visible turn')], + [llmCall('older-call', 1, undefined, 10, 5, 40)], + [{ ...errorLog('older-error', 1, ''), messageId: undefined }], + [toolCall('older-tool', 1, undefined, 'search', 40)], + ); + expect(turns[0].llmCalls).toEqual([]); + expect(turns[0].toolCalls).toEqual([]); + expect(turns[0].errors).toEqual([]); + }); + + test('isolates same-session messages and activity by bot identity', () => { + const first = message('first', 'user', 1, 'Bot one'); + const other = { + ...message('other', 'user', 2, 'Bot two'), + botId: 'other-bot', + }; + const reply = message('reply', 'assistant', 3, 'Bot one reply'); + const turns = buildConversationTurns( + [first, other, reply], + [llmCall('call', 3, undefined, 10, 5, 40)], + [errorLog('error', 3, first.id)], + [toolCall('tool', 3, undefined, 'search', 40)], + ); + const own = turns.find((turn) => turn.id === first.id)!; + expect(own.assistantMessages.map((item) => item.id)).toEqual(['reply']); + expect(own.llmCalls.map((item) => item.id)).toEqual(['call']); + expect(own.toolCalls.map((item) => item.id)).toEqual(['tool']); + expect(turns.find((turn) => turn.id === other.id)?.totalTokens).toBe(0); + }); + test('keeps a single user message as one observable turn', () => { const userOnly = message( 'single-user-only', diff --git a/web/tests/unit/session-monitor-pagination.test.mjs b/web/tests/unit/session-monitor-pagination.test.mjs index f855237f4..38b09f306 100644 --- a/web/tests/unit/session-monitor-pagination.test.mjs +++ b/web/tests/unit/session-monitor-pagination.test.mjs @@ -134,6 +134,22 @@ test('session tool calls are bounded to the visible message page', () => { const monitor = read( 'src/app/home/bots/components/bot-session/BotSessionMonitor.tsx', ); - includes(monitor, "analysisParams.set('startTime'", 'analysis page start'); - includes(monitor, "analysisParams.set('endTime'", 'analysis page end'); + includes(monitor, 'startTime: sorted[0]?.timestamp', 'analysis page start'); + includes( + monitor, + 'endTime: sorted[sorted.length - 1]?.timestamp', + 'analysis page end', + ); + includes(monitor, 'sessionId, botId, {', 'bot-scoped analysis'); + const client = read('src/app/infra/http/BackendClient.ts'); + includes( + client, + "queryParams.set('startTime', options.startTime)", + 'analysis start query', + ); + includes( + client, + "queryParams.set('endTime', options.endTime)", + 'analysis end query', + ); });