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
This commit is contained in:
BiFangKNT
2026-10-17 21:10:55 +08:00
parent b3574efd95
commit f12490bc73
10 changed files with 448 additions and 35 deletions
@@ -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,
@@ -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(
+1 -1
View File
@@ -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'],
@@ -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',
@@ -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)