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)