From 1318efc2488ed540083983eb80429173d34b0c14 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 19:41:55 +0800 Subject: [PATCH 01/15] perf(persistence): index workspace execution leases Add composite indexes on Workspace, status, and lease expiry for AgentRun and MonitoringMessage so lease maintenance can locate active executions without scanning completed history. Keep ORM metadata and Alembic migration 0039 consistent, including fresh installations where the indexes already exist. Cover migration round trips, data preservation, and SQLite query plans with regression tests. --- .../pkg/entity/persistence/agent_run.py | 1 + .../pkg/entity/persistence/monitoring.py | 6 + .../versions/0039_execution_lease_indexes.py | 29 ++++ .../test_execution_lease_indexes.py | 134 ++++++++++++++++++ 4 files changed, 170 insertions(+) create mode 100644 src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py create mode 100644 tests/unit_tests/persistence/test_execution_lease_indexes.py diff --git a/src/langbot/pkg/entity/persistence/agent_run.py b/src/langbot/pkg/entity/persistence/agent_run.py index 7d5dc0a8c..af1f9f6b6 100644 --- a/src/langbot/pkg/entity/persistence/agent_run.py +++ b/src/langbot/pkg/entity/persistence/agent_run.py @@ -116,6 +116,7 @@ class AgentRun(Base): ), sqlalchemy.Index('ix_agent_run_runner_status', 'runner_id', 'status'), sqlalchemy.Index('ix_agent_run_queue_claim', 'queue_name', 'status', 'priority', 'id'), + sqlalchemy.Index('ix_agent_run_workspace_status_lease', 'workspace_id', 'status', 'execution_lease_expires_at'), ) diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index ecf032986..e1a92159e 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -40,6 +40,12 @@ class MonitoringMessage(Base): __table_args__ = ( sqlalchemy.Index('ix_monitoring_messages_workspace_timestamp', 'workspace_uuid', 'timestamp'), sqlalchemy.Index('ix_monitoring_messages_workspace_session', 'workspace_uuid', 'session_id'), + sqlalchemy.Index( + 'ix_monitoring_messages_workspace_status_lease', + 'workspace_uuid', + 'status', + 'execution_lease_expires_at', + ), ) diff --git a/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py new file mode 100644 index 000000000..4d66427e4 --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py @@ -0,0 +1,29 @@ +"""Bound host lease maintenance to active executions within each Workspace.""" + +from alembic import op +import sqlalchemy as sa + +revision = '0039_execution_lease_indexes' +down_revision = '0038_llm_call_details' +branch_labels = None +depends_on = None + + +def upgrade(): + inspector = sa.inspect(op.get_bind()) + for table, workspace_column in ( + ('agent_run', 'workspace_id'), + ('monitoring_messages', 'workspace_uuid'), + ): + index = f'ix_{table}_workspace_status_lease' + # Fresh installations create these indexes from ORM metadata before migrations. + if index not in {item['name'] for item in inspector.get_indexes(table)}: + op.create_index(index, table, [workspace_column, 'status', 'execution_lease_expires_at']) + + +def downgrade(): + inspector = sa.inspect(op.get_bind()) + for table in ('agent_run', 'monitoring_messages'): + index = f'ix_{table}_workspace_status_lease' + if index in {item['name'] for item in inspector.get_indexes(table)}: + op.drop_index(index, table_name=table) diff --git a/tests/unit_tests/persistence/test_execution_lease_indexes.py b/tests/unit_tests/persistence/test_execution_lease_indexes.py new file mode 100644 index 000000000..468871087 --- /dev/null +++ b/tests/unit_tests/persistence/test_execution_lease_indexes.py @@ -0,0 +1,134 @@ +"""Lease maintenance must not scan a Workspace's completed execution history.""" + +import datetime +import importlib + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.entity.persistence import user, workspace # noqa: F401 +from langbot.pkg.entity.persistence.agent_run import AgentRun +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage +from langbot.pkg.utils.execution_lease import reconcile_execution_leases + + +MIGRATION = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') +LEASE_INDEXES = { + 'agent_run': ('ix_agent_run_workspace_status_lease', ['workspace_id', 'status', 'execution_lease_expires_at']), + 'monitoring_messages': ( + 'ix_monitoring_messages_workspace_status_lease', + ['workspace_uuid', 'status', 'execution_lease_expires_at'], + ), +} + + +def _create_schema(connection, source): + Base.metadata.create_all(connection) + if source == 'migration': + for table, (index, _) in LEASE_INDEXES.items(): + next(item for item in Base.metadata.tables[table].indexes if item.name == index).drop(connection) + with Operations.context(MigrationContext.configure(connection)): + MIGRATION.upgrade() + + +@pytest.mark.parametrize('source', ['metadata', 'migration']) +def test_lease_index_migration_preserves_data_and_existing_indexes(source): + engine = sa.create_engine('sqlite:///:memory:') + try: + with engine.begin() as connection: + _create_schema(connection, source) + connection.execute(sa.insert(AgentRun).values(run_id='history', runner_id='runner', status='completed')) + original_indexes = { + table: {index['name'] for index in sa.inspect(connection).get_indexes(table)} for table in LEASE_INDEXES + } + with Operations.context(MigrationContext.configure(connection)): + MIGRATION.upgrade() + for table, (index, columns) in LEASE_INDEXES.items(): + indexes = {item['name']: item['column_names'] for item in sa.inspect(connection).get_indexes(table)} + assert indexes[index] == columns + assert original_indexes[table] <= indexes.keys() + + MIGRATION.downgrade() + for table, (index, _) in LEASE_INDEXES.items(): + remaining = {item['name'] for item in sa.inspect(connection).get_indexes(table)} + assert remaining == original_indexes[table] - {index} + assert connection.scalar(sa.select(AgentRun.status)) == 'completed' + + MIGRATION.upgrade() + assert connection.scalar(sa.select(AgentRun.status)) == 'completed' + finally: + engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('source', ['metadata', 'migration']) +@pytest.mark.parametrize('model', [AgentRun, MonitoringMessage]) +async def test_lease_queries_use_workspace_status_expiry_index(model, source): + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + now = datetime.datetime(2026, 10, 17) + statements = [] + + def capture_statement(connection, cursor, statement, parameters, context, executemany): + if statement.startswith(('SELECT ', 'UPDATE ')) and model.__tablename__ in statement: + statements.append((statement, parameters)) + + try: + async with engine.begin() as connection: + await connection.run_sync(lambda sync: _create_schema(sync, source)) + common = { + 'execution_owner_id': 'live-host', + 'execution_lease_expires_at': now - datetime.timedelta(seconds=1), + } + if model is AgentRun: + rows = [ + dict( + common, + run_id=str(i), + runner_id='runner', + workspace_id='workspace', + status='completed', + created_at=now, + updated_at=now, + ) + for i in range(2000) + ] + else: + rows = [ + dict( + common, + id=str(i), + workspace_uuid='workspace', + timestamp=now, + bot_id='bot', + bot_name='bot', + pipeline_id='pipeline', + pipeline_name='pipeline', + message_content='Historical message ' * 64, + session_id='session', + status='success', + level='info', + ) + for i in range(2000) + ] + await connection.execute(sa.insert(model), rows) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture_statement) + try: + assert await reconcile_execution_leases(engine, 'workspace', now=now, owner_id='live-host') == 0 + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture_statement) + + assert len(statements) == 2 + async with engine.connect() as connection: + for statement, parameters in statements: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters)).all() + details = ' '.join(row[3] for row in plan) + assert LEASE_INDEXES[model.__tablename__][0] in details, details + assert 'execution_lease_expires_at' in details, details + assert 'USE TEMP B-TREE' not in details, details + finally: + await engine.dispose() From d663331889ecc463de198689c5280fa121b54322 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 19:45:19 +0800 Subject: [PATCH 02/15] fix(monitoring): bound inflight snapshot queries Require a valid owner and unexpired lease for pending Pipeline messages to appear as active tasks. Exclude historical ingress records without rewriting their stored status. Project only the fields needed by snapshots, skip linkage queries for empty results, and resolve linked runs from selected message IDs instead of a cross-table OR join that scans historical messages. Cover lease filtering, terminal transitions, Workspace isolation, payload exclusion, and indexed lookup plans with regression tests. --- .../pkg/api/http/service/monitoring.py | 92 ++++++-- .../service/monitoring_execution_details.py | 72 +++++- .../api/service/test_monitoring_executions.py | 211 ++++++++++++++++++ 3 files changed, 347 insertions(+), 28 deletions(-) diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 0ca23d7e1..cba2ee010 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -1291,15 +1291,16 @@ class MonitoringService(ExecutionDetailsMixin): 'input_preview': _message_preview(json.dumps(metadata.get('input') or {}, ensure_ascii=False), 200), } - def _serialize_pipeline_execution(self, row) -> dict: + def _serialize_pipeline_execution(self, row, *, summary_only=False) -> dict: created_at_ms = _epoch_ms(row.timestamp) + preview = '' if summary_only else _message_preview(row.message_content) return { 'source': 'pipeline', 'id': row.id, 'event_id': row.event_id, 'status': row.status, 'status_group': _status_group(row.status, 'pipeline'), - 'title': _message_preview(row.message_content), + 'title': row.pipeline_name if summary_only else preview, 'target_kind': 'pipeline', 'target_id': row.pipeline_id, 'target_name': row.pipeline_name, @@ -1319,7 +1320,7 @@ class MonitoringService(ExecutionDetailsMixin): 'queue_name': None, 'debug': row.bot_id == 'websocket-proxy-bot', 'has_error': row.status == 'error', - 'input_preview': _message_preview(row.message_content), + 'input_preview': preview, 'platform': row.platform, 'user_id': row.user_id, 'user_name': row.user_name, @@ -1533,6 +1534,23 @@ class MonitoringService(ExecutionDetailsMixin): 'real': total - debug_count, } + @staticmethod + def _inflight_pipeline_condition(*, pending_only: bool = False): + """Historical pending records without a live Host lease are not tasks.""" + Message = persistence_monitoring.MonitoringMessage + live_lease = sqlalchemy.and_( + Message.execution_owner_id.is_not(None), + Message.execution_lease_expires_at > datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), + ) + # Active-only queries must expose the lease range to the composite index. + # The status OR otherwise reads historical payload pages to check role. + if pending_only: + return live_lease + return sqlalchemy.or_( + Message.status != 'pending', + live_lease, + ) + async def get_executions( self, context: TenantContext, @@ -1548,6 +1566,7 @@ class MonitoringService(ExecutionDetailsMixin): limit: int = 50, offset: int = 0, include_summary: bool = True, + for_inflight: bool = False, ) -> dict: """List executions from both object types (agent runs and pipeline queries).""" workspace_uuid = require_workspace_uuid(context) @@ -1579,15 +1598,18 @@ class MonitoringService(ExecutionDetailsMixin): ).where(*conditions) ) if source in {'all', 'pipeline'} and (not agent_ids or pipeline_ids): + pipeline_statuses = self._raw_pipeline_statuses(statuses) conditions = self._pipeline_execution_conditions( workspace_uuid, bot_ids=bot_ids, pipeline_ids=pipeline_ids, - statuses=self._raw_pipeline_statuses(statuses), + statuses=pipeline_statuses, start_time=start_time, end_time=end_time, debug=debug, ) + if for_inflight: + conditions.append(self._inflight_pipeline_condition(pending_only=pipeline_statuses == ['pending'])) branches.append( sqlalchemy.select( sqlalchemy.literal('pipeline').label('source'), @@ -1595,7 +1617,7 @@ class MonitoringService(ExecutionDetailsMixin): Message.timestamp.label('time'), ).where(*conditions) ) - if source in {'all', 'event'} and mode != 'debug' and not pipeline_ids and not agent_ids: + if source in {'all', 'event'} and mode != 'debug' and not pipeline_ids and not agent_ids and not for_inflight: conditions = self._unhandled_event_conditions(workspace_uuid, bot_ids, start_time, end_time) if statuses: conditions.append(self._execution_json_text(EventLog.metadata_json, 'status').in_(statuses)) @@ -1634,20 +1656,23 @@ class MonitoringService(ExecutionDetailsMixin): workspace_column = model.workspace_uuid if kind == 'pipeline' else model.workspace_id records = await self._execution_rows( workspace_uuid, - sqlalchemy.select(model).where(workspace_column == workspace_uuid, identity_column.in_(ids)), + self._execution_statement(model, summary_only=for_inflight).where( + workspace_column == workspace_uuid, identity_column.in_(ids) + ), + summary_only=for_inflight, ) names = await self._agent_names(workspace_uuid, records) if kind == 'agent' else {} for record in records: item = ( self._serialize_agent_execution(record, names) if kind == 'agent' - else self._serialize_pipeline_execution(record) + else self._serialize_pipeline_execution(record, summary_only=for_inflight) if kind == 'pipeline' else self._serialize_event_execution(record) ) by_key[(kind, item['id'])] = item page = [by_key[(ref.source, ref.id)] for ref in refs if (ref.source, ref.id) in by_key] - await self.enrich_execution_rows(workspace_uuid, page) + await self.enrich_execution_rows(workspace_uuid, page, summary_only=for_inflight) if not include_summary: return {'items': page, 'total': total, 'has_more': total > offset + len(page)} return { @@ -1692,7 +1717,9 @@ class MonitoringService(ExecutionDetailsMixin): from ....utils.inflight import inflight_hub workspace = require_workspace_uuid(context) - active = await self.get_executions(context, statuses=['running', 'queued'], limit=100, include_summary=False) + active = await self.get_executions( + context, statuses=['running', 'queued'], limit=100, include_summary=False, for_inflight=True + ) # Execution timestamps are stored as UTC without a timezone. asyncpg # rejects aware datetimes for these columns, even though SQLite accepts them. recent_start = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta( @@ -1703,6 +1730,7 @@ class MonitoringService(ExecutionDetailsMixin): start_time=recent_start, limit=100, include_summary=False, + for_inflight=True, ) items = {(r['source'], r['id']): r for r in recent['items'] + active['items'] if r['source'] != 'event'} # Keep the actual terminal state of older runs that just left the active set. @@ -1722,28 +1750,56 @@ class MonitoringService(ExecutionDetailsMixin): ]: missing = [r['id'] for r in (previous or [])[:100] if r['source'] == kind and (kind, r['id']) not in items] if missing: - records = await self._execution_rows( - workspace, sqlalchemy.select(model).where(scope == workspace, column.in_(missing)) + statement = self._execution_statement(model, summary_only=True).where( + scope == workspace, column.in_(missing) ) + if kind == 'pipeline': + statement = statement.where(self._inflight_pipeline_condition()) + records = await self._execution_rows(workspace, statement, summary_only=True) names = await self._agent_names(workspace, records) if kind == 'agent' else {} for record in records: row = ( self._serialize_agent_execution(record, names) if kind == 'agent' - else self._serialize_pipeline_execution(record) + else self._serialize_pipeline_execution(record, summary_only=True) ) items[(kind, row['id'])] = row rows = list(items.values()) + snapshot = { + 'items': rows, + 'active_total': active['total'], + 'truncated': active['has_more'] or recent['has_more'], + } + if not rows: + inflight_hub.track(workspace, []) + return snapshot Run = persistence_agent_run.AgentRun Message = persistence_monitoring.MonitoringMessage agent_ids = [r['id'] for r in rows if r['source'] == 'agent'] pipeline_ids = [r['id'] for r in rows if r['source'] == 'pipeline'] - linked = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(Run.run_id, Message.id.label('message_id')) - .outerjoin(Message, sqlalchemy.and_(Message.workspace_uuid == workspace, Message.run_id == Run.run_id)) - .where( - Run.workspace_id == workspace, sqlalchemy.or_(Run.run_id.in_(agent_ids), Message.id.in_(pipeline_ids)) + links = [] + if agent_ids: + links.append( + sqlalchemy.select(Run.run_id, sqlalchemy.literal(None).label('message_id')).where( + Run.workspace_id == workspace, Run.run_id.in_(agent_ids) + ) ) + if pipeline_ids: + # Start from the selected message IDs. A cross-table OR after a + # LEFT JOIN makes SQLite scan the attachment-heavy message ledger + # whenever the snapshot contains even one Pipeline execution. + selected_messages = sqlalchemy.union_all( + *[sqlalchemy.select(sqlalchemy.literal(message_id).label('message_id')) for message_id in pipeline_ids] + ).subquery('selected_messages') + links.append( + sqlalchemy.select(Run.run_id, Message.id.label('message_id')) + .select_from(selected_messages) + .join(Message, Message.id == selected_messages.c.message_id) + .join(Run, sqlalchemy.and_(Run.workspace_id == workspace, Run.run_id == Message.run_id)) + .where(Message.workspace_uuid == workspace) + ) + linked = await self.ap.persistence_mgr.execute_async( + links[0] if len(links) == 1 else sqlalchemy.union_all(*links) ) identities = { r.run_id: ('pipeline', r.message_id) if r.message_id in pipeline_ids else ('agent', r.run_id) @@ -1766,7 +1822,7 @@ class MonitoringService(ExecutionDetailsMixin): row = items.get(identities[run_id]) if row is not None: row['progress_event'] = event_type - return {'items': rows, 'active_total': active['total'], 'truncated': active['has_more'] or recent['has_more']} + return snapshot async def get_token_statistics( self, diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py index 467a8026f..edeb17862 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -4,6 +4,7 @@ from __future__ import annotations import datetime import json +from types import SimpleNamespace from langbot_plugin.api.entities.builtin.platform.events import event_summary import sqlalchemy as sa @@ -354,8 +355,44 @@ class ExecutionDetailsMixin: async with AsyncSession(self.ap.persistence_mgr.get_db_engine()) as session: return await session.execute(statement) - async def _execution_rows(self, workspace, statement): - return list((await self._execution_query(workspace, statement)).scalars().all()) + async def _execution_rows(self, workspace, statement, *, summary_only=False): + result = await self._execution_query(workspace, statement) + if not summary_only: + return list(result.scalars().all()) + records = [] + for row in result.mappings(): + fields = dict(row) + metadata = { + key.removeprefix('display_metadata_'): fields.pop(key) + for key in row + if key.startswith('display_metadata_') + } + if metadata: + fields['metadata_json'] = json.dumps(metadata, ensure_ascii=False) + records.append(SimpleNamespace(**fields)) + return records + + def _execution_statement(self, model, *, summary_only=False): + """Keep periodic snapshots independent of message and attachment sizes.""" + if not summary_only: + return sa.select(model) + columns = [ + column + for column in model.__table__.columns + if column.name not in {'message_content', 'variables', 'input_json', 'authorization_json', 'metadata_json'} + ] + if model is models.MonitoringMessage: + return sa.select(*columns) + keys = ( + ('processor_id', 'processor_type', 'event_type', 'agent_name', 'target_name', 'bot_name') + if model is AgentRun + else ('platform', 'bot_name') + ) + # Metadata also contains complete inputs. Project only the small display + # fields in SQL, rather than deserializing an attachment-bearing object. + for key in keys: + columns.append(self._execution_json_text(model.metadata_json, key).label(f'display_metadata_{key}')) + return sa.select(*columns) async def _execution_page(self, workspace, statement, convert, offset, limit): rows = await self._execution_rows(workspace, statement.offset(offset).limit(limit + 1)) @@ -445,14 +482,21 @@ class ExecutionDetailsMixin: or None, } - async def enrich_execution_rows(self, workspace, items): + async def enrich_execution_rows(self, workspace, items, *, summary_only=False): pipeline_items = {item['id']: item for item in items if item.get('source') == 'pipeline'} if pipeline_items: # Pipeline rows own the UI identity; their linked Runner ledger owns # timing and usage. Read them in one batch for both list and detail. linked = await self._execution_query( workspace, - sa.select(models.MonitoringMessage.id, AgentRun) + sa.select( + models.MonitoringMessage.id, + AgentRun.started_at, + AgentRun.finished_at, + AgentRun.usage_json, + AgentRun.cost_json, + ) + .select_from(models.MonitoringMessage) .join( AgentRun, sa.and_( @@ -465,10 +509,15 @@ class ExecutionDetailsMixin: models.MonitoringMessage.id.in_(pipeline_items), ), ) - for message_id, run in linked.all(): - metrics = self._serialize_agent_execution(run) - pipeline_items[message_id].update( - {key: metrics[key] for key in ('started_at_ms', 'finished_at_ms', 'duration_ms', 'usage', 'cost')} + for run in linked.all(): + pipeline_items[run.id].update( + started_at_ms=epoch(run.started_at), + finished_at_ms=epoch(run.finished_at), + duration_ms=int(round((run.finished_at - run.started_at).total_seconds() * 1000)) + if run.started_at and run.finished_at + else None, + usage=self._deserialize_json(run.usage_json) or None, + cost=self._deserialize_json(run.cost_json) or None, ) calls = models.MonitoringLLMCall @@ -498,10 +547,11 @@ class ExecutionDetailsMixin: return events = await self._execution_rows( workspace, - sa.select(EventLog).where( + self._execution_statement(EventLog, summary_only=summary_only).where( EventLog.workspace_id == workspace, EventLog.event_id.in_(ids), ), + summary_only=summary_only, ) by_id = {row.event_id: row for row in events} for item in items: @@ -510,7 +560,9 @@ class ExecutionDetailsMixin: meta = parsed(event.metadata_json, {}) item.update( event_type=event.event_type, - input_preview=event_preview(parsed(event.input_json), event.input_summary, event.event_type), + input_preview=event_preview( + None if summary_only else parsed(event.input_json), event.input_summary, event.event_type + ), user_id=event.actor_id, user_name=event.actor_name, platform=meta.get('platform') or event.source, diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index 63fbd32b5..a9469fcab 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -739,6 +739,217 @@ async def test_pipeline_debug_is_consistent_across_rows_rollups_and_telemetry(en assert sum(p['llm_calls'] for p in traffic['points']) == 1 +@pytest.mark.asyncio +async def test_inflight_excludes_unowned_and_expired_messages_without_rewriting_history(engine, service): + import sqlalchemy as sa + + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + Message = persistence_monitoring.MonitoringMessage + async with AsyncSession(engine) as session: + for message_id, owner, expiry, timestamp in [ + ('legacy', None, None, _dt(1)), + ('recent-unowned', None, now + datetime.timedelta(minutes=1), now), + ('expired', 'stopped-host', now - datetime.timedelta(seconds=1), _dt(1)), + ('long-running', 'live-host', now + datetime.timedelta(minutes=1), _dt(1)), + ]: + session.add( + Message( + id=message_id, + workspace_uuid=WORKSPACE, + timestamp=timestamp, + bot_id='bot', + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session', + message_content='historical message', + status='pending', + level='info', + execution_owner_id=owner, + execution_lease_expires_at=expiry, + ) + ) + session.add( + persistence_agent_run.AgentRun( + run_id='remote-queue', + workspace_id=WORKSPACE, + runner_id='runner', + status='queued', + queue_name='remote', + created_at=_dt(1), + ) + ) + await session.commit() + # A previous connection must not reintroduce unowned pending rows either. + snapshot = await service.get_inflight_snapshot(WORKSPACE, [{'source': 'pipeline', 'id': 'legacy'}]) + assert snapshot['active_total'] == 2 + assert {row['id'] for row in snapshot['items']} == {'long-running', 'remote-queue'} + async with AsyncSession(engine) as session: + statuses = (await session.execute(sa.select(Message.status))).scalars().all() + assert statuses == ['pending'] * 4 + + +@pytest.mark.asyncio +@pytest.mark.parametrize('source', ['all', 'pipeline']) +@pytest.mark.parametrize('statuses', [['running'], ['running', 'queued']]) +async def test_active_inflight_count_and_page_use_the_indexed_lease_range(engine, service, source, statuses): + import sqlalchemy as sa + + await _seed(engine) + async with engine.begin() as connection: + await connection.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-user') + .values(status='pending', message_content='historical payload ' * 8192) + ) + statements = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + if statement.startswith('SELECT '): + statements.append((statement, parameters)) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture) + try: + result = await service.get_executions( + WORKSPACE, statuses=statuses, source=source, include_summary=False, for_inflight=True + ) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture) + + assert result == {'items': [], 'total': 0, 'has_more': False} + assert len(statements) == 2 # Both COUNT and the sorted page must skip historical payloads. + async with engine.connect() as connection: + for statement, parameters in statements: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters)).all() + assert any( + 'ix_monitoring_messages_workspace_status_lease' in row[3] and 'execution_lease_expires_at>?' in row[3] + for row in plan + ), plan + + +@pytest.mark.asyncio +async def test_empty_inflight_snapshot_skips_linked_history_and_clears_tracking(engine, service, monkeypatch): + import sqlalchemy as sa + from langbot.pkg.utils.inflight import inflight_hub + + await _seed(engine) + monkeypatch.setattr(inflight_hub, 'run_workspaces', {'finished-run': WORKSPACE, 'other-run': 'other-workspace'}) + statements = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + statements.append(statement) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture) + try: + snapshot = await service.get_inflight_snapshot(WORKSPACE) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture) + + assert snapshot == {'items': [], 'active_total': 0, 'truncated': False} + assert inflight_hub.run_workspaces == {'other-run': 'other-workspace'} + assert not any('JOIN monitoring_messages ON' in query or 'FROM agent_run_event' in query for query in statements) + + +@pytest.mark.asyncio +async def test_inflight_never_loads_payloads_in_active_recent_or_terminal_rows(engine, service, monkeypatch): + import json + import sqlalchemy as sa + from langbot.pkg.entity.persistence.event_log import EventLog + + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + payload = json.dumps({'text': 'full input', 'attachment': 'x' * (2 * 1024 * 1024)}) + async with AsyncSession(engine) as session: + for message_id, status, timestamp in [ + ('active-message', 'pending', _dt(1)), + ('recent-message', 'success', now), + ('finished-message', 'success', _dt(1)), + ]: + session.add( + persistence_monitoring.MonitoringMessage( + id=message_id, + workspace_uuid=WORKSPACE, + timestamp=timestamp, + bot_id='bot', + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session', + message_content=payload, + variables=payload, + status=status, + level='info', + event_id=message_id, + run_id='linked-run' if status == 'pending' else None, + execution_owner_id='host' if status == 'pending' else None, + execution_lease_expires_at=now + datetime.timedelta(minutes=1) if status == 'pending' else None, + ) + ) + session.add( + EventLog( + event_id=message_id, + workspace_id=WORKSPACE, + event_type='message.received', + source='platform', + input_summary='small summary', + input_json=payload, + metadata_json=json.dumps({'platform': 'test', 'unused_payload': payload}), + ) + ) + for run_id in ['linked-run', 'standalone-run']: + session.add( + persistence_agent_run.AgentRun( + run_id=run_id, + workspace_id=WORKSPACE, + runner_id='runner', + status='running', + created_at=_dt(1), + authorization_json=payload, + metadata_json=json.dumps( + {'event_type': 'message.received', 'agent_name': 'Agent', 'input': {'text': payload}} + ), + ) + ) + await session.commit() + + linkage_queries = [] + + def reject_payload_reads(connection, cursor, statement, parameters, context, executemany): + for column in [ + 'monitoring_messages.message_content', + 'monitoring_messages.variables', + 'agent_run.authorization_json', + 'event_log.input_json', + ]: + assert column not in statement + if statement.startswith('SELECT agent_run.run_id,') and 'monitoring_messages.id AS message_id' in statement: + linkage_queries.append((statement, parameters)) + + deserialize = service._deserialize_json + + def bounded_metadata(raw): + assert not raw or len(raw) < 10000 + return deserialize(raw) + + monkeypatch.setattr(service, '_deserialize_json', bounded_metadata) + sa.event.listen(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + try: + snapshot = await service.get_inflight_snapshot(WORKSPACE, [{'source': 'pipeline', 'id': 'finished-message'}]) + rows = {row['id']: row for row in snapshot['items']} + assert rows['active-message']['input_preview'] == 'small summary' + assert rows['recent-message']['status_group'] == 'completed' + assert rows['finished-message']['status_group'] == 'completed' + assert rows['standalone-run']['target_name'] == 'Agent' + assert snapshot['active_total'] == 2 + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + assert linkage_queries + async with engine.connect() as connection: + for statement, parameters in linkage_queries: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters)).all() + message_lookups = [row[3] for row in plan if 'SEARCH monitoring_messages' in row[3]] + assert message_lookups and all('(id=?)' in lookup for lookup in message_lookups), plan + + @pytest.mark.asyncio async def test_inflight_snapshot_is_scoped_and_keeps_terminal_transition(engine, service): import sqlalchemy as sa From 4d90358da082d825b85cc7e4b5a42226fd049b1c Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 19:46:39 +0800 Subject: [PATCH 03/15] perf(monitoring): exclude payloads from session summaries Exclude message_content and variables from the message source used by conversation session summaries. These fields were carried through window sorting even though the resulting summaries did not use them. Preserve complete payloads for message detail queries. Add regression coverage for summary counts, session metadata, and full message retrieval. --- .../service/monitoring_execution_details.py | 12 +++-- .../api/service/test_monitoring_sessions.py | 46 +++++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py index edeb17862..735aa51be 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -100,7 +100,7 @@ def transcript_item(row): class ExecutionDetailsMixin: - def _session_message_source(self, workspace, bot_ids): + def _session_message_source(self, workspace, bot_ids, *, summary_only=False): """Combine delivered messages with ingress missing from the legacy ledger.""" message = models.MonitoringMessage conversation = sa.func.coalesce( @@ -148,7 +148,11 @@ class ExecutionDetailsMixin: 'event_id': EventLog.event_id, 'run_id': EventLog.run_id, } - columns = list(message.__table__.columns) + columns = [ + column + for column in message.__table__.columns + if not summary_only or column.name not in {'message_content', 'variables'} + ] stored = sa.select(*columns).where(message.workspace_uuid == workspace, message.bot_id.in_(bot_ids)) incoming = sa.select( *[values.get(column.name, sa.literal(None)).label(column.name) for column in columns] @@ -184,7 +188,9 @@ class ExecutionDetailsMixin: self, workspace, bot_ids, start_time, end_time, user_query, is_active, limit, offset ): """Read session summaries across processor types without mutating history.""" - messages = self._session_message_source(workspace, bot_ids) + # Window functions sort and retain their input rows. Carrying historical + # inline media through that sort can consume gigabytes for a summary. + messages = self._session_message_source(workspace, bot_ids, summary_only=True) key = [messages.c.workspace_uuid, messages.c.bot_id, messages.c.session_id] ranked = ( sa.select( diff --git a/tests/unit_tests/api/service/test_monitoring_sessions.py b/tests/unit_tests/api/service/test_monitoring_sessions.py index 868f51589..c10a3e98c 100644 --- a/tests/unit_tests/api/service/test_monitoring_sessions.py +++ b/tests/unit_tests/api/service/test_monitoring_sessions.py @@ -243,3 +243,49 @@ async def test_old_active_sessions_are_listed_exported_and_not_cleaned(service): assert count == 1 rows, total = await service.get_sessions(context()) assert total == 1 and rows[0]['bot_id'] == 'b' + + +async def test_bot_session_summary_never_sorts_inline_message_payloads(service): + engine = service.ap.persistence_mgr.get_db_engine() + payload = 'historical inline media ' * 16384 + now = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None) + async with engine.begin() as connection: + await connection.execute( + sa.insert(models.MonitoringMessage), + [ + dict( + id=f'message-{index}', + workspace_uuid='workspace', + session_id='group_42', + **resource('a'), + timestamp=now + dt.timedelta(seconds=index), + role='user' if index == 0 else 'assistant', + message_content=payload, + variables=payload, + status='success', + level='info', + user_id='user', + user_name='User', + platform='test', + ) + for index in range(2) + ], + ) + + def reject_payload_reads(connection, cursor, statement, parameters, context, executemany): + assert 'monitoring_messages.message_content' not in statement + assert 'monitoring_messages.variables' not in statement + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + try: + rows, total = await service.get_sessions(context('a'), bot_ids=['a']) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + + assert total == 1 + assert rows[0]['session_id'] == 'group_42' + assert rows[0]['message_count'] == 2 + assert rows[0]['user_name'] == 'User' + messages, total = await service.get_messages(context('a'), bot_ids=['a'], session_ids=['group_42']) + assert total == 2 + assert all(message['message_content'] == payload and message['variables'] == payload for message in messages) From 2ca9c2c7fb5f25e8899ddf248c3d3065c0c9d7bb Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 19:48:15 +0800 Subject: [PATCH 04/15] fix(pipeline): enforce group response rules after routing Stop treating an explicit Pipeline route as permission to bypass group response rules. Routed group messages must still satisfy the configured mention or prefix matchers. Preserve the bypass for verified interaction submissions that resume an existing run. Cover ordinary messages, mentions, prefixes, and interaction callbacks with regression tests. --- src/langbot/pkg/pipeline/resprule/resprule.py | 6 ++- tests/unit_tests/pipeline/test_resprule.py | 39 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/src/langbot/pkg/pipeline/resprule/resprule.py b/src/langbot/pkg/pipeline/resprule/resprule.py index 6e7fc369e..5f7d74524 100644 --- a/src/langbot/pkg/pipeline/resprule/resprule.py +++ b/src/langbot/pkg/pipeline/resprule/resprule.py @@ -37,8 +37,10 @@ class GroupRespondRuleCheckStage(stage.PipelineStage): if query.launcher_type.value != 'group': # 只处理群消息 return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) - # 通过路由规则明确指定的流水线,跳过群响应规则检查 - if query.variables and query.variables.get('_routed_by_rule', False): + # Verified interaction callbacks resume the original run without a new + # mention or prefix. Routing ordinary messages only selects a pipeline; + # it must not bypass that pipeline's group response rules. + if query.variables and query.variables.get('_interaction_submission'): return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) rules = query.pipeline_config['trigger']['group-respond-rules'] diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index 63dfbfd0d..3bee08d6f 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -138,3 +138,42 @@ async def test_atbot_rule_no_match(mock_app, sample_query): result = await atbot_rule.match(str(message_chain), message_chain, {}, sample_query) assert result.matching is False + + +@pytest.mark.asyncio +@pytest.mark.parametrize('routed', [False, True]) +@pytest.mark.parametrize('kind', ['plain', 'mention', 'prefix']) +async def test_routed_group_messages_still_obey_response_rules(mock_app, sample_query, routed, kind): + resprule, entities, _, _ = get_modules() + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + sample_query.variables = {'_routed_by_rule': routed} + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': True, 'prefix': ['!bot '], 'regexp': [], 'random': 0}} + } + parts = [platform_message.Plain(text='!bot hello' if kind == 'prefix' else 'hello')] + if kind == 'mention': + parts.insert(0, platform_message.At(target='999')) + sample_query.message_chain = platform_message.MessageChain(parts) + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + expected = entities.ResultType.INTERRUPT if kind == 'plain' else entities.ResultType.CONTINUE + assert result.result_type == expected + if kind != 'plain': + assert str(sample_query.message_chain) == 'hello' + + +@pytest.mark.asyncio +async def test_interaction_submission_does_not_require_another_mention(mock_app, sample_query): + resprule, entities, _, _ = get_modules() + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.variables = {'_routed_by_rule': True, '_interaction_submission': {'interaction_id': 'form-1'}} + stage = resprule.GroupRespondRuleCheckStage(mock_app) + stage.rule_matchers = [] + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + assert result.result_type == entities.ResultType.CONTINUE From 807de8d24ad2f8ad43bc1b6de472a6248559d160 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 19:48:25 +0800 Subject: [PATCH 05/15] fix(provider): sanitize replayed Groq message metadata Remove response-only finish_reason metadata when converting stored messages into subsequent model requests. Strip provider_specific_fields from Groq messages and tool calls to avoid strict request validation failures after a tool response. Preserve metadata required by other providers, including Gemini thought signatures, without mutating the original message entities. Cover streaming and non-streaming request construction with regression tests. --- .../modelmgr/requesters/litellmchat.py | 15 +++++ .../provider/test_provider_specific_fields.py | 58 +++++++++++++++++++ 2 files changed, 73 insertions(+) diff --git a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py index 300fb7551..04fd5d0df 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py +++ b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py @@ -836,6 +836,13 @@ class LiteLLMRequester(requester.ProviderAPIRequester): req_messages = [] for m in messages: msg_dict = m.dict(exclude_none=True) + provider_fields = msg_dict.get('provider_specific_fields') + if isinstance(provider_fields, dict): + # Stream completion metadata belongs to the response, never to + # replayed chat messages. Keep provider signatures for LiteLLM. + provider_fields.pop('finish_reason', None) + if not provider_fields: + msg_dict.pop('provider_specific_fields', None) content = msg_dict.get('content') if msg_dict.get('role') == 'assistant' and reasoning_family: @@ -871,6 +878,14 @@ class LiteLLMRequester(requester.ProviderAPIRequester): else: msg_dict.pop('provider_specific_fields', None) + if reasoning_family == 'groq': + # Groq validates message and tool-call fields strictly and does + # not accept LiteLLM's provider metadata envelope. + msg_dict.pop('provider_specific_fields', None) + for tool_call in msg_dict.get('tool_calls') or []: + tool_call.pop('provider_specific_fields', None) + tool_call.get('function', {}).pop('provider_specific_fields', None) + if isinstance(content, list): converted_parts = [] for part in content: diff --git a/tests/unit_tests/provider/test_provider_specific_fields.py b/tests/unit_tests/provider/test_provider_specific_fields.py index fe239f3ea..2607ad7d5 100644 --- a/tests/unit_tests/provider/test_provider_specific_fields.py +++ b/tests/unit_tests/provider/test_provider_specific_fields.py @@ -148,6 +148,64 @@ def test_convert_messages_preserves_message_provider_specific_fields(): assert out[0]['provider_specific_fields']['thought_signatures'] == ['sig1', 'sig2'] +@pytest.mark.parametrize('family', ['', 'openai', 'gemini', 'groq']) +def test_response_finish_reason_is_not_replayed(family): + req = _make_requester() + msg = provider_message.Message( + role='assistant', content='Hello', provider_specific_fields={'finish_reason': 'stop'} + ) + + assert req._convert_messages([msg], reasoning_family=family) == [{'role': 'assistant', 'content': 'Hello'}] + assert msg.provider_specific_fields == {'finish_reason': 'stop'} + + +def test_response_metadata_removal_preserves_gemini_signatures(): + req = _make_requester() + msg = provider_message.Message( + role='assistant', + content='Hello', + provider_specific_fields={'finish_reason': 'stop', 'thought_signatures': ['signature']}, + ) + + out = req._convert_messages([msg], reasoning_family='gemini') + + assert out[0]['provider_specific_fields'] == {'thought_signatures': ['signature']} + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +async def test_groq_completion_drops_provider_metadata_from_replayed_messages(stream): + req = _make_configured_requester() + req.requester_cfg['custom_llm_provider'] = 'groq' + model = _make_runtime_model() + model.model_entity.name = 'openai/gpt-oss-120b' + msg = provider_message.Message( + role='assistant', + content='Hello', + provider_specific_fields={'finish_reason': 'tool_calls', 'other_response_metadata': True}, + tool_calls=[ + provider_message.ToolCall( + id='call-1', + type='function', + function=provider_message.FunctionCall(name='search', arguments='{}'), + provider_specific_fields={'thought_signature': 'signature'}, + ) + ], + ) + + args = await req._build_completion_args(model, [msg], None, {}, stream=stream) + + assert args['model'] == 'groq/openai/gpt-oss-120b' + assert args['messages'] == [ + { + 'role': 'assistant', + 'content': 'Hello', + 'tool_calls': [{'id': 'call-1', 'type': 'function', 'function': {'name': 'search', 'arguments': '{}'}}], + } + ] + assert msg.provider_specific_fields['finish_reason'] == 'tool_calls' + + def test_normalize_stream_tool_calls_preserves_provider_specific_fields(): """Streaming tool calls should retain provider_specific_fields.""" req = _make_requester() From b3574efd95f1cfe131bdc99ca01afca1e301c136 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 20:03:24 +0800 Subject: [PATCH 06/15] fix(pipeline): preserve form callbacks and verify all group response rules Allow routed Dify form callbacks to resume without another mention or prefix while keeping ordinary messages subject to group response rules. Cover mention, prefix, regex, random probability, and combined rules. --- src/langbot/pkg/pipeline/resprule/resprule.py | 11 +- tests/unit_tests/pipeline/test_resprule.py | 111 +++++++++++++++++- 2 files changed, 115 insertions(+), 7 deletions(-) diff --git a/src/langbot/pkg/pipeline/resprule/resprule.py b/src/langbot/pkg/pipeline/resprule/resprule.py index 5f7d74524..3e8c16a96 100644 --- a/src/langbot/pkg/pipeline/resprule/resprule.py +++ b/src/langbot/pkg/pipeline/resprule/resprule.py @@ -37,10 +37,13 @@ class GroupRespondRuleCheckStage(stage.PipelineStage): if query.launcher_type.value != 'group': # 只处理群消息 return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) - # Verified interaction callbacks resume the original run without a new - # mention or prefix. Routing ordinary messages only selects a pipeline; - # it must not bypass that pipeline's group response rules. - if query.variables and query.variables.get('_interaction_submission'): + # Interaction submissions and routed platform form callbacks resume the + # original run without another mention or prefix. Ordinary routing only + # selects a pipeline and must still obey its group response rules. + if query.variables and ( + query.variables.get('_interaction_submission') + or (query.variables.get('_routed_by_rule') and query.variables.get('_dify_form_action')) + ): return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) rules = query.pipeline_config['trigger']['group-respond-rules'] diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index 3bee08d6f..4032ad6ff 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -167,13 +167,118 @@ async def test_routed_group_messages_still_obey_response_rules(mock_app, sample_ @pytest.mark.asyncio -async def test_interaction_submission_does_not_require_another_mention(mock_app, sample_query): +@pytest.mark.parametrize('routed', [False, True]) +@pytest.mark.parametrize( + 'overrides,text,mention,draw,matches,expected_text', + [ + pytest.param({'regexp': ['^hello$']}, 'hello', False, 0.9, True, 'hello', id='regexp-match'), + pytest.param({'regexp': ['^hello$']}, 'bye', False, 0.9, False, 'bye', id='regexp-miss'), + pytest.param({'regexp': ['^bye$', '^hello$']}, 'hello', False, 0.9, True, 'hello', id='second-regexp'), + pytest.param({'regexp': ['[']}, 'hello', False, 0.9, False, 'hello', id='invalid-regexp'), + pytest.param({'random': 0}, 'hello', False, 0.0, False, 'hello', id='random-disabled'), + pytest.param({'random': 1}, 'hello', False, 0.999, True, 'hello', id='random-always'), + pytest.param({'random': 0.25}, 'hello', False, 0.249, True, 'hello', id='random-hit'), + pytest.param({'random': 0.25}, 'hello', False, 0.25, False, 'hello', id='random-boundary'), + pytest.param({}, 'hello', True, 0.0, False, 'hello', id='mention-disabled'), + pytest.param({}, '!bot hello', False, 0.0, False, '!bot hello', id='prefix-disabled'), + pytest.param({'prefix': ['!other ', '!bot ']}, '!bot hello', False, 0.9, True, 'hello', id='second-prefix'), + pytest.param( + {'at': True, 'prefix': ['!bot '], 'regexp': ['^hello$'], 'random': 0.25}, + 'hello', + False, + 0.9, + True, + 'hello', + id='combined-regexp-match', + ), + pytest.param( + {'at': True, 'prefix': ['!bot '], 'regexp': ['^bye$'], 'random': 0.25}, + 'hello', + False, + 0.1, + True, + 'hello', + id='combined-random-fallback', + ), + pytest.param( + {'at': True, 'prefix': ['!bot '], 'regexp': ['^bye$'], 'random': 0.25}, + 'hello', + False, + 0.9, + False, + 'hello', + id='combined-all-miss', + ), + ], +) +async def test_group_response_rule_configuration( + mock_app, sample_query, monkeypatch, routed, overrides, text, mention, draw, matches, expected_text +): + resprule, entities, _, _ = get_modules() + random_rule = import_module('langbot.pkg.pipeline.resprule.rules.random') + monkeypatch.setattr(random_rule.random, 'random', lambda: draw) + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + sample_query.variables = {'_routed_by_rule': routed} + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': False, 'prefix': [], 'regexp': [], 'random': 0, **overrides}} + } + parts = [platform_message.Plain(text=text)] + if mention: + parts.insert(0, platform_message.At(target='999')) + sample_query.message_chain = platform_message.MessageChain(parts) + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + assert result.result_type == (entities.ResultType.CONTINUE if matches else entities.ResultType.INTERRUPT) + assert str(result.new_query.message_chain) == expected_text + + +@pytest.mark.asyncio +@pytest.mark.parametrize('callback_key', ['_interaction_submission', '_dify_form_action']) +async def test_interaction_callbacks_do_not_require_another_mention(mock_app, sample_query, callback_key): resprule, entities, _, _ = get_modules() sample_query.launcher_type = provider_session.LauncherTypes.GROUP - sample_query.variables = {'_routed_by_rule': True, '_interaction_submission': {'interaction_id': 'form-1'}} + sample_query.adapter.bot_account_id = '999' + submission = {'action_id': 'approve'} + sample_query.variables = {'_routed_by_rule': True, callback_key: submission} + sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='[Form Action: Approve]')]) + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': True, 'prefix': ['!bot '], 'regexp': [], 'random': 0}} + } stage = resprule.GroupRespondRuleCheckStage(mock_app) - stage.rule_matchers = [] + await stage.initialize(sample_query.pipeline_config) result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') assert result.result_type == entities.ResultType.CONTINUE + assert str(result.new_query.message_chain) == '[Form Action: Approve]' + assert result.new_query.variables[callback_key] == submission + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'variables', + [ + {'_routed_by_rule': True, '_dify_form_action': None}, + {'_routed_by_rule': True, '_dify_form_action': {}}, + {'_routed_by_rule': False, '_dify_form_action': {'action_id': 'approve'}}, + ], +) +async def test_incomplete_form_callback_does_not_bypass_response_rules(mock_app, sample_query, variables): + resprule, entities, _, _ = get_modules() + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + sample_query.variables = variables + sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='[Form Action: Approve]')]) + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': True, 'prefix': ['!bot '], 'regexp': [], 'random': 0}} + } + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + assert result.result_type == entities.ResultType.INTERRUPT From f12490bc7370c32d367d3aacd1e71b9ea714ac0f Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 17 Oct 2026 21:10:55 +0800 Subject: [PATCH 07/15] perf(monitoring): load execution details on demand - Load overview first and fetch secondary sections when expanded - Add a covering index and select conversation IDs before loading page payloads - Preserve full-detail API compatibility and add pagination and lazy-loading tests --- skills/skills/langbot-mcp-ops/SKILL.md | 3 + .../api/http/controller/groups/monitoring.py | 2 +- .../service/monitoring_execution_details.py | 55 ++++-- src/langbot/pkg/api/mcp/server.py | 2 +- .../pkg/entity/persistence/monitoring.py | 3 + .../versions/0040_conversation_page_index.py | 23 +++ .../api/service/test_monitoring_executions.py | 93 ++++++++++ .../test_conversation_page_index.py | 42 +++++ .../executions/ExecutionDetailSheet.tsx | 88 ++++++--- web/tests/e2e/monitoring-executions.spec.ts | 172 ++++++++++++++++++ 10 files changed, 448 insertions(+), 35 deletions(-) create mode 100644 src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py create mode 100644 tests/unit_tests/persistence/test_conversation_page_index.py diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index ae73d104c..ec47b8bb1 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -203,6 +203,9 @@ message or event identifier outside the current list page. The detail exposes `inputs`, `outputs` (generated content), `deliveries` (recorded platform sends), `events`, `llm_calls`, `tool_calls`, `errors`, `related`, and `conversation` in `pages`. Follow each section's `has_more` and `next_offset` independently. +Start with `section=overview` to load only inputs, outputs, deliveries and errors. +Request `conversation`, `related`, calls and events separately when needed; +omitting `section` retains the complete trace response for existing consumers. Conversation history supplies context; historical messages without explicit links must not be asserted to belong to the selected execution. Events without a processor run use `source=event`. All lookups remain Workspace-scoped. diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 5cbdf56c6..d6feeb2bc 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/src/langbot/pkg/api/http/controller/groups/monitoring.py @@ -129,7 +129,7 @@ class MonitoringRouterGroup(group.RouterGroup): permission=Permission.RESOURCE_VIEW, ) async def get_execution_detail(source: str, execution_id: str, request_context: RequestContext) -> str: - """Full trace of a single execution (agent run or pipeline query).""" + """Read a full trace, its initial overview, or one paginated section.""" try: detail = await self.ap.monitoring_service.get_execution_detail( request_context, diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py index 735aa51be..9c65f3f2c 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -408,6 +408,35 @@ class ExecutionDetailsMixin: 'next_offset': offset + min(len(rows), limit), } + async def _conversation_message_page(self, workspace, statement, offset, limit): + """Select a page from the covering index before loading its payloads.""" + message = models.MonitoringMessage + identities = list( + ( + await self._execution_query( + workspace, statement.with_only_columns(message.id).offset(offset).limit(limit + 1) + ) + ).scalars() + ) + selected = identities[:limit] + records = {} + if selected: + # Equality joins keep SQLite on the primary key even for multiple IDs. + page = sa.union_all(*[sa.select(sa.literal(identity).label('id')) for identity in selected]).subquery() + rows = await self._execution_rows( + workspace, + sa.select(message) + .select_from(page) + .join(message, message.id == page.c.id) + .where(message.workspace_uuid == workspace), + ) + records = {row.id: row for row in rows} + return { + 'items': [message_item(records[identity]) for identity in selected if identity in records], + 'has_more': len(identities) > limit, + 'next_offset': offset + len(selected), + } + def _unhandled_event_conditions(self, workspace, bot_ids=None, start_time=None, end_time=None): # Absorbed steering events belong to an existing run, too. conditions = [ @@ -590,8 +619,13 @@ class ExecutionDetailsMixin: 'tool_calls', 'errors', } - if section is not None and section not in sections: + if section is not None and section not in sections | {'overview'}: raise ValueError('Unknown execution section') + requested = ( + sections + if section is None + else ({'inputs', 'outputs', 'deliveries', 'errors'} if section == 'overview' else {section}) + ) run = message = event = None if source in {'agent', 'auto'}: @@ -758,8 +792,12 @@ class ExecutionDetailsMixin: pages = {} for name, (query, convert) in statements.items(): - if section is None or section == name: - pages[name] = await self._execution_page(workspace, query, convert, offset, limit) + if name in requested: + pages[name] = ( + await self._conversation_message_page(workspace, query, offset, limit) + if name == 'conversation' and message + else await self._execution_page(workspace, query, convert, offset, limit) + ) if run: meta = parsed(run.metadata_json, {}) for item in pages.get('inputs', {}).get('items', []): @@ -767,17 +805,12 @@ class ExecutionDetailsMixin: # Do not apply the root snapshot to later steering events. if item['id'] == run.event_id and not item.get('metadata', {}).get('input_event'): item['content'] = event_content(item['content'], meta.get('input_event')) - if message and (section is None or section == 'inputs'): + if message and 'inputs' in requested: # The original rich message is authoritative even when the event # log is absent or only retained a compact event summary. if offset == 0: pages['inputs']['items'].insert(0, message_item(message)) - if ( - run - and not pages.get('inputs', {}).get('items') - and offset == 0 - and (section is None or section == 'inputs') - ): + if run and not pages.get('inputs', {}).get('items') and offset == 0 and 'inputs' in requested: meta = parsed(run.metadata_json, {}) content = event_content(meta.get('input'), meta.get('input_event')) if content: @@ -785,7 +818,7 @@ class ExecutionDetailsMixin: {'id': run.event_id or run.run_id, 'content': content, 'timestamp_ms': epoch(run.created_at)} ] - if section is None or section == 'related': + if 'related' in requested: # Union of ledger executions and legacy pipeline roots. SQL paging # avoids silently dropping siblings on highly fanned-out events. related_runs = sa.select( diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index d7ad628d6..527bd00d8 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -313,7 +313,7 @@ class LangBotMCPServer: ) @mcp.tool( - description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Each section is independently paginated; follow pages[section].next_offset when has_more is true.' + description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Use section=overview for inputs, outputs, deliveries and errors without loading history; request other sections on demand. Omit section for the full trace. Each section is independently paginated; follow pages[section].next_offset when has_more is true.' ) async def get_monitoring_execution_detail( source: typing.Literal['auto', 'agent', 'pipeline', 'event'], diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index e1a92159e..b43ec6f63 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -40,6 +40,9 @@ class MonitoringMessage(Base): __table_args__ = ( sqlalchemy.Index('ix_monitoring_messages_workspace_timestamp', 'workspace_uuid', 'timestamp'), sqlalchemy.Index('ix_monitoring_messages_workspace_session', 'workspace_uuid', 'session_id'), + sqlalchemy.Index( + 'ix_monitoring_messages_conversation_page', 'workspace_uuid', 'bot_id', 'session_id', 'timestamp', 'id' + ), sqlalchemy.Index( 'ix_monitoring_messages_workspace_status_lease', 'workspace_uuid', diff --git a/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py b/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py new file mode 100644 index 000000000..1b62980cb --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py @@ -0,0 +1,23 @@ +"""Page conversation identities without reading historical message payloads.""" + +from alembic import op +import sqlalchemy as sa + +revision = '0040_conversation_page_index' +down_revision = '0039_execution_lease_indexes' +branch_labels = None +depends_on = None + +TABLE = 'monitoring_messages' +INDEX = 'ix_monitoring_messages_conversation_page' + + +def upgrade(): + # Fresh databases already receive this index from ORM metadata. + if INDEX not in {index['name'] for index in sa.inspect(op.get_bind()).get_indexes(TABLE)}: + op.create_index(INDEX, TABLE, ['workspace_uuid', 'bot_id', 'session_id', 'timestamp', 'id']) + + +def downgrade(): + if INDEX in {index['name'] for index in sa.inspect(op.get_bind()).get_indexes(TABLE)}: + op.drop_index(INDEX, table_name=TABLE) diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index a9469fcab..39587f245 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -423,6 +423,99 @@ async def test_execution_detail_for_agent_and_pipeline(engine, service): await service.get_execution_detail(WORKSPACE, 'pipeline', 'missing-message') +@pytest.mark.asyncio +async def test_detail_overview_does_not_query_collapsed_sections(engine, service, monkeypatch): + await _seed(engine) + page = service._execution_page + converters = [] + + async def capture(workspace, statement, convert, offset, limit): + converters.append(convert.__name__) + return await page(workspace, statement, convert, offset, limit) + + async def reject_history(*args, **kwargs): + pytest.fail('Collapsed conversation history must not be loaded') + + monkeypatch.setattr(service, '_execution_page', capture) + monkeypatch.setattr(service, '_conversation_message_page', reject_history) + detail = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user', section='overview') + + assert set(detail['pages']) == {'inputs', 'outputs', 'deliveries', 'errors'} + assert detail['pages']['inputs']['items'][0]['id'] == 'msg-user' + assert len(converters) == 4 + assert 'events' not in detail + + +@pytest.mark.asyncio +async def test_conversation_pages_identities_from_covering_index_before_payloads(engine, service): + import sqlalchemy as sa + + await _seed(engine) + message = persistence_monitoring.MonitoringMessage + payload = 'historical inline media ' * 4096 + async with engine.begin() as connection: + await connection.execute( + sa.insert(message), + [ + dict( + id=f'history-{index:02d}', + workspace_uuid=workspace, + bot_id=bot, + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session-1', + timestamp=_dt(12), + role='user', + message_content=payload, + variables=payload, + status='success', + level='info', + ) + for index, (workspace, bot) in enumerate( + [(WORKSPACE, 'bot-1')] * 5 + [(WORKSPACE, 'other-bot'), ('other-workspace', 'bot-1')] + ) + ], + ) + statements = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + if statement.startswith('SELECT'): + statements.append((statement, parameters)) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture) + try: + first = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user', section='conversation', limit=2) + second = await service.get_execution_detail( + WORKSPACE, 'pipeline', 'msg-user', section='conversation', limit=2, offset=2 + ) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture) + + page = first['pages']['conversation'] + assert [row['id'] for row in page['items']] == ['history-04', 'history-03'] + assert page['has_more'] and page['next_offset'] == 2 + assert all(row['content'] == payload for row in page['items']) + assert [row['id'] for row in second['pages']['conversation']['items']] == ['history-02', 'history-01'] + identities = [(sql, params) for sql, params in statements if sql.startswith('SELECT monitoring_messages.id \n')] + assert len(identities) == 2 + async with engine.connect() as connection: + for sql, params in identities: + assert 'message_content' not in sql and 'variables' not in sql + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + sql, params)).all() + assert any('COVERING INDEX ix_monitoring_messages_conversation_page' in row[3] for row in plan), plan + assert not any('TEMP B-TREE' in row[3] for row in plan), plan + payload_queries = [(sql, params) for sql, params in statements if 'JOIN monitoring_messages ON' in sql] + assert len(payload_queries) == 2 + assert {value for value in payload_queries[0][1] if str(value).startswith('history-')} == { + 'history-04', + 'history-03', + } + for sql, params in payload_queries: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + sql, params)).all() + assert any('sqlite_autoindex_monitoring_messages_1 (id=?)' in row[3] for row in plan), plan + + @pytest.mark.asyncio async def test_explicit_deliveries_context_and_tenant_isolation(engine, service): import sqlalchemy as sa diff --git a/tests/unit_tests/persistence/test_conversation_page_index.py b/tests/unit_tests/persistence/test_conversation_page_index.py new file mode 100644 index 000000000..aad0cb52e --- /dev/null +++ b/tests/unit_tests/persistence/test_conversation_page_index.py @@ -0,0 +1,42 @@ +"""Conversation paging index is safe for existing and fresh databases.""" + +import importlib + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage + + +def test_conversation_index_migration_preserves_history_and_matches_metadata(): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0040_conversation_page_index') + engine = sa.create_engine('sqlite:///:memory:') + metadata = sa.MetaData() + table = sa.Table( + 'monitoring_messages', + metadata, + *[ + sa.Column(name, sa.String, primary_key=name == 'id') + for name in ('id', 'workspace_uuid', 'bot_id', 'session_id', 'timestamp', 'message_content') + ], + ) + sa.Index('existing_index', table.c.workspace_uuid) + try: + with engine.begin() as connection: + metadata.create_all(connection) + connection.execute(table.insert().values(id='history', message_content='inline media')) + with Operations.context(MigrationContext.configure(connection)): + migration.upgrade() + migration.upgrade() + indexes = { + index['name']: index['column_names'] for index in sa.inspect(connection).get_indexes(table.name) + } + expected = next(index for index in MonitoringMessage.__table__.indexes if index.name == migration.INDEX) + assert indexes[migration.INDEX] == [column.name for column in expected.columns] + migration.downgrade() + assert {index['name'] for index in sa.inspect(connection).get_indexes(table.name)} == {'existing_index'} + assert connection.scalar(sa.select(table.c.message_content)) == 'inline media' + migration.upgrade() + finally: + engine.dispose() diff --git a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx index 70d226487..bba19cb48 100644 --- a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx +++ b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx @@ -87,10 +87,14 @@ export default function ExecutionDetailSheet({ const [detail, setDetail] = useState(null); const [loading, setLoading] = useState(false); const [error, setError] = useState(false); - const [sectionError, setSectionError] = useState( - null, + const [sectionError, setSectionError] = useState< + Partial> + >({}); + const [busy, setBusy] = useState>>( + {}, ); - const [busy, setBusy] = useState(null); + const pendingSections = useRef(new Set()); + const expandedSections = useRef(new Set()); const [refresh, setRefresh] = useState(0); const request = useRef(0); const identity = `${workspace}:${row?.source}:${row?.id}`; @@ -106,20 +110,43 @@ export default function ExecutionDetailSheet({ if (requestedIdentity.current !== identity) { setDetail(null); requestedIdentity.current = identity; + expandedSections.current.clear(); } setError(false); - setSectionError(null); - setBusy(null); + setSectionError({}); + setBusy({}); + pendingSections.current.clear(); if (!id || !source) { setLoading(false); return; } setLoading(true); - backendClient - .getExecutionDetail(source, id) - .then((result) => { + const expanded = [...expandedSections.current].filter( + (section) => section !== 'errors', + ); + Promise.all([ + backendClient.getExecutionDetail(source, id, { section: 'overview' }), + ...expanded.map((section) => + backendClient + .getExecutionDetail(source, id, { section }) + .catch(() => null), + ), + ]) + .then(([result, ...sections]) => { if (version !== request.current) return; - setDetail(result); + if (!result) return; + const pages = { ...result.pages }; + const failures: Partial> = {}; + sections.forEach((loaded, index) => { + const section = expanded[index]; + if (loaded) Object.assign(pages, loaded.pages); + else failures[section] = true; + }); + setDetail((previous) => ({ + ...result, + pages: { ...previous?.pages, ...pages }, + })); + setSectionError((previous) => ({ ...previous, ...failures })); setLoadedIdentity(identity); }) .catch(() => { @@ -144,10 +171,11 @@ export default function ExecutionDetailSheet({ return () => clearInterval(timer); }, [status, visible]); const loadMore = async (section: ExecutionSection) => { - if (!visible || !row || busy) return; + if (!visible || !row || pendingSections.current.has(section)) return; const version = request.current; - setBusy(section); - setSectionError(null); + pendingSections.current.add(section); + setBusy((previous) => ({ ...previous, [section]: true })); + setSectionError((previous) => ({ ...previous, [section]: false })); try { const result = await backendClient.getExecutionDetail( row.source, @@ -174,9 +202,13 @@ export default function ExecutionDetailSheet({ }, ); } catch { - if (version === request.current) setSectionError(section); + if (version === request.current) + setSectionError((previous) => ({ ...previous, [section]: true })); } finally { - if (version === request.current) setBusy(null); + if (version === request.current) { + pendingSections.current.delete(section); + setBusy((previous) => ({ ...previous, [section]: false })); + } } }; const renderItem = (section: ExecutionSection, item: ExecutionDetailItem) => { @@ -298,9 +330,12 @@ export default function ExecutionDetailSheet({ }; const renderSection = (section: ExecutionSection, primary = false) => { const page = visible?.pages[section]; + if (primary && !page) return null; if ( - !page || - (!primary && page.items.length === 0 && section !== 'conversation') + !primary && + page?.items.length === 0 && + section !== 'conversation' && + !sectionError[section] ) return null; const body = ( @@ -310,26 +345,29 @@ export default function ExecutionDetailSheet({ {t('monitoring.unified.contextHint')}

)} - {page.items.length === 0 && ( + {!page && !sectionError[section] && ( + + )} + {page?.items.length === 0 && (

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

)} - {page.items.map((item, i) => ( + {page?.items.map((item, i) => (
{renderItem(section, item)}
))} - {sectionError === section && ( + {sectionError[section] && ( loadMore(section)} /> )} - {page.has_more && ( + {page?.has_more && (