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.
This commit is contained in:
BiFangKNT
2026-10-17 19:46:39 +08:00
parent d663331889
commit 4d90358da0
2 changed files with 55 additions and 3 deletions
@@ -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(
@@ -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)