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 && (