fix(monitoring): persist session summaries to avoid history scans

- Maintain counts and display snapshots in source transactions with replay-safe deduplication.
- Read session lists exclusively from summaries and remove execution-only filters.
- Automatically initialize historical summaries in bounded, resumable batches during OSS startup.
- Add behavioral contracts covering accounting, retention, query cost and PostgreSQL concurrency.

Verified with the required session_summary suite and real container upgrade startup.

.github/workflows/run-tests.yml

@@ -12,6 +12,8 @@ on:
       - 'run_tests.sh'
       - 'scripts/test-*.sh'
       - 'pytest.ini'
+      - 'AGENTS.md'
+      - 'ARCHITECTURE.md'
   push:
     branches:
       - master
@@ -157,6 +159,19 @@ jobs:
           path: postgres-contract-results.xml
           if-no-files-found: ignore
 
+      - name: Run required session summary contracts
+        run: |
+          uv run --no-sync pytest --required-suite session_summary \
+            -q --tb=short -rs --junitxml=session-summary-results.xml
+
+      - name: Upload session summary contract results
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: session-summary-results
+          path: session-summary-results.xml
+          if-no-files-found: ignore
+
   e2e:
     name: E2E Startup Tests
     runs-on: ubuntu-latest

AGENTS.md

@@ -89,6 +89,7 @@ Config keys to verify in `data/config.yaml` / `src/langbot/templates/config.yaml
 - New platform behavior belongs in platform adapters only for platform translation; pipeline/business logic belongs in `pkg/pipeline/` or services.
 - User-facing strings must support i18n; follow the backend language rules below for server-generated content.
 - Conversation payload columns, including new ones, must use `MediaReferenceText` and persist image references only. See the [media persistence contract](ARCHITECTURE.md#persistence-and-configuration).
+- Interactive session lists use `SessionSummaryReader` to read only `monitoring_sessions`; `SessionSummaryWriter` maintains summaries in the source transaction. Historical initialization completes before writers start, never in requests or application background tasks. Follow the [conversation summary contract](ARCHITECTURE.md#conversation-summary-boundary) and keep `uv run pytest --required-suite session_summary` passing.
 - Code comments and docstrings must be English.
 - Keep compatibility and security in mind; LangBot is used in both self-hosted/community and toB deployments.
 - Commit message format: `<type>(<scope>): <subject>`.

ARCHITECTURE.md

@@ -276,6 +276,26 @@ Configuration starts from `src/langbot/templates/config.yaml` and is generated i
 
 Storage maintenance applies `storage.media_cache.retention_days` to both CAS files and historical inline images. Inline image expiry uses the owning record's creation time and preserves text, URLs, and record identity. It does not backfill CAS. `pkg/storage/inline_media_cleanup.py` processes one payload row at a time, with a per-table limit of 100 indexed candidates or 8 MiB of JSON characters per maintenance pass (one oversized row may exceed the latter limit). Run events lack a Workspace column, so their candidate cursor also advances past foreign or missing records; payloads are loaded only after checking Workspace ownership. Durable Workspace metadata cursors and timestamp/identity indexes keep completed history out of subsequent scans; newly written history must retain its actual creation timestamp. Updates and cursor advancement commit together. Invalid historical JSON is preserved. SQLite pages become reusable after payload removal; the database file is not automatically vacuumed by this task.
 
+### Conversation summary boundary
+
+`MonitoringService` delegates all interactive session lists to `SessionSummaryReader` in `pkg/persistence/`. The reader selects public columns from `monitoring_sessions` without history joins or repairs, keeping list cost independent of historical payload size. Sessions describe conversations, not executions; API filter semantics are documented in [LangBot MCP Operations](skills/skills/langbot-mcp-ops/SKILL.md#unified-execution-monitoring).
+
+`SessionSummaryWriter` updates summaries in the source write's Workspace transaction. Each distinct message record counts once, including assistant and failure records. A platform `message.received` event contributes only while no non-assistant message matches its event ID or legacy message ID. Replays add no credit; late links retire ingress credit once, including across group/topic sessions. Observed start/activity timestamps never shrink the session's time range. Live writes and offline initialization share the same source mapping.
+
+All display fields come from one source snapshot, including empty values. Missing names display the user ID instead of inheriting another record's name. Non-assistant sources outrank assistant sources, followed by timestamp and binary source-key order. Processor filters match the selected snapshot. Immutable message winners fold into `base_snapshot`; uncovered ingress retains a replaceable `candidate_snapshot` on its receipt. Covering or transferring ingress restores the remaining winner atomically, without reading history.
+
+Receipts and event anchors are Workspace-scoped and expire in bounded batches under history retention. Expiry preserves counters and seals any uncovered ingress snapshot into the base in the same transaction. Summaries are deleted only after their receipts are gone. Deduplication and late linking are guaranteed within this retention window.
+
+Migration `0040_session_summaries` marks historical Workspaces as pending. OSS startup automatically completes `SessionSummaryBackfill` before starting HTTP or message producers, then continues in the same process; no manual backfill or additional container restart is needed. Batches and progress commit together, so an interrupted startup resumes from its cursor. Initialization is bounded, never selects message bodies or variables, and preserves historical counts as a lower bound when source history has expired. Invalid or oversized required metadata stops startup for diagnosis. Fresh databases and completed initialization skip history scanning; pending source writes remain blocked. There is no request-time or background repair.
+
+Cloud runtimes retain the initialization gate because other instances may write concurrently. For Cloud deployment or manual maintenance, stop **all** writers, back up the database, apply the migration, and run the command below for each pending Workspace before starting services. Set the named environment variable to a `sqlite+aiosqlite` or `postgresql+asyncpg` database URL; SQLite must point to an existing file. The connection must have access to the migrated schema and Workspace.
+
+```bash
+uv run python -m langbot.pkg.persistence.session_summary_backfill \
+  --workspace WORKSPACE_UUID --writers-stopped \
+  --database-url-env LANGBOT_SUMMARY_DATABASE_URL
+```
+
 ## Frontend
 
 The frontend lives in `web/` and is a Vite SPA using React Router 7, shadcn/ui, Tailwind CSS, and pnpm. It is not Next.js, despite some historical filenames.

pytest.ini

@@ -30,6 +30,7 @@ addopts =
 markers =
     contract(name): stable behavior obligation for required suites
     postgres: required postgres execution suite
+    session_summary: required session summary boundary and accounting suite
     local_agent: required local_agent execution suite
     security: required security execution suite
     compiled_cli: required compiled_cli execution suite

skills/skills/langbot-mcp-ops/SKILL.md

@@ -210,11 +210,17 @@ 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.
 
-Monitoring record filters accept `mode` (`all`, `real`, `debug`) and
+Monitoring record filters, except for `kind="sessions"`, accept `mode` (`all`, `real`, `debug`) and
 `execution_statuses` (normalized execution statuses). These select the owning
 execution, not the individual model/tool call outcome. Calls without a recorded
 execution link are excluded when an execution filter is active.
 
+Session lists support `bot_ids`, `pipeline_ids`, `start_time`, `end_time`,
+`user_query`, `is_active`, and pagination. For execution status or real/debug
+filters, use `get_monitoring_executions`; HTTP `/monitoring/sessions` rejects
+them. In HTTP `/monitoring/data`, `sessions` and `totalCount.sessions` use only
+bot/pipeline/time filters; execution filters apply to the other records.
+
 ### Workspace message language
 
 `get_workspace_message_language` reads the current Workspace's backend language

src/langbot/pkg/agent/runner/event_log_store.py

@@ -8,6 +8,7 @@ import typing
 import uuid
 
 import sqlalchemy
+from sqlalchemy.dialects import postgresql, sqlite
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
 from sqlalchemy.orm import sessionmaker
 
@@ -70,6 +71,7 @@ class EventLogStore:
         runner_id: str | None = None,
         event_time: datetime.datetime | None = None,
         metadata: dict[str, typing.Any] | None = None,
+        session: AsyncSession | None = None,
     ) -> str:
         """Append an event to the event log.
 
@@ -93,6 +95,7 @@ class EventLogStore:
             runner_id: Runner ID processing this event
             event_time: When the event occurred
             metadata: Additional metadata
+            session: Optional owning transaction; the caller commits or rolls back
 
         Returns:
             The event_id
@@ -104,60 +107,70 @@ class EventLogStore:
         if input_summary and len(input_summary) > self.MAX_INPUT_SUMMARY_LENGTH:
             input_summary = input_summary[: self.MAX_INPUT_SUMMARY_LENGTH - 3] + '...'
 
-        async with self._session_factory() as session:
-            event = EventLog(
-                event_id=event_id,
-                event_type=event_type,
-                event_time=as_naive_utc(event_time),
-                source=source,
-                bot_id=bot_id,
-                workspace_id=workspace_id,
-                conversation_id=conversation_id,
-                thread_id=thread_id,
-                actor_type=actor_type,
-                actor_id=actor_id,
-                actor_name=actor_name,
-                subject_type=subject_type,
-                subject_id=subject_id,
-                input_summary=input_summary,
-                input_json=json.dumps(input_json) if input_json else None,
-                raw_ref=raw_ref,
-                run_id=run_id,
-                runner_id=runner_id,
-                metadata_json=json.dumps(metadata) if metadata else None,
-                created_at=as_naive_utc(_utc_now()),
-            )
-            session.add(event)
-            try:
-                await session.commit()
-            except sqlalchemy.exc.IntegrityError:
-                await session.rollback()
-                existing = (
-                    (await session.execute(sqlalchemy.select(EventLog).where(EventLog.event_id == event_id)))
-                    .scalars()
-                    .first()
-                )
-                if existing is None or existing.workspace_id != workspace_id or existing.bot_id != bot_id:
-                    raise
-                # Complete the ingress snapshot once a runner claims it. A
-                # second subscriber must not replace another run's identity.
-                for field, value in {
-                    'run_id': run_id,
-                    'runner_id': runner_id,
-                    'conversation_id': conversation_id,
-                    'thread_id': thread_id,
-                    'actor_id': actor_id,
-                    'actor_name': actor_name,
-                }.items():
-                    if getattr(existing, field) is None and value is not None:
-                        setattr(existing, field, value)
-                if input_json and ('contents' in input_json or 'attachments' in input_json):
-                    existing.input_json = json.dumps(input_json)
-                    existing.input_summary = input_summary
-                await session.commit()
-
+        values = dict(
+            event_id=event_id,
+            event_type=event_type,
+            event_time=as_naive_utc(event_time),
+            source=source,
+            bot_id=bot_id,
+            workspace_id=workspace_id,
+            conversation_id=conversation_id,
+            thread_id=thread_id,
+            actor_type=actor_type,
+            actor_id=actor_id,
+            actor_name=actor_name,
+            subject_type=subject_type,
+            subject_id=subject_id,
+            input_summary=input_summary,
+            input_json=json.dumps(input_json) if input_json else None,
+            raw_ref=raw_ref,
+            run_id=run_id,
+            runner_id=runner_id,
+            metadata_json=json.dumps(metadata) if metadata else None,
+            created_at=as_naive_utc(_utc_now()),
+        )
+        if session is not None:
+            await self._append_in_transaction(session, values, input_json)
+        else:
+            async with self._session_factory() as owned:
+                await self._append_in_transaction(owned, values, input_json)
+                await owned.commit()
         return event_id
 
+    async def _append_in_transaction(self, session, values, input_json):
+        insert = {'postgresql': postgresql.insert, 'sqlite': sqlite.insert}[self.engine.dialect.name]
+        result = await session.execute(
+            insert(EventLog).values(**values).on_conflict_do_nothing(index_elements=['event_id'])
+        )
+        if result.rowcount:
+            return
+        existing = (
+            await session.execute(
+                sqlalchemy.select(EventLog).where(
+                    EventLog.event_id == values['event_id'],
+                )
+            )
+        ).scalar_one_or_none()
+        if existing is None or existing.workspace_id != values['workspace_id'] or existing.bot_id != values['bot_id']:
+            raise ValueError('An event ID cannot be reused across Workspaces or bots')
+        # Only fill missing identities; another subscriber cannot replace a claimed run.
+        updates = {
+            field: sqlalchemy.func.coalesce(getattr(EventLog, field), values[field])
+            for field in ('run_id', 'runner_id', 'conversation_id', 'thread_id', 'actor_id', 'actor_name')
+            if values[field] is not None
+        }
+        if input_json and ('contents' in input_json or 'attachments' in input_json):
+            updates.update(input_json=values['input_json'], input_summary=values['input_summary'])
+        if updates:
+            await session.execute(
+                sqlalchemy.update(EventLog)
+                .where(
+                    EventLog.event_id == values['event_id'],
+                    EventLog.workspace_id == values['workspace_id'],
+                )
+                .values(**updates)
+            )
+
     async def get_event(
         self,
         event_id: str,

src/langbot/pkg/api/http/controller/groups/monitoring.py

@@ -343,6 +343,10 @@ class MonitoringRouterGroup(group.RouterGroup):
         )
         async def get_sessions(request_context: RequestContext) -> str:
             """Get session information"""
+            if quart.request.args.getlist('status') or quart.request.args.get('mode', 'all') != 'all':
+                return self.http_status(
+                    400, 'invalid_request', 'Session lists do not support execution status or mode filters'
+                )
             # Parse query parameters
             bot_ids = quart.request.args.getlist('botId')
             pipeline_ids = quart.request.args.getlist('pipelineId')
@@ -428,7 +432,7 @@ class MonitoringRouterGroup(group.RouterGroup):
 
         @self.route('/data', methods=['GET'], permission=Permission.RESOURCE_VIEW)
         async def get_all_data(request_context: RequestContext) -> str:
-            """Get all monitoring data in a single request"""
+            """Read monitoring data; session summaries use only bot, pipeline and time filters."""
             execution_filters = {
                 'execution_statuses': quart.request.args.getlist('status') or None,
                 'mode': quart.request.args.get('mode', 'all'),
@@ -491,10 +495,9 @@ class MonitoringRouterGroup(group.RouterGroup):
                 offset=0,
             )
 
-            # Get sessions
+            # Sessions have conversation scope, independent of execution status/mode.
             sessions, sessions_total = await self.ap.monitoring_service.get_sessions(
                 request_context,
-                **execution_filters,
                 bot_ids=bot_ids if bot_ids else None,
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,

src/langbot/pkg/api/http/service/monitoring.py

@@ -5,6 +5,7 @@ import uuid
 import datetime
 import functools
 import json
+from contextlib import asynccontextmanager
 import sqlalchemy
 from sqlalchemy.dialects import postgresql as postgresql_dialect
 from sqlalchemy.dialects import sqlite as sqlite_dialect
@@ -20,6 +21,12 @@ from ..authz import WorkspaceRequiredError
 from ..context import ExecutionContext
 from .tenant import TenantContext, require_workspace_uuid
 from .monitoring_execution_details import ExecutionDetailsMixin
+from ....persistence.session_summary_contract import (
+    SessionListFilter,
+    SessionSummaryReader,
+)
+from ....persistence.session_summary_reader import SqlSessionSummaryReader, SUMMARY_FIELDS
+from ....persistence.session_summary_writer import SessionSummaryWriter, message_summary_change
 
 
 _DEFAULT_MONITORING_PAGE_ROWS = 1000
@@ -147,6 +154,28 @@ class MonitoringService(ExecutionDetailsMixin):
 
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
+        self._session_list: SessionSummaryReader = SqlSessionSummaryReader(self._summary_session)
+
+    @asynccontextmanager
+    async def _summary_session(self, workspace: str):
+        """Join the owning Workspace transaction, including source writes."""
+        async with self.ap.persistence_mgr.tenant_uow(workspace) as uow:
+            yield uow.session
+
+    def _summary_writer(self):
+        return SessionSummaryWriter(self.ap.persistence_mgr.get_db_engine().dialect.name)
+
+    @asynccontextmanager
+    async def _summary_write_session(self, workspace: str):
+        from ....persistence.session_summary_backfill import require_initialized
+
+        async with self._summary_session(workspace) as session:
+            await require_initialized(session, workspace)
+            yield session
+
+    def _serialize_session(self, row):
+        values = self.ap.persistence_mgr.serialize_model(persistence_monitoring.MonitoringSession, row)
+        return {field: values[field] for field in SUMMARY_FIELDS}
 
     def _configured_query_limit(self, name: str, default: int, hard_max: int) -> int:
         config = (
@@ -292,6 +321,12 @@ class MonitoringService(ExecutionDetailsMixin):
         )
 
         tables_and_columns: list[tuple[str, type, sqlalchemy.Column, sqlalchemy.Column]] = [
+            (
+                'monitoring_session_receipts',
+                persistence_monitoring.MonitoringSessionReceipt,
+                persistence_monitoring.MonitoringSessionReceipt.timestamp,
+                persistence_monitoring.MonitoringSessionReceipt.source_key,
+            ),
             (
                 'monitoring_messages',
                 persistence_monitoring.MonitoringMessage,
@@ -359,7 +394,8 @@ class MonitoringService(ExecutionDetailsMixin):
         else:
             deleted_counts = await delete_records()
 
-        if sum(deleted_counts.values()) > 0:
+        # Receipt expiry alone must not trigger database-wide VACUUM work.
+        if any(count for table, count in deleted_counts.items() if table != 'monitoring_session_receipts'):
             await self._release_sqlite_space()
 
         return deleted_counts
@@ -375,15 +411,28 @@ class MonitoringService(ExecutionDetailsMixin):
         max_batches: int,
     ) -> int:
         workspace_uuid = self._require_write_context(context)
+        if model_cls is persistence_monitoring.MonitoringSessionReceipt:
+            return await self._expire_session_receipts(workspace_uuid, cutoff, batch_size, max_batches)
         deleted_total = 0
 
         for _batch_number in range(max_batches):
 
             async def delete_batch() -> tuple[int, int]:
                 key_columns = list(model_cls.__table__.primary_key.columns)
+                conditions = [model_cls.workspace_uuid == workspace_uuid, ts_column < cutoff]
+                if model_cls is persistence_monitoring.MonitoringSession:
+                    receipt = persistence_monitoring.MonitoringSessionReceipt
+                    conditions.append(
+                        ~sqlalchemy.exists().where(
+                            receipt.workspace_uuid == workspace_uuid,
+                            receipt.bot_id == model_cls.bot_id,
+                            receipt.session_id == model_cls.session_id,
+                        )
+                    )
                 select_result = await self.ap.persistence_mgr.execute_async(
                     sqlalchemy.select(*key_columns)
-                    .where(model_cls.workspace_uuid == workspace_uuid, ts_column < cutoff)
+                    .where(*conditions)
+                    .order_by(ts_column, *key_columns)
                     .limit(batch_size)
                 )
                 pk_values = [tuple(row) for row in select_result.all()]
@@ -392,9 +441,8 @@ class MonitoringService(ExecutionDetailsMixin):
 
                 delete_result = await self.ap.persistence_mgr.execute_async(
                     sqlalchemy.delete(model_cls).where(
-                        model_cls.workspace_uuid == workspace_uuid,
+                        *conditions,
                         sqlalchemy.tuple_(*key_columns).in_(pk_values),
-                        ts_column < cutoff,
                     )
                 )
                 return len(pk_values), int(delete_result.rowcount or 0)
@@ -414,6 +462,47 @@ class MonitoringService(ExecutionDetailsMixin):
 
         return deleted_total
 
+    async def _expire_session_receipts(self, workspace, cutoff, batch_size, max_batches):
+        receipt = persistence_monitoring.MonitoringSessionReceipt
+        expired = [receipt.workspace_uuid == workspace, receipt.timestamp < cutoff]
+        deleted = 0
+        for _ in range(max_batches):
+            async with self._summary_session(workspace) as session:
+                rows = (
+                    (
+                        await session.execute(
+                            sqlalchemy.select(receipt.bot_id, receipt.source_key)
+                            .where(*expired)
+                            .order_by(receipt.timestamp, receipt.bot_id, receipt.source_key)
+                            .limit(batch_size)
+                        )
+                    )
+                    .mappings()
+                    .all()
+                )
+                event_keys = [
+                    (row['bot_id'], row['source_key']) for row in rows if row['source_key'].startswith('event:')
+                ]
+                deleted += await self._summary_writer().expire_ingress(session, workspace, cutoff, event_keys)
+            # A message receipt can be locked by late linking, which next locks
+            # summaries. Never hold a message lock while sealing ingress.
+            message_keys = [
+                (row['bot_id'], row['source_key']) for row in rows if not row['source_key'].startswith('event:')
+            ]
+            if message_keys:
+                async with self._summary_session(workspace) as session:
+                    result = await session.execute(
+                        sqlalchemy.delete(receipt).where(
+                            *expired,
+                            ~receipt.source_key.startswith('event:'),
+                            sqlalchemy.tuple_(receipt.bot_id, receipt.source_key).in_(message_keys),
+                        )
+                    )
+                    deleted += result.rowcount
+            if len(rows) < batch_size:
+                break
+        return deleted
+
     async def _release_sqlite_space(self) -> None:
         database_type = self.ap.instance_config.data.get('database', {}).get('use', 'sqlite')
         if database_type != 'sqlite':
@@ -587,9 +676,12 @@ class MonitoringService(ExecutionDetailsMixin):
             'parent_message_id': parent_message_id,
         }
 
-        await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.insert(persistence_monitoring.MonitoringMessage).values(message_data)
-        )
+        async with self._summary_write_session(workspace_uuid) as session:
+            await session.execute(sqlalchemy.insert(persistence_monitoring.MonitoringMessage).values(message_data))
+            await self._summary_writer().apply_once(
+                session,
+                message_summary_change(workspace_uuid, message_data),
+            )
 
         return message_id
 
@@ -745,96 +837,6 @@ class MonitoringService(ExecutionDetailsMixin):
 
         return call_id
 
-    @_workspace_transaction
-    async def record_session_start(
-        self,
-        context: ExecutionContext,
-        session_id: str,
-        bot_id: str,
-        bot_name: str,
-        pipeline_id: str,
-        pipeline_name: str,
-        platform: str | None = None,
-        user_id: str | int | None = None,
-        user_name: str | None = None,
-    ) -> None:
-        """Record a new session"""
-        workspace_uuid = self._require_write_context(context)
-        session_data = {
-            'workspace_uuid': workspace_uuid,
-            'session_id': session_id,
-            'bot_id': bot_id,
-            'bot_name': bot_name,
-            'pipeline_id': pipeline_id,
-            'pipeline_name': pipeline_name,
-            'message_count': 1,
-            'start_time': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
-            'last_activity': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
-            'is_active': True,
-            'platform': platform,
-            'user_id': _normalize_user_id(user_id),
-            'user_name': user_name,
-        }
-
-        model = persistence_monitoring.MonitoringSession
-        dialect = self.ap.persistence_mgr.get_db_engine().dialect.name
-        insert = postgresql_dialect.insert if dialect == 'postgresql' else sqlite_dialect.insert
-        statement = insert(model).values(session_data)
-        await self.ap.persistence_mgr.execute_async(
-            statement.on_conflict_do_update(
-                index_elements=['workspace_uuid', 'bot_id', 'session_id'],
-                set_={
-                    'message_count': model.message_count + 1,
-                    'last_activity': statement.excluded.last_activity,
-                    'pipeline_id': statement.excluded.pipeline_id,
-                    'pipeline_name': statement.excluded.pipeline_name,
-                },
-            )
-        )
-
-    @_workspace_transaction
-    async def update_session_activity(
-        self,
-        context: ExecutionContext,
-        session_id: str,
-        pipeline_id: str | None = None,
-        pipeline_name: str | None = None,
-        bot_id: str | None = None,
-    ) -> bool:
-        """Update session last activity time and increment message count.
-
-        Also updates pipeline info if the bot's pipeline has changed.
-
-        Returns:
-            True if session was found and updated, False if session doesn't exist.
-        """
-        workspace_uuid = self._require_write_context(context)
-        bot_id = bot_id if bot_id is not None else context.bot_uuid
-        if not bot_id:
-            raise ValueError('Session activity requires a bot_id')
-        update_values = {
-            'last_activity': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
-            'message_count': persistence_monitoring.MonitoringSession.message_count + 1,
-        }
-
-        # Update pipeline info if provided (handles pipeline switch)
-        if pipeline_id is not None:
-            update_values['pipeline_id'] = pipeline_id
-        if pipeline_name is not None:
-            update_values['pipeline_name'] = pipeline_name
-
-        result = await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.update(persistence_monitoring.MonitoringSession)
-            .where(
-                persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid,
-                persistence_monitoring.MonitoringSession.session_id == session_id,
-                persistence_monitoring.MonitoringSession.bot_id == bot_id,
-            )
-            .values(update_values)
-        )
-        # Check if any rows were updated
-        return result.rowcount > 0
-
     @_workspace_transaction
     async def record_error(
         self,
@@ -2050,13 +2052,13 @@ class MonitoringService(ExecutionDetailsMixin):
         limit, offset = self.normalize_page_window(limit, offset)
         workspace_uuid = require_workspace_uuid(context)
         if bot_ids and session_ids and not pipeline_ids and not execution_statuses and mode == 'all':
-            source = self._session_message_source(workspace_uuid, bot_ids)
+            source = self._conversation_message_source(workspace_uuid, bot_ids)
             query = sqlalchemy.select(source).where(source.c.session_id.in_(session_ids))
             if start_time:
                 query = query.where(source.c.timestamp >= start_time)
             if end_time:
                 query = query.where(source.c.timestamp <= end_time)
-            return await self._session_projection_page(
+            return await self._conversation_projection_page(
                 workspace_uuid,
                 query.order_by(source.c.timestamp.desc(), source.c.id.desc()),
                 limit,
@@ -2311,7 +2313,7 @@ class MonitoringService(ExecutionDetailsMixin):
             raise ValueError('bot_id and session_id are required')
         workspace = require_workspace_uuid(context)
         execute = self.ap.persistence_mgr.execute_async
-        source = self._session_message_source(workspace, [bot_id])
+        source = self._conversation_message_source(workspace, [bot_id])
         exists = (
             await execute(
                 sqlalchemy.select(source.c.session_id)
@@ -2396,81 +2398,22 @@ class MonitoringService(ExecutionDetailsMixin):
         is_active: bool | None = None,
         limit: int = 100,
         offset: int = 0,
-        execution_statuses: list[str] | None = None,
-        mode: str = 'all',
     ) -> tuple[list[dict], int]:
-        """Get sessions with filters"""
+        """Interactive lists may only read the materialized summary contract."""
+        workspace = require_workspace_uuid(context)
         limit, offset = self.normalize_page_window(limit, offset)
-        workspace_uuid = require_workspace_uuid(context)
-        if bot_ids and not pipeline_ids and not execution_statuses and mode == 'all':
-            return await self.get_bot_conversation_sessions(
-                workspace_uuid,
-                bot_ids,
-                start_time,
-                end_time,
-                user_query,
-                is_active,
-                limit,
-                offset,
-            )
-        conditions = [persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid]
-
-        conditions.append(
-            self._monitoring_execution_condition(
-                persistence_monitoring.MonitoringSession, workspace_uuid, execution_statuses, mode
-            )
-        )
-        if bot_ids:
-            conditions.append(persistence_monitoring.MonitoringSession.bot_id.in_(bot_ids))
-        if pipeline_ids:
-            conditions.append(
-                self._monitoring_processor_condition(
-                    persistence_monitoring.MonitoringSession, workspace_uuid, pipeline_ids
-                )
-            )
-        if start_time:
-            conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time)
-        if end_time:
-            conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time)
-        if user_query and user_query.strip():
-            user_pattern = f'%{user_query.strip()}%'
-            conditions.append(
-                sqlalchemy.or_(
-                    persistence_monitoring.MonitoringSession.user_id.ilike(user_pattern),
-                    persistence_monitoring.MonitoringSession.user_name.ilike(user_pattern),
-                )
-            )
-        if is_active is not None:
-            conditions.append(persistence_monitoring.MonitoringSession.is_active == is_active)
-
-        # Get total count
-        count_query = sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringSession.session_id))
-        if conditions:
-            count_query = count_query.where(sqlalchemy.and_(*conditions))
-
-        count_result = await self.ap.persistence_mgr.execute_async(count_query)
-        total = count_result.scalar() or 0
-
-        # Get sessions
-        query = sqlalchemy.select(persistence_monitoring.MonitoringSession).order_by(
-            persistence_monitoring.MonitoringSession.last_activity.desc()
-        )
-        if conditions:
-            query = query.where(sqlalchemy.and_(*conditions))
-
-        query = query.limit(limit).offset(offset)
-
-        result = await self.ap.persistence_mgr.execute_async(query)
-        sessions_rows = result.all()
-
-        return (
-            [
-                self.ap.persistence_mgr.serialize_model(
-                    persistence_monitoring.MonitoringSession, row[0] if isinstance(row, tuple) else row
-                )
-                for row in sessions_rows
-            ],
-            total,
+        return await self._session_list.list_page(
+            workspace,
+            SessionListFilter(
+                bot_ids=tuple(bot_ids or ()),
+                pipeline_ids=tuple(pipeline_ids or ()),
+                start_time=start_time,
+                end_time=end_time,
+                user_query=user_query,
+                is_active=is_active,
+            ),
+            limit,
+            offset,
         )
 
     async def get_errors(
@@ -2728,7 +2671,7 @@ class MonitoringService(ExecutionDetailsMixin):
         return {
             'session_id': session_id,
             'found': True,
-            'session': self.ap.persistence_mgr.serialize_model(persistence_monitoring.MonitoringSession, session),
+            'session': self._serialize_session(session),
             'message_stats': {
                 'total': int(message_stats.total or 0),
                 'success': int(message_stats.success or 0),

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, *, summary_only=False):
+    def _conversation_message_source(self, workspace, bot_ids):
         """Combine delivered messages with ingress missing from the legacy ledger."""
         message = models.MonitoringMessage
         conversation = sa.func.coalesce(
@@ -148,11 +148,7 @@ class ExecutionDetailsMixin:
             'event_id': EventLog.event_id,
             'run_id': EventLog.run_id,
         }
-        columns = [
-            column
-            for column in message.__table__.columns
-            if not summary_only or column.name not in {'message_content', 'variables'}
-        ]
+        columns = list(message.__table__.columns)
         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]
@@ -168,13 +164,21 @@ class ExecutionDetailsMixin:
                     message.workspace_uuid == workspace,
                     message.bot_id == EventLog.bot_id,
                     sa.func.coalesce(message.role, 'user') != 'assistant',
-                    sa.or_(message.event_id == EventLog.event_id, message.id == EventLog.event_id),
+                    message.event_id == EventLog.event_id,
+                )
+            ),
+            ~sa.exists(
+                sa.select(message.id).where(
+                    message.workspace_uuid == workspace,
+                    message.bot_id == EventLog.bot_id,
+                    sa.func.coalesce(message.role, 'user') != 'assistant',
+                    message.id == EventLog.event_id,
                 )
             ),
         )
         return sa.union_all(stored, incoming).subquery()
 
-    async def _session_projection_page(self, workspace, statement, limit, offset):
+    async def _conversation_projection_page(self, workspace, statement, limit, offset):
         count = (
             await self._execution_query(workspace, sa.select(sa.func.count()).select_from(statement.subquery()))
         ).scalar_one()
@@ -184,79 +188,6 @@ class ExecutionDetailsMixin:
             for row in rows
         ], count
 
-    async def get_bot_conversation_sessions(
-        self, workspace, bot_ids, start_time, end_time, user_query, is_active, limit, offset
-    ):
-        """Read session summaries across processor types without mutating history."""
-        # 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(
-                *messages.c,
-                sa.func.row_number()
-                .over(
-                    partition_by=key,
-                    order_by=[
-                        sa.case((messages.c.role == 'user', 0), else_=1),
-                        messages.c.timestamp.desc(),
-                        messages.c.id.desc(),
-                    ],
-                )
-                .label('position'),
-                sa.func.min(messages.c.timestamp).over(partition_by=key).label('first_seen'),
-                sa.func.max(messages.c.timestamp).over(partition_by=key).label('last_seen'),
-                sa.func.count().over(partition_by=key).label('count'),
-            )
-            .where(messages.c.session_id != '')
-            .subquery()
-        )
-        session = models.MonitoringSession
-        fields = list(session.__table__.columns)
-        projected = {
-            'start_time': ranked.c.first_seen,
-            'last_activity': ranked.c.last_seen,
-            'message_count': ranked.c['count'],
-            'is_active': sa.literal(True),
-        }
-        derived = sa.select(
-            *[
-                (projected[field.name] if field.name in projected else ranked.c[field.name]).label(field.name)
-                for field in fields
-            ]
-        ).where(ranked.c.position == 1)
-        existing = sa.select(*fields).where(session.workspace_uuid == workspace, session.bot_id.in_(bot_ids))
-        combined = sa.union_all(existing, derived).subquery()
-        group = [combined.c.workspace_uuid, combined.c.bot_id, combined.c.session_id]
-        merged = sa.select(
-            *[
-                combined.c[field.name]
-                for field in fields
-                if field.name not in {'message_count', 'start_time', 'last_activity'}
-            ],
-            sa.func.max(combined.c.message_count).over(partition_by=group).label('message_count'),
-            sa.func.min(combined.c.start_time).over(partition_by=group).label('start_time'),
-            sa.func.max(combined.c.last_activity).over(partition_by=group).label('last_activity'),
-            sa.func.row_number().over(partition_by=group, order_by=combined.c.last_activity.desc()).label('position'),
-        ).subquery()
-        query = sa.select(*[merged.c[field.name] for field in fields]).where(merged.c.position == 1)
-        if start_time:
-            query = query.where(merged.c.last_activity >= start_time)
-        if end_time:
-            query = query.where(merged.c.last_activity <= end_time)
-        if user_query and user_query.strip():
-            pattern = f'%{user_query.strip()}%'
-            query = query.where(sa.or_(merged.c.user_id.ilike(pattern), merged.c.user_name.ilike(pattern)))
-        if is_active is not None:
-            query = query.where(merged.c.is_active == is_active)
-        return await self._session_projection_page(
-            workspace,
-            query.order_by(merged.c.last_activity.desc(), merged.c.session_id),
-            limit,
-            offset,
-        )
-
     def _execution_json_text(self, column, key):
         # SQLite JSON functions accept TEXT directly; PostgreSQL operators
         # require an explicit JSON cast for the existing text-backed journals.
@@ -276,12 +207,21 @@ class ExecutionDetailsMixin:
             )
             .values(run_id=run_id, event_id=sa.func.coalesce(models.MonitoringMessage.event_id, event_id))
         )
-        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
-        if callable(tenant_uow):
-            async with tenant_uow(workspace) as uow:
-                await uow.session.execute(statement)
-        else:
-            await self.ap.persistence_mgr.execute_async(statement)
+        async with self._summary_write_session(workspace) as session:
+            message = (
+                await session.execute(
+                    sa.select(
+                        models.MonitoringMessage.bot_id,
+                        models.MonitoringMessage.event_id,
+                        models.MonitoringMessage.role,
+                    ).where(
+                        models.MonitoringMessage.workspace_uuid == workspace, models.MonitoringMessage.id == message_id
+                    )
+                )
+            ).first()
+            await session.execute(statement)
+            if message is not None and not message.event_id and message.role != 'assistant':
+                await self._summary_writer().link_event(session, workspace, message.bot_id, message_id, event_id)
 
     async def record_ingress_event(
         self,
@@ -308,41 +248,44 @@ class ExecutionDetailsMixin:
         # of storing an unbounded adapter object in the event journal.
         content = parsed(self._sanitize_message_content(json.dumps(payload, ensure_ascii=False, default=str)), {})
         summary = event_preview(content, event_type=event_type)
-        await store.append_event(
-            event_id=event_id,
-            event_type=event_type,
-            source='platform',
-            workspace_id=workspace,
-            bot_id=bot_id,
-            actor_id=actor_id,
-            actor_name=actor_name,
-            conversation_id=conversation_id,
-            actor_type='user' if actor_id else 'system',
-            input_summary=summary,
-            input_json=content,
-            metadata={'status': status, 'bot_name': bot_name, 'platform': platform},
-        )
+        safe_routes = []
+        for route in routes or ():
+            if isinstance(route, BaseException):
+                safe_routes.append({'status': 'failed', 'reason': str(route) or type(route).__name__})
+            elif isinstance(route, dict):
+                safe_routes.append(
+                    {
+                        key: route.get(key)
+                        for key in (
+                            'status',
+                            'reason',
+                            'target_type',
+                            'target_uuid',
+                            'run_id',
+                            'failure_code',
+                        )
+                    }
+                )
+        metadata = {'status': status, 'bot_name': bot_name, 'platform': platform}
         if routes is not None:
-            safe_routes = []
-            for route in routes:
-                if isinstance(route, BaseException):
-                    safe_routes.append({'status': 'failed', 'reason': str(route) or type(route).__name__})
-                elif isinstance(route, dict):
-                    safe_routes.append(
-                        {
-                            key: route.get(key)
-                            for key in (
-                                'status',
-                                'reason',
-                                'target_type',
-                                'target_uuid',
-                                'run_id',
-                                'failure_code',
-                            )
-                        }
-                    )
-            metadata = {'status': status, 'bot_name': bot_name, 'platform': platform, 'routes': safe_routes}
-            async with AsyncSession(self.ap.persistence_mgr.get_db_engine()) as session:
+            metadata['routes'] = safe_routes
+        async with self._summary_write_session(workspace) as session:
+            await store.append_event(
+                event_id=event_id,
+                event_type=event_type,
+                source='platform',
+                workspace_id=workspace,
+                bot_id=bot_id,
+                actor_id=actor_id,
+                actor_name=actor_name,
+                conversation_id=conversation_id,
+                actor_type='user' if actor_id else 'system',
+                input_summary=summary,
+                input_json=content,
+                metadata=metadata,
+                session=session,
+            )
+            if routes is not None:
                 await session.execute(
                     sa.update(EventLog)
                     .where(
@@ -351,7 +294,31 @@ class ExecutionDetailsMixin:
                     )
                     .values(metadata_json=json.dumps(metadata, ensure_ascii=False))
                 )
-                await session.commit()
+            if event_type == 'message.received':
+                from ....persistence.session_summary_writer import ingress_summary_change
+
+                event = (
+                    (
+                        await session.execute(
+                            sa.select(
+                                EventLog.bot_id,
+                                EventLog.event_id,
+                                EventLog.conversation_id,
+                                EventLog.created_at,
+                                EventLog.event_time,
+                                EventLog.actor_id,
+                                EventLog.actor_name,
+                                EventLog.metadata_json,
+                            ).where(EventLog.workspace_id == workspace, EventLog.event_id == event_id)
+                        )
+                    )
+                    .mappings()
+                    .one()
+                )
+                await self._summary_writer().apply_once(
+                    session,
+                    ingress_summary_change(workspace, event),
+                )
 
     async def _execution_query(self, workspace, statement):
         tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)

src/langbot/pkg/api/mcp/server.py

@@ -182,7 +182,7 @@ class LangBotMCPServer:
             return _dump(ap.box_service.get_recent_errors(context))
 
         @mcp.tool(
-            description='Read bounded Workspace runtime records. LLM calls include optional details (timing in ms, complete provider usage, structured error); older records have null details. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind.'
+            description='Read bounded Workspace runtime records. LLM calls include optional details (timing in ms, complete provider usage, structured error); older records have null details. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind. Sessions read stored conversation summaries and do not support mode or execution_statuses.'
         )
         async def get_monitoring_records(
             kind: typing.Literal['messages', 'llm_calls', 'tool_calls', 'embedding_calls', 'sessions', 'errors'],
@@ -201,6 +201,7 @@ class LangBotMCPServer:
             if kind == 'embedding_calls':
                 allowed.add('knowledge_base_id')
             if kind == 'sessions':
+                allowed -= {'mode', 'execution_statuses'}
                 allowed |= {'user_query', 'is_active'}
             kwargs = dict(filters or {})
             if set(kwargs) - allowed:

src/langbot/pkg/core/app.py

@@ -359,6 +359,9 @@ class Application:
         return task
 
     async def run(self):
+        from ..persistence.session_summary_backfill import initialize_application_summaries
+
+        await initialize_application_summaries(self)
         self.event_loop_monitor.start()
         from ..utils.execution_lease import maintain_execution_leases
 

src/langbot/pkg/entity/persistence/agent_run.py

@@ -112,6 +112,7 @@ class AgentRun(Base):
     """Additional metadata JSON."""
 
     __table_args__ = (
+        sqlalchemy.Index('ix_agent_run_summary_event', 'workspace_id', 'bot_id', 'event_id', 'id'),
         sqlalchemy.Index('ix_agent_run_media_ttl', 'workspace_id', 'created_at', 'id'),
         sqlalchemy.Index(
             'ix_agent_run_scope_status', 'bot_id', 'workspace_id', 'conversation_id', 'thread_id', 'status'

src/langbot/pkg/entity/persistence/monitoring.py

@@ -39,6 +39,7 @@ class MonitoringMessage(Base):
     parent_message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
 
     __table_args__ = (
+        sqlalchemy.Index('ix_monitoring_messages_summary_event', 'workspace_uuid', 'bot_id', 'event_id', 'role'),
         sqlalchemy.Index('ix_monitoring_messages_workspace_timestamp', 'workspace_uuid', 'timestamp'),
         sqlalchemy.Index('ix_monitoring_messages_media_ttl', 'workspace_uuid', 'timestamp', 'id'),
         sqlalchemy.Index('ix_monitoring_messages_workspace_session', 'workspace_uuid', 'session_id'),
@@ -144,12 +145,70 @@ class MonitoringSession(Base):
     user_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
     user_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)  # User display name
 
+    metadata_timestamp = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+    metadata_role = sqlalchemy.Column(sqlalchemy.String(50), nullable=True)
+    metadata_source_key = sqlalchemy.Column(sqlalchemy.String(512), nullable=True)
+    base_snapshot = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    source_count = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, default=0, server_default='0')
+
     __table_args__ = (
         sqlalchemy.Index('ix_monitoring_sessions_workspace_activity', 'workspace_uuid', 'last_activity'),
         sqlalchemy.Index('ix_monitoring_sessions_workspace_active', 'workspace_uuid', 'is_active'),
+        sqlalchemy.Index('ix_monitoring_sessions_workspace_bot', 'workspace_uuid', 'bot_id', 'last_activity'),
     )
 
 
+class MonitoringSessionReceipt(Base):
+    """Retained message identities and replaceable, optionally detached event credits."""
+
+    __tablename__ = 'monitoring_session_receipts'
+
+    workspace_uuid = sqlalchemy.Column(
+        sqlalchemy.String(36), sqlalchemy.ForeignKey('workspaces.uuid', ondelete='CASCADE'), primary_key=True
+    )
+    bot_id = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
+    source_key = sqlalchemy.Column(sqlalchemy.String(512), primary_key=True)
+    session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
+    event_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
+    role = sqlalchemy.Column(sqlalchemy.String(50), nullable=False)
+    covered = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, server_default='false')
+    timestamp = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    candidate_snapshot = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    source_timestamp = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+
+    __table_args__ = (
+        sqlalchemy.ForeignKeyConstraint(
+            ['workspace_uuid', 'bot_id', 'session_id'],
+            ['monitoring_sessions.workspace_uuid', 'monitoring_sessions.bot_id', 'monitoring_sessions.session_id'],
+            ondelete='CASCADE',
+        ),
+        sqlalchemy.Index('ix_session_receipts_session', 'workspace_uuid', 'bot_id', 'session_id'),
+        sqlalchemy.Index('ix_session_receipts_expiry', 'workspace_uuid', 'timestamp', 'bot_id', 'source_key'),
+    )
+
+
+# Only uncovered ingress needs a replaceable metadata candidate. C/BINARY order
+# agrees with UTF-8 source-key ordering in the projection on both databases.
+sqlalchemy.Index(
+    'ix_session_receipts_candidate',
+    MonitoringSessionReceipt.workspace_uuid,
+    MonitoringSessionReceipt.bot_id,
+    MonitoringSessionReceipt.session_id,
+    MonitoringSessionReceipt.source_timestamp.desc(),
+    MonitoringSessionReceipt.source_key.collate('C').desc(),
+    postgresql_where=MonitoringSessionReceipt.candidate_snapshot.is_not(None),
+).ddl_if(dialect='postgresql')
+sqlalchemy.Index(
+    'ix_session_receipts_candidate',
+    MonitoringSessionReceipt.workspace_uuid,
+    MonitoringSessionReceipt.bot_id,
+    MonitoringSessionReceipt.session_id,
+    MonitoringSessionReceipt.source_timestamp.desc(),
+    MonitoringSessionReceipt.source_key.collate('BINARY').desc(),
+    sqlite_where=MonitoringSessionReceipt.candidate_snapshot.is_not(None),
+).ddl_if(dialect='sqlite')
+
+
 class MonitoringError(Base):
     """Error log records"""
 

src/langbot/pkg/persistence/alembic/versions/0040_session_summaries.py

@@ -0,0 +1,121 @@
+"""Persist idempotent conversation summaries independently of message history."""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = '0040_session_summaries'
+down_revision = '0039_execution_lease_indexes'
+branch_labels = None
+depends_on = None
+
+TABLE = 'monitoring_session_receipts'
+INDEXES = (
+    ('monitoring_messages', 'ix_monitoring_messages_summary_event', ['workspace_uuid', 'bot_id', 'event_id', 'role']),
+    ('agent_run', 'ix_agent_run_summary_event', ['workspace_id', 'bot_id', 'event_id', 'id']),
+)
+
+
+def upgrade():
+    connection = op.get_bind()
+    for table, index, fields in INDEXES:
+        if not sa.inspect(connection).has_table(table):
+            continue
+        if index not in {item['name'] for item in sa.inspect(connection).get_indexes(table)}:
+            op.create_index(index, table, fields)
+    # Bare Alembic and partial-schema upgrades rely on ORM bootstrap for core tables.
+    if not sa.inspect(connection).has_table('monitoring_sessions'):
+        return
+    columns = {column['name'] for column in sa.inspect(connection).get_columns('monitoring_sessions')}
+    for name, type_ in [
+        ('metadata_timestamp', sa.DateTime()),
+        ('metadata_role', sa.String(50)),
+        ('metadata_source_key', sa.String(512)),
+        ('base_snapshot', sa.Text()),
+    ]:
+        if name not in columns:
+            op.add_column('monitoring_sessions', sa.Column(name, type_, nullable=True))
+    if 'source_count' not in columns:
+        op.add_column(
+            'monitoring_sessions', sa.Column('source_count', sa.Integer(), nullable=False, server_default='0')
+        )
+    # Legacy metadata provenance is unknown. Keep existing display values, but
+    # leave provenance and the base snapshot NULL until bounded projection.
+    if not sa.inspect(connection).has_table(TABLE):
+        op.create_table(
+            TABLE,
+            sa.Column(
+                'workspace_uuid', sa.String(36), sa.ForeignKey('workspaces.uuid', ondelete='CASCADE'), primary_key=True
+            ),
+            sa.Column('bot_id', sa.String(255), primary_key=True),
+            sa.Column('source_key', sa.String(512), primary_key=True),
+            sa.Column('session_id', sa.String(255), nullable=True),
+            sa.Column('event_id', sa.String(255), nullable=True),
+            sa.Column('role', sa.String(50), nullable=False),
+            sa.Column('covered', sa.Boolean(), nullable=False, server_default='false'),
+            sa.Column('timestamp', sa.DateTime(), nullable=False),
+            sa.Column('candidate_snapshot', sa.Text(), nullable=True),
+            sa.Column('source_timestamp', sa.DateTime(), nullable=True),
+            sa.ForeignKeyConstraint(
+                ['workspace_uuid', 'bot_id', 'session_id'],
+                ['monitoring_sessions.workspace_uuid', 'monitoring_sessions.bot_id', 'monitoring_sessions.session_id'],
+                ondelete='CASCADE',
+            ),
+        )
+        op.create_index('ix_session_receipts_session', TABLE, ['workspace_uuid', 'bot_id', 'session_id'])
+        op.create_index('ix_session_receipts_expiry', TABLE, ['workspace_uuid', 'timestamp', 'bot_id', 'source_key'])
+        # Historical receipts must be initialized offline before live projection.
+        if sa.inspect(connection).has_table('workspace_metadata'):
+            sources = ['EXISTS (SELECT 1 FROM monitoring_sessions s WHERE s.workspace_uuid=w.uuid)']
+            if sa.inspect(connection).has_table('monitoring_messages'):
+                sources.append('EXISTS (SELECT 1 FROM monitoring_messages m WHERE m.workspace_uuid=w.uuid)')
+            if sa.inspect(connection).has_table('event_log'):
+                sources.append('EXISTS (SELECT 1 FROM event_log e WHERE e.workspace_id=w.uuid)')
+            connection.execute(
+                sa.text(
+                    'INSERT INTO workspace_metadata (workspace_uuid, key, value) '
+                    "SELECT w.uuid, 'session_summary_backfill_v1', :cursor FROM workspaces w WHERE ("
+                    + ' OR '.join(sources)
+                    + ') ON CONFLICT (workspace_uuid, key) DO UPDATE SET value=excluded.value'
+                ),
+                {'cursor': '{"source": "messages"}'},
+            )
+    collation = '"C"' if connection.dialect.name == 'postgresql' else 'BINARY'
+    op.execute(
+        sa.text(
+            f'CREATE INDEX IF NOT EXISTS ix_session_receipts_candidate ON {TABLE} '
+            '(workspace_uuid, bot_id, session_id, source_timestamp DESC, '
+            f'source_key COLLATE {collation} DESC) WHERE candidate_snapshot IS NOT NULL'
+        )
+    )
+    if connection.dialect.name == 'postgresql':
+        predicate = "workspace_uuid::text = NULLIF(current_setting('langbot.workspace_uuid', true), '')"
+        op.execute(sa.text(f'ALTER TABLE {TABLE} ENABLE ROW LEVEL SECURITY'))
+        op.execute(sa.text(f'ALTER TABLE {TABLE} FORCE ROW LEVEL SECURITY'))
+        op.execute(sa.text(f'DROP POLICY IF EXISTS langbot_workspace_isolation ON {TABLE}'))
+        op.execute(
+            sa.text(
+                f'CREATE POLICY langbot_workspace_isolation ON {TABLE} AS PERMISSIVE FOR ALL TO PUBLIC '
+                f'USING ({predicate}) WITH CHECK ({predicate})'
+            )
+        )
+
+
+def downgrade():
+    connection = op.get_bind()
+    for table, index, _ in reversed(INDEXES):
+        if sa.inspect(connection).has_table(table):
+            if index in {item['name'] for item in sa.inspect(connection).get_indexes(table)}:
+                op.drop_index(index, table_name=table)
+    if sa.inspect(connection).has_table(TABLE):
+        op.drop_table(TABLE)
+    if sa.inspect(connection).has_table('monitoring_sessions'):
+        columns = {item['name'] for item in sa.inspect(connection).get_columns('monitoring_sessions')}
+        for column in (
+            'base_snapshot',
+            'metadata_source_key',
+            'metadata_role',
+            'metadata_timestamp',
+            'source_count',
+        ):
+            if column in columns:
+                op.drop_column('monitoring_sessions', column)

src/langbot/pkg/persistence/mgr.py

@@ -82,6 +82,7 @@ _ALEMBIC_TENANT_TABLES = {
     'monitoring_llm_calls',
     'monitoring_tool_calls',
     'monitoring_sessions',
+    'monitoring_session_receipts',
     'monitoring_errors',
     'monitoring_embedding_calls',
     'monitoring_feedback',
@@ -273,6 +274,16 @@ class PersistenceManager:
             if self.mode == PersistenceMode.OSS_COMPAT:
                 await self._install_oss_postgres_tenant_scope()
         else:
+            current_revision = await alembic_runner.get_alembic_current(engine)
+            if (
+                current_revision is None
+                or current_revision in _PRE_WORKSPACE_ALEMBIC_REVISIONS
+                or current_revision == _WORKSPACE_ALEMBIC_REVISION
+            ):
+                await self._run_alembic_migrations(_RESOURCE_SCOPE_ALEMBIC_REVISION)
+                # Deferred parent tables must exist before later migrations
+                # create dependent tables such as session summary receipts.
+                await self.create_tables()
             await self._run_alembic_migrations()
 
             # SQLite keeps the historical post-migration create_all pass. New
@@ -294,7 +305,10 @@ class PersistenceManager:
                 # without workspace_uuid and Workspace itself references the
                 # account UUID introduced by 0009.  Alembic must expand those
                 # tables before SQLAlchemy may create any new tenant table.
-                excluded_tables = _ALEMBIC_TENANT_TABLES if legacy_users else set()
+                excluded_tables = set(_ALEMBIC_TENANT_TABLES) if legacy_users else set()
+                # 0010 and 0023 rewrite the session primary key on an OSS bootstrap.
+                # Only 0040 may create its dependent composite receipt foreign key.
+                excluded_tables.add('monitoring_session_receipts')
                 tables_to_create = [table for table in self.meta.sorted_tables if table.name not in excluded_tables]
                 self.meta.create_all(sync_conn, tables=tables_to_create)
 

src/langbot/pkg/persistence/session_summary_backfill.py

@@ -0,0 +1,284 @@
+"""Resumable, bounded historical repair, never called by interactive reads."""
+
+import asyncio
+import argparse
+import datetime
+import json
+import os
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql, sqlite
+
+from ..entity.persistence.agent_run import AgentRun
+from ..entity.persistence.event_log import EventLog
+from ..entity.persistence.metadata import WorkspaceMetadata
+from ..entity.persistence.monitoring import MonitoringMessage
+from .session_summary_reader import SessionScope
+from .session_summary_writer import SessionSummaryWriter, ingress_summary_change, message_summary_change
+
+
+CURSOR_KEY = 'session_summary_backfill_v1'
+BATCH_SIZE = 100
+JSON_LIMIT = 65536
+
+
+async def _initialization_pending(session, workspace: str) -> bool:
+    value = (
+        await session.execute(
+            sa.select(WorkspaceMetadata.value).where(
+                WorkspaceMetadata.workspace_uuid == workspace,
+                WorkspaceMetadata.key == CURSOR_KEY,
+            )
+        )
+    ).scalar_one_or_none()
+    return value is not None and json.loads(value)['source'] != 'done'
+
+
+async def require_initialized(session, workspace: str):
+    if await _initialization_pending(session, workspace):
+        raise RuntimeError(
+            f'Session summaries for Workspace {workspace} require offline initialization. '
+            'Stop writers and run: uv run python -m langbot.pkg.persistence.session_summary_backfill '
+            f'--workspace {workspace} --writers-stopped --database-url-env LANGBOT_SUMMARY_DATABASE_URL'
+        )
+
+
+async def require_application_initialized(ap):
+    """Reject startup before producers can replay uninitialized history."""
+    for binding in await ap.workspace_service.list_active_execution_bindings():
+        async with ap.monitoring_service._summary_session(binding.workspace_uuid) as session:
+            await require_initialized(session, binding.workspace_uuid)
+
+
+async def initialize_application_summaries(ap):
+    """Complete OSS initialization before starting HTTP or message producers."""
+    from .mgr import PersistenceMode
+
+    if ap.persistence_mgr.mode != PersistenceMode.OSS_COMPAT:
+        await require_application_initialized(ap)
+        return
+
+    backfill = SessionSummaryBackfill(
+        ap.monitoring_service._summary_session, ap.persistence_mgr.get_db_engine().dialect.name
+    )
+    for binding in await ap.workspace_service.list_active_execution_bindings():
+        workspace = binding.workspace_uuid
+        async with ap.monitoring_service._summary_session(workspace) as session:
+            pending = await _initialization_pending(session, workspace)
+        if not pending:
+            continue
+        ap.logger.info(f'Initializing session summaries for Workspace {workspace} before starting services.')
+        while not await backfill.run_batch(workspace):
+            await asyncio.sleep(0)
+        ap.logger.info(f'Session summaries for Workspace {workspace} initialized; continuing startup.')
+
+
+class SessionSummaryBackfill:
+    def __init__(self, session_scope: SessionScope, dialect: str):
+        self._session_scope = session_scope
+        self._dialect = dialect
+        self._insert = {'postgresql': postgresql.insert, 'sqlite': sqlite.insert}[dialect]
+        self._writer = SessionSummaryWriter(dialect)
+
+    def _json_text(self, column, key):
+        expression = sa.type_coerce(column, sa.JSON) if self._dialect == 'sqlite' else sa.cast(column, sa.JSON)
+        relevant = sa.and_(
+            EventLog.source == 'platform',
+            EventLog.event_type == 'message.received',
+            sa.func.length(column) <= JSON_LIMIT,
+        )
+        return sa.case((relevant, expression[key].as_string()), else_=None)
+
+    async def run_batch(self, workspace: str, batch_size: int = BATCH_SIZE) -> bool:
+        """Commit at most one page and its cursor atomically; return completion."""
+        if not 1 <= batch_size <= BATCH_SIZE:
+            raise ValueError(f'batch_size must be between 1 and {BATCH_SIZE}')
+        async with self._session_scope(workspace) as session:
+            await session.execute(
+                self._insert(WorkspaceMetadata)
+                .values(
+                    workspace_uuid=workspace,
+                    key=CURSOR_KEY,
+                    value=json.dumps({'source': 'messages'}),
+                )
+                .on_conflict_do_nothing(index_elements=['workspace_uuid', 'key'])
+            )
+            cursor = json.loads(
+                (
+                    await session.execute(
+                        sa.select(WorkspaceMetadata.value)
+                        .where(
+                            WorkspaceMetadata.workspace_uuid == workspace,
+                            WorkspaceMetadata.key == CURSOR_KEY,
+                        )
+                        .with_for_update()
+                    )
+                ).scalar_one()
+            )
+            source = cursor['source']
+            if source == 'done':
+                return True
+            model = MonitoringMessage if source == 'messages' else EventLog
+            timestamp = model.timestamp if source == 'messages' else model.created_at
+            tenant = model.workspace_uuid if source == 'messages' else model.workspace_id
+            # The same (Workspace, timestamp, id) indexes also drive media expiry.
+            page = sa.select(model.id.label('id'), timestamp.label('cursor_time')).where(tenant == workspace)
+            if cursor.get('after'):
+                date, identity = cursor['after']
+                page = page.where(
+                    sa.tuple_(timestamp, model.id) > sa.tuple_(datetime.datetime.fromisoformat(date), identity)
+                )
+            page = page.order_by(timestamp, model.id).limit(batch_size).subquery()
+            if source == 'messages':
+                fields = (
+                    'bot_id',
+                    'session_id',
+                    'timestamp',
+                    'bot_name',
+                    'pipeline_id',
+                    'pipeline_name',
+                    'platform',
+                    'user_id',
+                    'user_name',
+                    'role',
+                    'event_id',
+                )
+                statement = sa.select(page.c.cursor_time, model.id, *(getattr(model, field) for field in fields))
+            else:
+                known_conversation = sa.func.coalesce(
+                    model.conversation_id,
+                    sa.select(AgentRun.conversation_id)
+                    .where(
+                        AgentRun.workspace_id == workspace,
+                        AgentRun.bot_id == model.bot_id,
+                        AgentRun.event_id == model.event_id,
+                    )
+                    .order_by(AgentRun.id)
+                    .limit(1)
+                    .scalar_subquery(),
+                )
+                conversation = sa.func.coalesce(
+                    known_conversation,
+                    sa.case(
+                        (
+                            self._json_text(model.input_json, 'chat_type') == 'group',
+                            sa.literal('group_') + self._json_text(model.input_json, 'chat_id'),
+                        ),
+                        (
+                            self._json_text(model.input_json, 'chat_type') == 'private',
+                            sa.literal('person_') + self._json_text(model.input_json, ('sender', 'id')),
+                        ),
+                    ),
+                )
+                statement = sa.select(
+                    page.c.cursor_time,
+                    model.id,
+                    model.bot_id,
+                    model.event_id,
+                    model.event_time,
+                    model.created_at,
+                    conversation.label('conversation_id'),
+                    sa.case(
+                        (
+                            sa.and_(
+                                model.source == 'platform',
+                                model.event_type == 'message.received',
+                                sa.func.length(model.metadata_json) <= JSON_LIMIT,
+                            ),
+                            model.metadata_json,
+                        ),
+                        else_=None,
+                    ).label('metadata_json'),
+                    model.actor_id,
+                    model.actor_name,
+                    model.source,
+                    model.event_type,
+                    sa.or_(
+                        sa.func.length(model.metadata_json) > JSON_LIMIT,
+                        sa.and_(known_conversation.is_(None), sa.func.length(model.input_json) > JSON_LIMIT),
+                    ).label('oversized'),
+                )
+            rows = (
+                (
+                    await session.execute(
+                        statement.join(model, model.id == page.c.id).order_by(page.c.cursor_time, model.id)
+                    )
+                )
+                .mappings()
+                .all()
+            )
+            for row in rows:
+                if source == 'events' and (row['source'] != 'platform' or row['event_type'] != 'message.received'):
+                    continue
+                if source == 'events' and row['oversized']:
+                    raise ValueError(f'EventLog {row["id"]} exceeds the {JSON_LIMIT}-character legacy metadata limit')
+                if not row['bot_id']:
+                    continue
+                await self._writer.apply_once(
+                    session,
+                    (message_summary_change if source == 'messages' else ingress_summary_change)(workspace, row),
+                    backfill=True,
+                )
+            if len(rows) < batch_size:
+                cursor = {'source': 'events' if source == 'messages' else 'done'}
+            else:
+                cursor['after'] = [rows[-1]['cursor_time'].isoformat(), rows[-1]['id']]
+            await session.execute(
+                sa.update(WorkspaceMetadata)
+                .where(
+                    WorkspaceMetadata.workspace_uuid == workspace,
+                    WorkspaceMetadata.key == CURSOR_KEY,
+                )
+                .values(value=json.dumps(cursor))
+            )
+            return cursor['source'] == 'done'
+
+
+async def _main(args):
+    from sqlalchemy.engine import make_url
+    from sqlalchemy.ext.asyncio import create_async_engine
+    from .tenant_uow import TenantUnitOfWork
+
+    url = make_url(os.environ[args.database_url_env])
+    if url.drivername not in {'sqlite+aiosqlite', 'postgresql+asyncpg'}:
+        raise ValueError('Use an explicit sqlite+aiosqlite or postgresql+asyncpg database URL')
+    if url.drivername == 'sqlite+aiosqlite' and (not url.database or not Path(url.database).is_file()):
+        raise ValueError('The SQLite database must already exist and be migrated')
+    engine = create_async_engine(url)
+    if engine.dialect.name == 'sqlite':
+
+        @sa.event.listens_for(engine.sync_engine, 'connect')
+        def enable_foreign_keys(connection, _record):
+            cursor = connection.cursor()
+            cursor.execute('PRAGMA foreign_keys=ON')
+            cursor.close()
+
+    @asynccontextmanager
+    async def scope(workspace):
+        async with TenantUnitOfWork(engine, workspace) as uow:
+            yield uow.session
+
+    try:
+        async with engine.connect() as connection:
+            columns = await connection.run_sync(
+                lambda conn: {item['name'] for item in sa.inspect(conn).get_columns('monitoring_session_receipts')}
+            )
+            if not {'session_id', 'covered', 'timestamp'} <= columns:
+                raise ValueError('Apply migration 0040_session_summaries before initializing summaries')
+        backfill = SessionSummaryBackfill(scope, engine.dialect.name)
+        for workspace in args.workspace:
+            while not await backfill.run_batch(workspace):
+                await asyncio.sleep(0)
+            print(f'Workspace {workspace}: session summaries initialized')
+    finally:
+        await engine.dispose()
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='Initialize session summaries offline; stop all writers first')
+    parser.add_argument('--workspace', action='append', required=True)
+    parser.add_argument('--database-url-env', default='LANGBOT_SUMMARY_DATABASE_URL')
+    parser.add_argument('--writers-stopped', action='store_true', required=True)
+    asyncio.run(_main(parser.parse_args()))

src/langbot/pkg/persistence/session_summary_contract.py

@@ -0,0 +1,45 @@
+"""Payload-free contracts for interactive conversation lists."""
+
+from dataclasses import dataclass
+import datetime
+from typing import Protocol
+
+
+@dataclass(frozen=True, slots=True)
+class SessionListFilter:
+    bot_ids: tuple[str, ...] = ()
+    pipeline_ids: tuple[str, ...] = ()
+    start_time: datetime.datetime | None = None
+    end_time: datetime.datetime | None = None
+    user_query: str | None = None
+    is_active: bool | None = None
+
+
+class SessionSummaryReader(Protocol):
+    """Read stored summaries without consulting or repairing historical ledgers."""
+
+    async def list_page(
+        self, workspace: str, filters: SessionListFilter, limit: int, offset: int
+    ) -> tuple[list[dict], int]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class SessionSummaryChange:
+    workspace: str
+    bot_id: str
+    session_id: str
+    source_key: str
+    timestamp: datetime.datetime
+    bot_name: str
+    pipeline_id: str = ''
+    pipeline_name: str = ''
+    platform: str | None = None
+    user_id: str | None = None
+    user_name: str | None = None
+    role: str = 'user'
+    event_id: str | None = None
+
+
+def message_source_key(message_id: str) -> str:
+    """Distinct message records remain distinct even when they share an event."""
+    return f'message:{message_id}'

src/langbot/pkg/persistence/session_summary_reader.py

@@ -0,0 +1,74 @@
+"""Only the summary table is available to the interactive list implementation."""
+
+from collections.abc import Callable
+from contextlib import AbstractAsyncContextManager
+import datetime
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..entity.persistence.monitoring import MonitoringSession
+from .session_summary_contract import SessionListFilter
+
+
+SessionScope = Callable[[str], AbstractAsyncContextManager[AsyncSession]]
+SUMMARY_FIELDS = (
+    'workspace_uuid',
+    'bot_id',
+    'session_id',
+    'bot_name',
+    'pipeline_id',
+    'pipeline_name',
+    'message_count',
+    'start_time',
+    'last_activity',
+    'is_active',
+    'platform',
+    'user_id',
+    'user_name',
+)
+
+
+class SqlSessionSummaryReader:
+    def __init__(self, session_scope: SessionScope):
+        self._session_scope = session_scope
+
+    async def list_page(
+        self, workspace: str, filters: SessionListFilter, limit: int, offset: int
+    ) -> tuple[list[dict], int]:
+        model = MonitoringSession
+        conditions = [model.workspace_uuid == workspace]
+        if filters.bot_ids:
+            conditions.append(model.bot_id.in_(filters.bot_ids))
+        if filters.pipeline_ids:
+            conditions.append(model.pipeline_id.in_(filters.pipeline_ids))
+        if filters.start_time is not None:
+            conditions.append(model.last_activity >= filters.start_time)
+        if filters.end_time is not None:
+            conditions.append(model.last_activity <= filters.end_time)
+        if filters.user_query and filters.user_query.strip():
+            pattern = f'%{filters.user_query.strip()}%'
+            conditions.append(sa.or_(model.user_id.ilike(pattern), model.user_name.ilike(pattern)))
+        if filters.is_active is not None:
+            conditions.append(model.is_active == filters.is_active)
+        async with self._session_scope(workspace) as session:
+            total = (
+                await session.execute(sa.select(sa.func.count()).select_from(model).where(*conditions))
+            ).scalar_one()
+            rows = (
+                (
+                    await session.execute(
+                        sa.select(*(getattr(model, field) for field in SUMMARY_FIELDS))
+                        .where(*conditions)
+                        .order_by(model.last_activity.desc(), model.bot_id, model.session_id)
+                        .limit(limit)
+                        .offset(offset)
+                    )
+                )
+                .mappings()
+                .all()
+            )
+        return [
+            {key: value.isoformat() if isinstance(value, datetime.datetime) else value for key, value in row.items()}
+            for row in rows
+        ], total

src/langbot/pkg/persistence/session_summary_writer.py

@@ -0,0 +1,363 @@
+"""Transaction-bound message counts and replaceable ingress credits."""
+
+import datetime
+import json
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql, sqlite
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..entity.persistence.monitoring import MonitoringSession, MonitoringSessionReceipt
+from .datetime_utils import as_naive_utc
+from .session_summary_contract import SessionSummaryChange, message_source_key
+
+
+SESSION_KEY = ['workspace_uuid', 'bot_id', 'session_id']
+RECEIPT_KEY = ['workspace_uuid', 'bot_id', 'source_key']
+
+
+def message_summary_change(workspace: str, row) -> SessionSummaryChange:
+    """Use the same stored message facts for live writes and historical repair."""
+    return SessionSummaryChange(
+        workspace=workspace,
+        bot_id=row['bot_id'],
+        session_id=row['session_id'] or '',
+        source_key=message_source_key(row['id']),
+        event_id=row['event_id'],
+        timestamp=row['timestamp'],
+        bot_name=row['bot_name'] or '',
+        pipeline_id=row['pipeline_id'] or '',
+        pipeline_name=row['pipeline_name'] or '',
+        platform=row['platform'],
+        user_id=row['user_id'],
+        user_name=row['user_name'],
+        role=row['role'] or 'user',
+    )
+
+
+def ingress_summary_change(workspace: str, row) -> SessionSummaryChange:
+    """Project persisted ingress identities and the first route with a target."""
+    metadata = json.loads(row['metadata_json'] or '{}')
+    if not isinstance(metadata, dict):
+        raise ValueError('Ingress metadata must be a JSON object')
+    route = next(
+        (route for route in metadata.get('routes') or () if isinstance(route, dict) and route.get('target_uuid')),
+        {},
+    )
+    return SessionSummaryChange(
+        workspace=workspace,
+        bot_id=row['bot_id'],
+        session_id=row['conversation_id'] or '',
+        source_key=f'event:{row["event_id"]}',
+        timestamp=row['event_time'] or row['created_at'],
+        bot_name=metadata.get('bot_name') or '',
+        pipeline_id=route.get('target_uuid') or '',
+        pipeline_name=route.get('target_type') or '',
+        platform=metadata.get('platform'),
+        user_id=row['actor_id'],
+        user_name=row['actor_name'],
+    )
+
+
+SNAPSHOT_FIELDS = ('bot_name', 'platform', 'user_id', 'user_name', 'pipeline_id', 'pipeline_name')
+
+
+def _snapshot(change):
+    return {
+        'source_key': change.source_key,
+        'timestamp': as_naive_utc(change.timestamp).isoformat(),
+        'role': change.role,
+        **{field: getattr(change, field) for field in SNAPSHOT_FIELDS},
+    }
+
+
+def _rank(snapshot):
+    if snapshot['source_key'] is None:
+        return (-1, '', b'')
+    return (int(snapshot['role'] != 'assistant'), snapshot['timestamp'], snapshot['source_key'].encode('utf-8'))
+
+
+class SessionSummaryWriter:
+    def __init__(self, dialect: str):
+        self._insert = {'postgresql': postgresql.insert, 'sqlite': sqlite.insert}[dialect]
+        self._collation = 'C' if dialect == 'postgresql' else 'BINARY'
+
+    @staticmethod
+    def _scope(model, workspace, bot_id=None, identity=None):
+        conditions = [model.workspace_uuid == workspace]
+        if bot_id is not None:
+            conditions.append(model.bot_id == bot_id)
+        if identity is not None:
+            conditions.append(model.session_id == identity)
+        return conditions
+
+    async def _lock_event(self, session, workspace, bot_id, event_id, timestamp):
+        """Serialize an event even when it has not arrived yet."""
+        receipt = MonitoringSessionReceipt
+        return (
+            (
+                await session.execute(
+                    self._insert(receipt)
+                    .values(
+                        workspace_uuid=workspace,
+                        bot_id=bot_id,
+                        source_key=f'event:{event_id}',
+                        event_id=event_id,
+                        role='event',
+                        covered=False,
+                        timestamp=timestamp,
+                    )
+                    .on_conflict_do_update(
+                        index_elements=RECEIPT_KEY,
+                        set_={
+                            'timestamp': sa.case((receipt.timestamp < timestamp, timestamp), else_=receipt.timestamp)
+                        },
+                    )
+                    .returning(receipt.session_id, receipt.covered)
+                )
+            )
+            .mappings()
+            .one()
+        )
+
+    async def _cover_event(self, session, workspace, bot_id, event_id, timestamp):
+        previous = await self._lock_event(session, workspace, bot_id, event_id, timestamp)
+        receipt = MonitoringSessionReceipt
+        await session.execute(
+            sa.update(receipt)
+            .where(*self._scope(receipt, workspace, bot_id), receipt.source_key == f'event:{event_id}')
+            .values(covered=True, session_id=None, candidate_snapshot=None, source_timestamp=None)
+        )
+        return previous['session_id']
+
+    async def _lock_summaries(self, session, workspace, identities, change=None):
+        """Lock sorted (bot, session) keys before message receipt FK checks."""
+        model = MonitoringSession
+        for bot_id, identity in sorted(identities):
+            key = self._scope(model, workspace, bot_id, identity)
+            if change is not None and (bot_id, identity) == (change.bot_id, change.session_id):
+                timestamp = as_naive_utc(change.timestamp)
+                # Updating the key takes FOR UPDATE, and recreates a deleted row.
+                await session.execute(
+                    self._insert(model)
+                    .values(
+                        workspace_uuid=workspace,
+                        bot_id=bot_id,
+                        session_id=identity,
+                        bot_name='',
+                        pipeline_id='',
+                        pipeline_name='',
+                        message_count=0,
+                        source_count=0,
+                        start_time=timestamp,
+                        last_activity=timestamp,
+                        is_active=True,
+                    )
+                    .on_conflict_do_update(index_elements=SESSION_KEY, set_={'session_id': identity})
+                )
+            else:
+                await session.execute(sa.select(model.session_id).where(*key).with_for_update())
+            row = (await session.execute(sa.select(model.__table__).where(*key))).mappings().one_or_none()
+            if row is not None:
+                # Legacy values are an unranked fallback, seeded one row at a time.
+                # Never promote a replaceable candidate into the immutable base.
+                if row['base_snapshot'] is None and row['metadata_source_key'] is None and row['message_count'] > 0:
+                    fallback = {
+                        'source_key': None,
+                        'timestamp': None,
+                        'role': None,
+                        **{field: row[field] for field in SNAPSHOT_FIELDS},
+                    }
+                    await session.execute(sa.update(model).where(*key).values(base_snapshot=json.dumps(fallback)))
+
+    async def _merge_base(self, session, workspace, bot_id, identity, snapshot):
+        """Fold immutable messages and expired ingress under the summary lock."""
+        model = MonitoringSession
+        key = self._scope(model, workspace, bot_id, identity)
+        encoded = (await session.execute(sa.select(model.base_snapshot).where(*key))).scalar_one()
+        previous = json.loads(encoded) if encoded else None
+        if previous is None or _rank(snapshot) >= _rank(previous):
+            await session.execute(sa.update(model).where(*key).values(base_snapshot=json.dumps(snapshot)))
+
+    async def _refresh(self, session, workspace, bot_id, identity, delta=0, *, backfill=False, observed=None):
+        model, receipt = MonitoringSession, MonitoringSessionReceipt
+        key = self._scope(model, workspace, bot_id, identity)
+        row = (await session.execute(sa.select(model.__table__).where(*key))).mappings().one()
+        candidate = (
+            await session.execute(
+                sa.select(receipt.candidate_snapshot)
+                .where(
+                    *self._scope(receipt, workspace, bot_id, identity),
+                    receipt.candidate_snapshot.is_not(None),
+                )
+                .order_by(receipt.source_timestamp.desc(), receipt.source_key.collate(self._collation).desc())
+                .limit(1)
+            )
+        ).scalar_one_or_none()
+        snapshots = [json.loads(value) for value in (row['base_snapshot'], candidate) if value is not None]
+        count = row['source_count'] + delta
+        displayed = max(row['message_count'], count) if backfill else row['message_count'] + delta
+        if count < 0 or displayed < 0:
+            raise ValueError('Session summary counts cannot be negative')
+        values = dict(source_count=count, message_count=displayed)
+        if snapshots:
+            selected = max(snapshots, key=_rank)
+            values.update({field: selected[field] for field in SNAPSHOT_FIELDS})
+            values.update(
+                metadata_source_key=selected['source_key'],
+                metadata_role=selected['role'],
+                metadata_timestamp=(
+                    datetime.datetime.fromisoformat(selected['timestamp']) if selected['timestamp'] else None
+                ),
+            )
+        if observed is not None:
+            values.update(
+                start_time=min(row['start_time'], observed), last_activity=max(row['last_activity'], observed)
+            )
+        await session.execute(sa.update(model).where(*key).values(**values))
+        if count == 0 and displayed == 0:
+            await session.execute(sa.delete(model).where(*key))
+
+    async def link_event(self, session: AsyncSession, workspace: str, bot_id: str, message_id: str, event_id: str):
+        """Keep the message credit and retire an uncovered ingress credit once."""
+        timestamp = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+        state = await self._lock_event(session, workspace, bot_id, event_id, timestamp)
+        receipt = MonitoringSessionReceipt
+        predicates = [*self._scope(receipt, workspace, bot_id), receipt.source_key == f'message:{message_id}']
+        statement = sa.select(receipt.session_id, receipt.event_id, receipt.role).where(*predicates)
+        message = (await session.execute(statement)).mappings().one_or_none()
+        if message is None or message['role'] == 'assistant':
+            return
+        identities = {(bot_id, message['session_id'])}
+        owner = state['session_id']
+        if owner is not None:
+            identities.add((bot_id, owner))
+        # The same lock order as apply_once also serializes a message replay.
+        await self._lock_summaries(session, workspace, identities)
+        message = (await session.execute(statement.with_for_update())).mappings().one_or_none()
+        if message is None:
+            return
+        if message['event_id'] is not None:
+            if message['event_id'] != event_id:
+                raise ValueError('A message cannot be linked to two events')
+            return
+        await session.execute(
+            sa.update(receipt)
+            .where(*self._scope(receipt, workspace, bot_id), receipt.source_key == f'event:{event_id}')
+            .values(covered=True, session_id=None, candidate_snapshot=None, source_timestamp=None)
+        )
+        await session.execute(sa.update(receipt).where(*predicates).values(event_id=event_id, timestamp=timestamp))
+        if owner is not None:
+            await self._refresh(session, workspace, bot_id, owner, -1)
+
+    async def apply_once(self, session: AsyncSession, change: SessionSummaryChange, *, backfill: bool = False) -> bool:
+        """Project a source in its transaction; distinct message rows always count."""
+        if not change.session_id:
+            return False
+        receipt = MonitoringSessionReceipt
+        workspace, bot_id, identity = change.workspace, change.bot_id, change.session_id
+        timestamp = as_naive_utc(change.timestamp)
+        deltas = {}
+        ingress = change.source_key.startswith('event:')
+        if ingress:
+            state = await self._lock_event(
+                session, workspace, bot_id, change.source_key.removeprefix('event:'), timestamp
+            )
+            if state['covered']:
+                return False
+            if state['session_id'] is not None and state['session_id'] != identity:
+                raise ValueError('An ingress event cannot belong to two conversations')
+            claimed = state['session_id'] is None
+        elif change.role != 'assistant':
+            # Legacy message.id matching is part of the historical detail contract.
+            for event_id in sorted({change.source_key.removeprefix('message:'), change.event_id} - {None}):
+                owner = await self._cover_event(session, workspace, bot_id, event_id, timestamp)
+                if owner is not None:
+                    deltas[owner] = deltas.get(owner, 0) - 1
+        await self._lock_summaries(session, workspace, {(bot_id, key) for key in deltas} | {(bot_id, identity)}, change)
+        if ingress:
+            await session.execute(
+                sa.update(receipt)
+                .where(*self._scope(receipt, workspace, bot_id), receipt.source_key == change.source_key)
+                .values(
+                    session_id=identity, candidate_snapshot=json.dumps(_snapshot(change)), source_timestamp=timestamp
+                )
+            )
+        else:
+            result = await session.execute(
+                self._insert(receipt)
+                .values(
+                    workspace_uuid=workspace,
+                    bot_id=bot_id,
+                    session_id=identity,
+                    source_key=change.source_key,
+                    event_id=change.event_id,
+                    role=change.role,
+                    covered=False,
+                    timestamp=timestamp,
+                )
+                .on_conflict_do_nothing(index_elements=RECEIPT_KEY)
+            )
+            claimed = bool(result.rowcount)
+            if claimed:
+                await self._merge_base(session, workspace, bot_id, identity, _snapshot(change))
+            else:
+                previous = (
+                    await session.execute(
+                        sa.select(receipt.session_id, receipt.role, receipt.event_id).where(
+                            *self._scope(receipt, workspace, bot_id), receipt.source_key == change.source_key
+                        )
+                    )
+                ).one()
+                if tuple(previous) != (identity, change.role, change.event_id):
+                    raise ValueError('A summary message identity cannot change')
+        deltas[identity] = deltas.get(identity, 0) + int(claimed)
+        for owner, delta in sorted(deltas.items()):
+            await self._refresh(
+                session,
+                workspace,
+                bot_id,
+                owner,
+                delta,
+                backfill=backfill,
+                observed=timestamp if owner == identity else None,
+            )
+        return claimed
+
+    async def expire_ingress(self, session: AsyncSession, workspace: str, cutoff, keys) -> int:
+        """Seal expired candidates before deleting their event anchors, in one transaction."""
+        if not keys:
+            return 0
+        receipt = MonitoringSessionReceipt
+        rows = (
+            (
+                await session.execute(
+                    sa.select(receipt.__table__)
+                    .where(
+                        receipt.workspace_uuid == workspace,
+                        receipt.source_key.startswith('event:'),
+                        receipt.timestamp < cutoff,
+                        sa.tuple_(receipt.bot_id, receipt.source_key).in_(keys),
+                    )
+                    .order_by(receipt.bot_id, receipt.source_key)
+                    .with_for_update(skip_locked=True)
+                )
+            )
+            .mappings()
+            .all()
+        )
+        owners = {(row['bot_id'], row['session_id']) for row in rows if row['session_id'] is not None}
+        await self._lock_summaries(session, workspace, owners)
+        for row in rows:
+            if row['candidate_snapshot'] is not None:
+                await self._merge_base(
+                    session, workspace, row['bot_id'], row['session_id'], json.loads(row['candidate_snapshot'])
+                )
+            await session.execute(
+                sa.delete(receipt).where(
+                    *self._scope(receipt, workspace, row['bot_id']), receipt.source_key == row['source_key']
+                )
+            )
+        for bot_id, identity in sorted(owners):
+            await self._refresh(session, workspace, bot_id, identity)
+        return len(rows)

src/langbot/pkg/persistence/tenant_uow.py

@@ -71,6 +71,7 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
     'monitoring_llm_calls': 'workspace_uuid',
     'monitoring_tool_calls': 'workspace_uuid',
     'monitoring_sessions': 'workspace_uuid',
+    'monitoring_session_receipts': 'workspace_uuid',
     'monitoring_errors': 'workspace_uuid',
     'monitoring_embedding_calls': 'workspace_uuid',
     'monitoring_feedback': 'workspace_uuid',

src/langbot/pkg/pipeline/monitoring_helper.py

@@ -130,7 +130,7 @@ class MonitoringHelper:
     ) -> str:
         """Record the start of query processing, returns message_id"""
         try:
-            # Check if session exists, if not, record session start
+            # Message persistence updates the summary in the same transaction.
             session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
 
             # Get sender name from message event
@@ -171,31 +171,6 @@ class MonitoringHelper:
                 variables=None,  # Will be updated in record_query_success
             )
 
-            # Update session activity or create new session if it doesn't exist
-            # Always pass pipeline info to handle pipeline switches
-            session_updated = await ap.monitoring_service.update_session_activity(
-                get_query_execution_context(query),
-                session_id,
-                bot_id=bot_id,
-                pipeline_id=pipeline_id,
-                pipeline_name=pipeline_name,
-            )
-            if not session_updated:
-                # Session doesn't exist, create it
-                await ap.monitoring_service.record_session_start(
-                    get_query_execution_context(query),
-                    session_id=session_id,
-                    bot_id=bot_id,
-                    bot_name=bot_name,
-                    pipeline_id=pipeline_id,
-                    pipeline_name=pipeline_name,
-                    platform=query.launcher_type.value
-                    if hasattr(query.launcher_type, 'value')
-                    else str(query.launcher_type),
-                    user_id=query.sender_id,
-                    user_name=sender_name,
-                )
-
             return message_id
         except Exception as e:
             ap.logger.error(f'Failed to record query start: {e}')

src/langbot/pkg/platform/botmgr.py

@@ -1386,26 +1386,6 @@ class RuntimeBot:
                 user_name=sender_name,
             )
 
-            # Ensure the session exists so the message appears in the session monitor.
-            # Don't overwrite pipeline info — a session may have messages from
-            # multiple pipelines; discarding shouldn't change the displayed pipeline.
-            session_updated = await self.ap.monitoring_service.update_session_activity(
-                self.execution_context,
-                session_id,
-            )
-            if not session_updated:
-                # No session yet (first message for this launcher was discarded).
-                await self.ap.monitoring_service.record_session_start(
-                    self.execution_context,
-                    session_id=session_id,
-                    bot_id=self.bot_entity.uuid,
-                    bot_name=self.bot_entity.name or self.bot_entity.uuid,
-                    pipeline_id=self.PIPELINE_DISCARD,
-                    pipeline_name=self.PIPELINE_DISCARD_DISPLAY_NAME,
-                    platform=platform,
-                    user_id=str(sender_id),
-                    user_name=sender_name,
-                )
         except Exception as e:
             await self.logger.error(f'Failed to record discarded message: {e}')
 

tests/integration/api/test_monitoring.py

@@ -300,6 +300,17 @@ class TestMonitoringSessionsEndpoint:
         assert kwargs['limit'] == 20
         assert kwargs['offset'] == 40
 
+    @pytest.mark.asyncio
+    @pytest.mark.session_summary
+    @pytest.mark.parametrize('query', ['mode=real', 'mode=debug', 'status=failed'])
+    async def test_session_list_rejects_execution_filters(self, quart_test_client, fake_monitoring_app, query):
+        response = await quart_test_client.get(
+            '/api/v1/monitoring/sessions?' + query,
+            headers={'Authorization': 'Bearer test_token'},
+        )
+        assert response.status_code == 400
+        fake_monitoring_app.monitoring_service.get_sessions.assert_not_awaited()
+
 
 @pytest.mark.usefixtures('mock_circular_import_chain')
 class TestMonitoringErrorsEndpoint:
@@ -364,6 +375,98 @@ class TestMonitoringAllDataEndpoint:
             'errors': 2,
         }
 
+    @pytest.mark.asyncio
+    @pytest.mark.session_summary
+    @pytest.mark.contract('reader_http')
+    @pytest.mark.parametrize(
+        'query,mode,statuses',
+        [
+            ('/data?mode=all', 'all', None),
+            ('/data?mode=real&status=failed', 'real', ['failed']),
+            ('/data?mode=debug&status=running', 'debug', ['running']),
+            ('/data?status=completed&status=failed', 'all', ['completed', 'failed']),
+            ('/sessions?mode=all&userQuery=Alice', None, None),
+        ],
+    )
+    async def test_http_session_queries_only_read_summaries(
+        self,
+        quart_test_client,
+        fake_monitoring_app,
+        monkeypatch,
+        query,
+        mode,
+        statuses,
+    ):
+        """Other /data records may read history; its session query may not."""
+        import datetime
+        import sqlite3
+        from contextlib import asynccontextmanager
+        import sqlalchemy as sa
+        from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+        from langbot.pkg.api.http.service.monitoring import MonitoringService
+        from langbot.pkg.entity.persistence.monitoring import MonitoringSession
+        from langbot.pkg.persistence.session_summary_reader import SqlSessionSummaryReader
+
+        engine = create_async_engine('sqlite+aiosqlite:///:memory:')
+        try:
+            async with engine.begin() as connection:
+                await connection.run_sync(MonitoringSession.__table__.create)
+                await connection.execute(
+                    sa.insert(MonitoringSession).values(
+                        workspace_uuid='00000000-0000-0000-0000-00000000000a',
+                        bot_id='bot-a',
+                        session_id='summary',
+                        bot_name='Bot',
+                        pipeline_id='pipeline-a',
+                        pipeline_name='Pipeline',
+                        message_count=7,
+                        start_time=datetime.datetime(2026, 1, 2),
+                        last_activity=datetime.datetime(2026, 1, 2),
+                        user_name='Alice',
+                    )
+                )
+
+            @asynccontextmanager
+            async def scope(workspace):
+                assert workspace == '00000000-0000-0000-0000-00000000000a'
+                async with AsyncSession(engine) as session:
+                    yield session
+
+            service = MonitoringService(fake_monitoring_app)
+            service._session_list = SqlSessionSummaryReader(scope)
+            monkeypatch.setattr(fake_monitoring_app.monitoring_service, 'get_sessions', service.get_sessions)
+
+            def authorize(action, table, column, database, trigger):
+                if action == sqlite3.SQLITE_READ and table != 'monitoring_sessions':
+                    return sqlite3.SQLITE_DENY
+                return sqlite3.SQLITE_OK
+
+            async with engine.connect() as connection:
+                await connection.run_sync(
+                    lambda sync: sync.connection.dbapi_connection.run_async(
+                        lambda raw: raw._execute(raw._conn.set_authorizer, authorize)
+                    )
+                )
+            with patch(
+                'langbot.pkg.api.http.controller.groups.monitoring.get_traffic_series',
+                new=AsyncMock(return_value={}),
+            ):
+                response = await quart_test_client.get(
+                    '/api/v1/monitoring' + query + '&botId=bot-a&pipelineId=pipeline-a'
+                    '&startTime=2026-01-01T00:00:00&endTime=2026-01-03T00:00:00',
+                    headers={'Authorization': 'Bearer test_token'},
+                )
+            assert response.status_code == 200
+            result = (await response.get_json())['data']
+            assert [row['session_id'] for row in result['sessions']] == ['summary']
+            assert result['sessions'][0]['message_count'] == 7
+            assert (result['totalCount']['sessions'] if mode is not None else result['total']) == 1
+            if mode is not None:
+                kwargs = fake_monitoring_app.monitoring_service.get_messages.await_args.kwargs
+                assert kwargs['mode'] == mode and kwargs['execution_statuses'] == statuses
+        finally:
+            await engine.dispose()
+
 
 @pytest.mark.usefixtures('mock_circular_import_chain')
 class TestMonitoringDetailsEndpoints:

tests/integration/persistence/test_monitoring_postgres.py

@@ -224,7 +224,7 @@ def _query(context, sender_id):
 
 
 @pytest.mark.parametrize('user_id', [123456789, -100123456789, 0, None, '', '00123', '  opaque用户  '])
-@pytest.mark.parametrize('record_type', ['message', 'session', 'feedback'])
+@pytest.mark.parametrize('record_type', ['message', 'feedback'])
 async def test_optional_user_ids_round_trip_through_asyncpg(service, user_id, record_type):
     context = _context(WORKSPACE_A)
     expected = str(user_id) if isinstance(user_id, int) else user_id
@@ -238,8 +238,6 @@ async def test_optional_user_ids_round_trip_through_asyncpg(service, user_id, re
         )
         details = await _read(service, 'get_message_details', context, record_id)
         assert details['message']['user_id'] == expected
-    elif record_type == 'session':
-        await service.record_session_start(context, **RESOURCE, session_id='same-session', user_id=user_id)
         rows, total = await _read(service, 'get_sessions', context)
         assert total == 1
         assert rows[0]['user_id'] == expected
@@ -302,7 +300,7 @@ async def test_query_error_persists_error_message_and_linked_log(service, user_i
 
 
 @pytest.mark.parametrize('user_id', [True, 1.5, b'123', ['123']])
-@pytest.mark.parametrize('record_type', ['message', 'session', 'feedback'])
+@pytest.mark.parametrize('record_type', ['message', 'feedback'])
 async def test_unsupported_user_ids_fail_at_the_write_boundary(service, user_id, record_type):
     context = _context(WORKSPACE_A)
     with pytest.raises(TypeError, match='user_id must be a string, integer, or None'):
@@ -314,8 +312,6 @@ async def test_unsupported_user_ids_fail_at_the_write_boundary(service, user_id,
                 session_id='session',
                 user_id=user_id,
             )
-        elif record_type == 'session':
-            await service.record_session_start(context, **RESOURCE, session_id='session', user_id=user_id)
         else:
             await service.record_feedback(context, feedback_id='feedback', feedback_type=1, user_id=user_id)
     async with service.ap.persistence_mgr.tenant_scope(WORKSPACE_A):
@@ -326,7 +322,6 @@ async def test_unsupported_user_ids_fail_at_the_write_boundary(service, user_id,
 
 async def test_session_analysis_aggregates_under_cloud_sql_guard(service):
     context = _context(WORKSPACE_A)
-    await service.record_session_start(context, **RESOURCE, session_id='same-session')
     await service.record_message(context, **RESOURCE, session_id='same-session', message_content='hello')
     result = await _read(service, 'get_session_analysis', context, 'same-session')
     assert result['found'] is True

tests/integration/persistence/test_resource_tenancy_migration.py

@@ -107,7 +107,10 @@ async def test_legacy_sqlite_resources_are_backfilled_and_contracted(tmp_path):
             )
             assert (
                 await conn.scalar(
-                    sa.text('SELECT COUNT(*) FROM workspace_metadata WHERE workspace_uuid = :workspace_uuid'),
+                    sa.text(
+                        'SELECT COUNT(*) FROM workspace_metadata WHERE workspace_uuid = :workspace_uuid '
+                        "AND key IN ('wizard_status', 'wizard_progress', 'rag_plugin_migration_needed')"
+                    ),
                     {'workspace_uuid': workspace_uuid},
                 )
                 == 3
@@ -296,7 +299,15 @@ async def test_fresh_sqlite_schema_matches_resource_tenancy_contract(tmp_path):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "fresh-resources.db"}')
     try:
         async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
+            # Match PersistenceManager: 0040 creates the receipt FK after older PK migrations.
+            await conn.run_sync(
+                lambda sync: Base.metadata.create_all(
+                    sync,
+                    tables=[
+                        table for table in Base.metadata.sorted_tables if table.name != 'monitoring_session_receipts'
+                    ],
+                )
+            )
         await run_alembic_stamp(engine, '0001_baseline')
         await run_alembic_upgrade(engine, 'head')
 
@@ -312,3 +323,76 @@ async def test_fresh_sqlite_schema_matches_resource_tenancy_contract(tmp_path):
                 assert columns['legacy_vector_collection']['nullable'] is False
     finally:
         await engine.dispose()
+
+
+@pytest.mark.session_summary
+@pytest.mark.contract('summary_schema')
+@pytest.mark.parametrize('has_history', [False, True])
+async def test_legacy_sqlite_without_monitoring_tables_initializes_summaries(tmp_path, monkeypatch, has_history):
+    import datetime
+    import logging
+    from types import SimpleNamespace
+
+    from langbot.pkg.api.http.context import ExecutionContext
+    from langbot.pkg.api.http.service.monitoring import MonitoringService
+    from langbot.pkg.entity.persistence.event_log import EventLog
+    from langbot.pkg.persistence.mgr import PersistenceManager
+    from langbot.pkg.persistence.session_summary_backfill import SessionSummaryBackfill, require_initialized
+    from langbot.pkg.utils import constants
+
+    instance = str(uuid.uuid4())
+    monkeypatch.setattr(constants, 'instance_id', instance)
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "legacy-without-monitoring.db"}')
+    try:
+        await create_legacy_resource_schema(engine, instance_uuid=instance)
+        async with engine.begin() as conn:
+            for name in TENANT_TABLES:
+                if name.startswith('monitoring_'):
+                    await conn.exec_driver_sql(f'DROP TABLE {name}')
+            if has_history:
+                await conn.run_sync(EventLog.__table__.create)
+                await conn.execute(
+                    sa.insert(EventLog).values(
+                        event_id='legacy-event',
+                        event_type='message.received',
+                        source='platform',
+                        workspace_id=instance,
+                        bot_id='bot-1',
+                        conversation_id='group_legacy',
+                        created_at=datetime.datetime(2026, 1, 1),
+                    )
+                )
+        await run_alembic_stamp(engine, '0008_mcp_resource_prefs')
+        ap = SimpleNamespace(logger=logging.getLogger(__name__))
+        ap.persistence_mgr = PersistenceManager(ap)
+        ap.persistence_mgr.db = SimpleNamespace(get_engine=lambda: engine)
+        await ap.persistence_mgr._initialize_managed_schema()
+        await ap.persistence_mgr._enable_sqlite_foreign_keys_after_migration()
+        tables = await _inspect(engine, lambda conn: set(sa.inspect(conn).get_table_names()))
+        assert {'monitoring_sessions', 'monitoring_session_receipts'} <= tables
+        service = MonitoringService(ap)
+        if has_history:
+            with pytest.raises(RuntimeError, match='offline initialization'):
+                async with service._summary_session(instance) as session:
+                    await require_initialized(session, instance)
+        worker = SessionSummaryBackfill(service._summary_session, 'sqlite')
+        while not await worker.run_batch(instance):
+            pass
+        ctx = ExecutionContext(instance_uuid=instance, workspace_uuid=instance, placement_generation=1)
+        await service.record_message(
+            ctx,
+            bot_id='bot-1',
+            bot_name='Bot',
+            pipeline_id='pipeline-1',
+            pipeline_name='Pipeline',
+            session_id='group_live',
+            message_content='hello',
+        )
+        await ap.persistence_mgr._initialize_managed_schema()
+        rows, total = await service.get_sessions(ctx)
+        assert total == 1 + int(has_history)
+        assert all(row['message_count'] == 1 for row in rows)
+        async with engine.connect() as conn:
+            assert (await conn.exec_driver_sql('PRAGMA foreign_key_check')).all() == []
+    finally:
+        await engine.dispose()

tests/unit_tests/agent/test_event_log_transcript.py

@@ -78,7 +78,8 @@ class TestEventLogStore:
         store = EventLogStore(mock_db_engine)
 
         mock_session = AsyncMock()
-        mock_session.add = MagicMock()
+        mock_db_engine.dialect.name = 'sqlite'
+        mock_session.execute.return_value = MagicMock(rowcount=1)
         mock_session.commit = AsyncMock()
 
         with patch.object(store, '_session_factory') as mock_factory:
@@ -98,8 +99,8 @@ class TestEventLogStore:
             )
 
             assert event_id == 'evt_1'
-            stored_event = mock_session.add.call_args.args[0]
-            assert stored_event.metadata_json is None
+            stored_event = mock_session.execute.call_args.args[0].compile().params
+            assert stored_event['metadata_json'] is None
 
     @pytest.mark.asyncio
     async def test_append_event_stores_metadata_json(self, mock_db_engine):
@@ -109,7 +110,8 @@ class TestEventLogStore:
         store = EventLogStore(mock_db_engine)
 
         mock_session = AsyncMock()
-        mock_session.add = MagicMock()
+        mock_db_engine.dialect.name = 'sqlite'
+        mock_session.execute.return_value = MagicMock(rowcount=1)
         mock_session.commit = AsyncMock()
 
         with patch.object(store, '_session_factory') as mock_factory:
@@ -130,9 +132,9 @@ class TestEventLogStore:
             )
 
             assert event_id == 'evt_steering'
-            stored_event = mock_session.add.call_args.args[0]
-            assert '"status": "queued"' in stored_event.metadata_json
-            assert '"claimed_by_run_id": "run_1"' in stored_event.metadata_json
+            stored_event = mock_session.execute.call_args.args[0].compile().params
+            assert '"status": "queued"' in stored_event['metadata_json']
+            assert '"claimed_by_run_id": "run_1"' in stored_event['metadata_json']
 
 
 class TestTranscriptStore:

tests/unit_tests/api/service/test_monitoring_tenancy.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+import contextvars
 from types import SimpleNamespace
 
 import pytest
@@ -35,6 +36,12 @@ def _context(workspace_uuid: str) -> ExecutionContext:
 class _PersistenceManager:
     def __init__(self, engine):
         self.engine = engine
+        self.active_transaction = contextvars.ContextVar('monitoring-test-transaction', default=None)
+
+    def tenant_uow(self, workspace):
+        from langbot.pkg.persistence.tenant_uow import TenantUnitOfWork
+
+        return TenantUnitOfWork(self.engine, workspace, active_transaction=self.active_transaction)
 
     async def execute_async(self, *args, **kwargs):
         async with self.engine.connect() as connection:
@@ -112,22 +119,6 @@ async def test_same_session_and_resource_ids_do_not_collide(service):
     context_b = _context(WORKSPACE_B)
     message_a = await _record_message(service, context_a, 'tenant-a')
     message_b = await _record_message(service, context_b, 'tenant-b')
-    await service.record_session_start(
-        context_a,
-        session_id='same-session',
-        bot_id='same-bot',
-        bot_name='Same Bot',
-        pipeline_id='same-pipeline',
-        pipeline_name='Same Pipeline',
-    )
-    await service.record_session_start(
-        context_b,
-        session_id='same-session',
-        bot_id='same-bot',
-        bot_name='Same Bot',
-        pipeline_id='same-pipeline',
-        pipeline_name='Same Pipeline',
-    )
 
     messages_a, total_a = await service.get_messages(context_a)
     messages_b, total_b = await service.get_messages(context_b)
@@ -149,8 +140,9 @@ async def test_session_search_matches_user_id_or_name_within_workspace(service,
         (context_b, 'session-other-workspace', 'customer-42', 'Alice'),
     ]
     for context, session_id, user_id, user_name in fixtures:
-        await service.record_session_start(
+        await service.record_message(
             context,
+            message_content='hello',
             session_id=session_id,
             bot_id='same-bot',
             bot_name='Same Bot',
@@ -221,14 +213,6 @@ async def test_monitoring_queries_and_detail_views_are_strictly_bounded(service)
             'max_offset': 10,
         }
     }
-    await service.record_session_start(
-        context,
-        session_id='same-session',
-        bot_id='same-bot',
-        bot_name='Same Bot',
-        pipeline_id='same-pipeline',
-        pipeline_name='Same Pipeline',
-    )
     message_ids = [await _record_message(service, context, f'message-{index}') for index in range(4)]
     for index in range(3):
         await service.record_llm_call(
@@ -430,14 +414,26 @@ async def test_cloud_cleanup_uses_plan_retention_and_isolates_workspaces(service
     from unittest.mock import AsyncMock
 
     service.ap.deployment = SimpleNamespace(mode='cloud')
-    service.ap.entitlement_resolver = SimpleNamespace(resolve=AsyncMock(return_value=SimpleNamespace(limits={'chat_history.retention_days': 7})))
+    service.ap.entitlement_resolver = SimpleNamespace(
+        resolve=AsyncMock(return_value=SimpleNamespace(limits={'chat_history.retention_days': 7}))
+    )
     old = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta(days=10)
     for workspace in (WORKSPACE_A, WORKSPACE_B):
-        await service.ap.persistence_mgr.execute_async(sqlalchemy.insert(MonitoringMessage).values(
-            id=f'policy-{workspace}', workspace_uuid=workspace, timestamp=old,
-            bot_id='bot', bot_name='Bot', pipeline_id='pipeline', pipeline_name='Pipeline',
-            message_content='history', session_id='session', status='success', level='info',
-        ))
+        await service.ap.persistence_mgr.execute_async(
+            sqlalchemy.insert(MonitoringMessage).values(
+                id=f'policy-{workspace}',
+                workspace_uuid=workspace,
+                timestamp=old,
+                bot_id='bot',
+                bot_name='Bot',
+                pipeline_id='pipeline',
+                pipeline_name='Pipeline',
+                message_content='history',
+                session_id='session',
+                status='success',
+                level='info',
+            )
+        )
     deleted = await service.cleanup_expired_records(_context(WORKSPACE_A), retention_days=90)
     assert deleted['monitoring_messages'] == 1
     rows = await service.ap.persistence_mgr.execute_async(sqlalchemy.select(MonitoringMessage.workspace_uuid))

tests/unit_tests/api/test_mcp_server.py

@@ -4,7 +4,9 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
+from mcp.server.fastmcp.exceptions import ToolError
 
+from langbot.pkg.api.mcp import server as mcp_server
 from langbot.pkg.api.mcp.server import LangBotMCPServer
 
 
@@ -77,3 +79,29 @@ async def test_mcp_server_exposes_bot_event_route_tools():
         'th_TH',
         'vi_VN',
     }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'filters', [{}, {'bot_ids': ['bot-1'], 'user_query': 'Alice'}, {'mode': 'real'}, {'execution_statuses': ['failed']}]
+)
+async def test_monitoring_sessions_accept_only_conversation_filters(monkeypatch, filters):
+    app = _make_app()
+    app.monitoring_service = SimpleNamespace(
+        normalize_page_window=lambda limit, offset: (limit, offset),
+        get_sessions=AsyncMock(return_value=([], 0)),
+    )
+    context = SimpleNamespace(workspace_uuid='workspace-1')
+    monkeypatch.setattr(mcp_server, 'get_request_context', lambda: context)
+    monkeypatch.setattr(mcp_server, '_authorized', lambda permission: context)
+    monkeypatch.setattr(mcp_server, 'get_workspace_language', AsyncMock(return_value='en_US'))
+    server = LangBotMCPServer(app)
+
+    arguments = {'kind': 'sessions', 'filters': filters}
+    if set(filters) & {'mode', 'execution_statuses'}:
+        with pytest.raises(ToolError, match='Unsupported monitoring filter'):
+            await server.mcp.call_tool('get_monitoring_records', arguments)
+        app.monitoring_service.get_sessions.assert_not_awaited()
+    else:
+        await server.mcp.call_tool('get_monitoring_records', arguments)
+        app.monitoring_service.get_sessions.assert_awaited_once_with(context, limit=100, offset=0, **filters)

tests/unit_tests/core/test_app_resource_maintenance.py

@@ -32,11 +32,33 @@ async def _wait_forever() -> None:
     await asyncio.Event().wait()
 
 
+@pytest.mark.session_summary
+@pytest.mark.contract('summary_backfill')
+async def test_failed_summary_initialization_prevents_producer_startup(monkeypatch):
+    async def failed(_ap):
+        raise RuntimeError('summary initialization failed')
+
+    monkeypatch.setattr('langbot.pkg.persistence.session_summary_backfill.initialize_application_summaries', failed)
+    application = Application()
+    started = []
+    application.event_loop_monitor = SimpleNamespace(start=lambda: started.append(True))
+    with pytest.raises(RuntimeError, match='summary initialization failed'):
+        await application.run()
+    assert started == []
+
+
 async def test_resource_maintenance_waits_and_shares_workspace_discovery(monkeypatch) -> None:
     async def idle_leases(_ap):
         await _wait_forever()
 
     monkeypatch.setattr('langbot.pkg.utils.execution_lease.maintain_execution_leases', idle_leases)
+
+    async def initialized(_ap):
+        return None
+
+    monkeypatch.setattr(
+        'langbot.pkg.persistence.session_summary_backfill.initialize_application_summaries', initialized
+    )
     stop = asyncio.Event()
     completed = asyncio.Event()
     discovery_calls = 0
@@ -67,7 +89,7 @@ async def test_resource_maintenance_waits_and_shares_workspace_discovery(monkeyp
     application.event_loop = asyncio.get_running_loop()
     application.event_loop_monitor = SimpleNamespace(start=lambda: None)
     application.task_mgr = _TaskManager(stop)
-    application.plugin_connector = SimpleNamespace(initialize_plugins=lambda: asyncio.sleep(0))
+    application.plugin_connector = SimpleNamespace(initialize=lambda: asyncio.sleep(0), schedule_reconnect=lambda: None)
     application.platform_mgr = SimpleNamespace(run=_wait_forever)
     application.ctrl = SimpleNamespace(run=_wait_forever)
     application.http_ctrl = SimpleNamespace(run=_wait_forever)

tests/unit_tests/persistence/test_session_summary_contract.py

@@ -0,0 +1,144 @@
+"""Enforce the read boundary and the schema needed for bounded repair."""
+
+import ast
+import datetime
+import importlib
+import importlib.util
+import inspect
+
+import pytest
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence import monitoring as monitoring_models
+from langbot.pkg.entity.persistence.monitoring import MonitoringSession, MonitoringSessionReceipt
+from langbot.pkg.entity.persistence.workspace import Workspace
+from langbot.pkg.entity.persistence.metadata import WorkspaceMetadata
+from langbot.pkg.persistence import session_summary_reader, session_summary_contract
+from langbot.pkg.persistence.tenant_uow import TENANT_TABLE_COLUMNS
+
+
+pytestmark = pytest.mark.session_summary
+
+
+def assert_summary_dependencies(source, package):
+    """Restrict historical and application capabilities, not utility/type imports."""
+    forbidden = (
+        'langbot.pkg.agent',
+        'langbot.pkg.api',
+        'langbot.pkg.core',
+        'langbot.pkg.pipeline',
+        'langbot.pkg.persistence.mgr',
+        'langbot.pkg.persistence.session_summary_backfill',
+        'langbot.pkg.persistence.session_summary_writer',
+        'langbot.pkg.entity.persistence',
+    )
+    for node in ast.walk(ast.parse(source)):
+        if isinstance(node, ast.ImportFrom):
+            target = importlib.util.resolve_name('.' * node.level + (node.module or ''), package)
+            imports = [target + '.' + alias.name for alias in node.names]
+            if target == monitoring_models.__name__:
+                for alias in node.names:
+                    model = getattr(monitoring_models, alias.name, None)
+                    table = getattr(model, '__table__', None)
+                    assert alias.name != '*' and (table is None or table.name == 'monitoring_sessions')
+                continue
+        elif isinstance(node, ast.Import):
+            imports = [alias.name for alias in node.names]
+        else:
+            continue
+        for target in imports:
+            assert not any(target == prefix or target.startswith(prefix + '.') for prefix in forbidden), target
+            assert target != monitoring_models.__name__, 'Import the summary model without the history namespace'
+
+
+@pytest.mark.contract('reader_dependencies')
+def test_list_modules_have_no_history_or_application_capabilities():
+    for module in (session_summary_contract, session_summary_reader):
+        assert_summary_dependencies(inspect.getsource(module), module.__package__)
+
+
+@pytest.mark.parametrize(
+    'extra_import', ['from typing import Any', 'import functools', 'from datetime import timezone']
+)
+def test_summary_dependency_check_allows_utility_and_type_imports(extra_import):
+    assert_summary_dependencies(extra_import, session_summary_reader.__package__)
+
+
+@pytest.mark.parametrize(
+    'extra_import',
+    [
+        'from ..entity.persistence.monitoring import MonitoringMessage',
+        'from .session_summary_backfill import SessionSummaryBackfill',
+        'from ..entity.persistence import event_log',
+        'from ..entity.persistence.agent_run import AgentRun',
+    ],
+)
+def test_summary_dependency_check_rejects_history_and_repair(extra_import):
+    with pytest.raises(AssertionError):
+        assert_summary_dependencies(extra_import, session_summary_reader.__package__)
+
+
+@pytest.mark.contract('summary_schema')
+def test_summary_migration_is_idempotent_and_preserves_legacy_counts():
+    migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0040_session_summaries')
+    engine = sa.create_engine('sqlite:///:memory:')
+
+    @sa.event.listens_for(engine, 'connect')
+    def foreign_keys(connection, _record):
+        connection.execute('PRAGMA foreign_keys=ON')
+
+    try:
+        with engine.begin() as connection:
+            Base.metadata.create_all(connection)
+            with Operations.context(MigrationContext.configure(connection)):
+                migration.downgrade()
+            connection.execute(sa.insert(Workspace).values(uuid='w', instance_uuid='i', name='W', slug='w'))
+            connection.execute(
+                sa.text(
+                    'INSERT INTO monitoring_sessions (workspace_uuid,bot_id,session_id,bot_name,pipeline_id,pipeline_name,'
+                    'message_count,start_time,last_activity,is_active,user_id,user_name,platform) '
+                    "VALUES ('w','b','s','Bot','legacy','Legacy Pipeline',99,:now,:now,1,'legacy-user','Legacy User','legacy')"
+                ),
+                {'now': datetime.datetime(2026, 1, 1)},
+            )
+            with Operations.context(MigrationContext.configure(connection)):
+                migration.upgrade()
+                migration.upgrade()
+            row = connection.execute(sa.select(MonitoringSession)).mappings().one()
+            assert row['message_count'] == 99 and row['source_count'] == 0
+            assert row['metadata_role'] is None and row['metadata_timestamp'] is None
+            assert row['metadata_source_key'] is None and row['base_snapshot'] is None
+            assert (row['user_id'], row['user_name'], row['platform']) == ('legacy-user', 'Legacy User', 'legacy')
+            assert (row['pipeline_id'], row['pipeline_name']) == ('legacy', 'Legacy Pipeline')
+            assert connection.execute(sa.select(WorkspaceMetadata.value)).scalar_one() == '{"source": "messages"}'
+            inspector = sa.inspect(connection)
+            assert inspector.get_pk_constraint('monitoring_session_receipts')['constrained_columns'] == [
+                'workspace_uuid',
+                'bot_id',
+                'source_key',
+            ]
+            assert inspector.get_foreign_keys('monitoring_session_receipts')[0]['options']['ondelete'] == 'CASCADE'
+            assert TENANT_TABLE_COLUMNS['monitoring_session_receipts'] == 'workspace_uuid'
+            for table, index, fields in migration.INDEXES:
+                indexes = {item['name']: item['column_names'] for item in inspector.get_indexes(table)}
+                assert indexes[index] == fields
+            connection.execute(
+                sa.insert(MonitoringSessionReceipt).values(
+                    workspace_uuid='w',
+                    bot_id='b',
+                    source_key='event:detached',
+                    role='event',
+                    covered=True,
+                    timestamp=datetime.datetime(2026, 1, 1),
+                )
+            )
+            connection.execute(sa.delete(Workspace).where(Workspace.uuid == 'w'))
+            assert (
+                connection.execute(sa.select(sa.func.count()).select_from(MonitoringSessionReceipt)).scalar_one() == 0
+            )
+    finally:
+        engine.dispose()

tests/unit_tests/persistence/test_session_summary_snapshots.py

@@ -0,0 +1,553 @@
+"""Whole-source metadata, reversible ingress, and bounded candidate selection."""
+
+import datetime as dt
+import json
+import random
+from collections import defaultdict
+
+import pytest
+import sqlalchemy as sa
+
+from langbot.pkg.entity.persistence import monitoring as models
+from langbot.pkg.entity.persistence.event_log import EventLog
+from langbot.pkg.persistence.session_summary_writer import SessionSummaryWriter
+from tests.integration.persistence.test_monitoring_postgres import cloud_database as postgres_fixture
+from tests.unit_tests.api.service.test_monitoring_sessions import (
+    context,
+    rebuild_summaries,
+    replay_summary_message,
+    service as summary_service_fixture,
+    session_counts_from_facts,
+)
+
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.session_summary]
+service = summary_service_fixture
+cloud_database = postgres_fixture
+FIELDS = ('bot_name', 'platform', 'user_id', 'user_name', 'pipeline_id', 'pipeline_name')
+
+
+async def ingress(service, identity, user, pipeline, *, bot='a', owner='group'):
+    await service.record_ingress_event(
+        context(bot),
+        event_id=identity,
+        event_type='message.received',
+        bot_id=bot,
+        bot_name=user,
+        platform=user,
+        payload={},
+        conversation_id=owner,
+        actor_id=user,
+        actor_name=user,
+        routes=[{'target_uuid': pipeline, 'target_type': pipeline}] if pipeline else [],
+    )
+
+
+async def message(service, user, pipeline, *, bot='a', owner='topic', event=None, role='user'):
+    return await service.record_message(
+        context(bot),
+        bot_id=bot,
+        bot_name=user or '',
+        platform=user,
+        pipeline_id=pipeline,
+        pipeline_name=pipeline,
+        user_id=user,
+        user_name=None,
+        session_id=owner,
+        message_content='hello',
+        event_id=event,
+        role=role,
+    )
+
+
+async def projection(service):
+    return {
+        (row['bot_id'], row['session_id']): (row['message_count'], *(row[field] for field in FIELDS))
+        for row in (await service.get_sessions(context()))[0]
+    }
+
+
+async def facts(service):
+    """Calculate independently from retained sources, never receipts or Writer state."""
+    async with service.ap.persistence_mgr.get_db_engine().connect() as connection:
+        messages = (await connection.execute(sa.select(models.MonitoringMessage))).mappings().all()
+        events = (await connection.execute(sa.select(EventLog))).mappings().all()
+    messages = {row['id']: dict(row) for row in messages}
+    ingresses = {}
+    for row in events:
+        metadata = json.loads(row['metadata_json'] or '{}')
+        route = next((route for route in metadata.get('routes', []) if route.get('target_uuid')), {})
+        ingresses[row['event_id']] = dict(
+            bot_id=row['bot_id'],
+            session_id=row['conversation_id'],
+            timestamp=row['event_time'] or row['created_at'],
+            role='user',
+            bot_name=metadata.get('bot_name') or '',
+            platform=metadata.get('platform'),
+            user_id=row['actor_id'],
+            user_name=row['actor_name'],
+            pipeline_id=route.get('target_uuid') or '',
+            pipeline_name=route.get('target_type') or '',
+        )
+    counts = session_counts_from_facts(messages, ingresses)
+    covered = {
+        (row['bot_id'], alias)
+        for identity, row in messages.items()
+        if row['role'] != 'assistant'
+        for alias in (identity, row['event_id'])
+        if alias is not None
+    }
+    sources = defaultdict(list)
+    for identity, row in messages.items():
+        sources[(row['bot_id'], row['session_id'])].append(('message:' + identity, row))
+    for identity, row in ingresses.items():
+        if (row['bot_id'], identity) not in covered:
+            sources[(row['bot_id'], row['session_id'])].append(('event:' + identity, row))
+    expected = {}
+    for key, entries in sources.items():
+        _, selected = max(
+            entries,
+            key=lambda entry: (
+                entry[1]['role'] != 'assistant',
+                entry[1]['timestamp'],
+                entry[0].encode('utf-8'),
+            ),
+        )
+        expected[key] = (counts[key], *(selected[field] for field in FIELDS))
+    return expected
+
+
+@pytest.mark.contract('summary_fact_equivalence')
+@pytest.mark.contract('summary_snapshot')
+@pytest.mark.parametrize('transfer', ['message', 'late_link'])
+@pytest.mark.parametrize('base', ['message', 'event', 'expired_event', 'legacy'])
+async def test_transfer_restores_the_remaining_whole_snapshot(service, transfer, base):
+    receipt = models.MonitoringSessionReceipt
+    engine = service.ap.persistence_mgr.get_db_engine()
+    if base == 'legacy':
+        async with engine.begin() as connection:
+            await connection.execute(
+                sa.insert(models.MonitoringSession).values(
+                    workspace_uuid='workspace',
+                    bot_id='a',
+                    session_id='group',
+                    bot_name='Alice',
+                    platform='Alice',
+                    user_id='Alice',
+                    user_name='Alice',
+                    pipeline_id='A',
+                    pipeline_name='A',
+                    message_count=99,
+                    start_time=dt.datetime(2026, 1, 1),
+                    last_activity=dt.datetime(2026, 1, 1),
+                )
+            )
+    elif base == 'message':
+        await message(service, 'Alice', 'A', owner='group')
+    else:
+        await ingress(service, 'alice-event', 'Alice', 'A')
+        if base == 'expired_event':
+            assert (
+                await service._delete_expired_in_batches(
+                    context(),
+                    receipt,
+                    receipt.timestamp,
+                    receipt.source_key,
+                    dt.datetime(2100, 1, 1),
+                    1,
+                    1,
+                )
+                == 1
+            )
+    baseline = await projection(service)
+    await ingress(service, 'bob-event', 'Bob', 'B')
+    identity = await message(service, 'Bob', 'B', event='bob-event' if transfer == 'message' else None)
+    if transfer == 'late_link':
+        await service.link_execution_message(context(), identity, 'run', 'bob-event')
+        await service.link_execution_message(context(), identity, 'run', 'bob-event')
+    projected = await projection(service)
+    assert projected['a', 'group'] == baseline['a', 'group']
+    assert projected['a', 'topic'][0] == 1
+    if base in ('message', 'event'):
+        assert projected == await facts(service)
+        await rebuild_summaries(service, batch_size=1)
+        assert await projection(service) == projected
+
+
+@pytest.mark.contract('summary_retention')
+@pytest.mark.contract('summary_atomicity')
+@pytest.mark.contract('summary_snapshot')
+@pytest.mark.parametrize('rollback', [False, True])
+async def test_expiry_seals_the_selected_candidate_atomically(service, monkeypatch, rollback):
+    await ingress(service, 'alice-event', 'Alice', 'A')
+    before = await projection(service)
+    receipt = models.MonitoringSessionReceipt
+    async with service.ap.persistence_mgr.get_db_engine().begin() as connection:
+        await connection.execute(
+            sa.update(receipt)
+            .where(receipt.source_key == 'event:alice-event')
+            .values(
+                timestamp=dt.datetime(2000, 1, 1),
+            )
+        )
+    if rollback:
+        original = SessionSummaryWriter._merge_base
+
+        async def abort_after_seal(self, *args):
+            await original(self, *args)
+            raise RuntimeError('abort seal')
+
+        with monkeypatch.context() as patch:
+            patch.setattr(SessionSummaryWriter, '_merge_base', abort_after_seal)
+            with pytest.raises(RuntimeError, match='abort seal'):
+                await service.cleanup_expired_records(
+                    context(), retention_days=1, batch_size=1, max_batches_per_table=1
+                )
+        assert await projection(service) == before
+        async with service.ap.persistence_mgr.get_db_engine().connect() as connection:
+            assert (await connection.execute(sa.select(sa.func.count()).select_from(receipt))).scalar_one() == 1
+    deleted = await service.cleanup_expired_records(context(), retention_days=1, batch_size=1, max_batches_per_table=1)
+    assert deleted['monitoring_session_receipts'] == 1
+    assert await projection(service) == before
+    await ingress(service, 'bob-event', 'Bob', 'B')
+    identity = await message(service, 'Bob', 'B')
+    await service.link_execution_message(context(), identity, 'run', 'bob-event')
+    assert (await projection(service))['a', 'group'] == before['a', 'group']
+
+
+@pytest.mark.contract('summary_fact_equivalence')
+@pytest.mark.contract('summary_snapshot')
+@pytest.mark.parametrize('seed', [17, 43, 97])
+async def test_every_randomized_write_replay_and_link_matches_source_facts(service, seed):
+    rng = random.Random(seed)
+    events, messages = {}, []
+    for index in range(45):
+        operation = rng.choice(('ingress', 'ingress', 'message', 'message', 'replay', 'link'))
+        bot = rng.choice(('a', 'b'))
+        if operation == 'ingress':
+            if events and rng.randrange(3) == 0:
+                event = rng.choice(list(events))
+                bot, owner, user, pipeline = events[event]
+            else:
+                event, owner = f'event-{index}', rng.choice(('group', 'topic'))
+                user, pipeline = rng.choice(('Alice', 'Bob', 'Carol')), rng.choice(('A', 'B', '', '__discard__'))
+                events[event] = bot, owner, user, pipeline
+            await ingress(service, event, user, pipeline, bot=bot, owner=owner)
+        elif operation == 'message':
+            event = rng.choice(list(events)) if events and rng.randrange(2) else None
+            if event:
+                bot = events[event][0]
+            identity = await message(
+                service,
+                rng.choice(('Alice', None)),
+                rng.choice(('A', 'B', '', '__discard__')),
+                bot=bot,
+                owner=rng.choice(('group', 'topic')),
+                event=event,
+                role=rng.choice(('user', 'assistant')),
+            )
+            messages.append(identity)
+        elif operation == 'replay' and messages:
+            await replay_summary_message(service, 'workspace', rng.choice(messages))
+        elif operation == 'link' and events and messages:
+            async with service.ap.persistence_mgr.get_db_engine().connect() as connection:
+                unlinked = (
+                    await connection.execute(
+                        sa.select(models.MonitoringMessage.id, models.MonitoringMessage.bot_id).where(
+                            models.MonitoringMessage.event_id.is_(None),
+                            models.MonitoringMessage.role != 'assistant',
+                        )
+                    )
+                ).all()
+            if unlinked:
+                identity, bot = rng.choice(unlinked)
+                matching = [key for key, value in events.items() if value[0] == bot]
+                if matching:
+                    await service.link_execution_message(context(), identity, 'run', rng.choice(matching))
+        assert await projection(service) == await facts(service), (seed, index, operation)
+    expected = await projection(service)
+    await rebuild_summaries(service, batch_size=1)
+    assert await projection(service) == expected
+
+
+@pytest.mark.contract('summary_retention')
+@pytest.mark.contract('summary_snapshot')
+async def test_receipt_expiry_uses_source_identity_instead_of_message_role(service):
+    await message(service, 'Alice', 'A', owner='group', role='event')
+    before = await projection(service)
+    receipt = models.MonitoringSessionReceipt
+    async with service.ap.persistence_mgr.get_db_engine().begin() as connection:
+        await connection.execute(
+            sa.update(receipt)
+            .where(receipt.source_key.startswith('message:'))
+            .values(timestamp=dt.datetime(2000, 1, 1))
+        )
+    deleted = await service.cleanup_expired_records(context(), 1, batch_size=1, max_batches_per_table=1)
+    assert deleted['monitoring_session_receipts'] == 1
+    assert await projection(service) == before
+
+
+@pytest.mark.contract('summary_retention')
+@pytest.mark.contract('summary_snapshot')
+async def test_candidate_selection_work_does_not_grow_with_ingress_history(service):
+    receipt = models.MonitoringSessionReceipt
+    engine = service.ap.persistence_mgr.get_db_engine()
+    await ingress(service, 'seed', 'Alice', 'A')
+    measurements = []
+    for size in (100, 10000):
+        timestamp = dt.datetime(2026, 1, 1)
+        async with engine.begin() as connection:
+            await connection.execute(sa.delete(receipt))
+            await connection.execute(
+                sa.insert(receipt),
+                [
+                    dict(
+                        workspace_uuid='workspace',
+                        bot_id='a',
+                        session_id='group',
+                        source_key=f'event:{index:08}',
+                        role='event',
+                        covered=False,
+                        timestamp=timestamp,
+                        source_timestamp=timestamp,
+                        candidate_snapshot=json.dumps(
+                            dict(
+                                source_key=f'event:{index:08}',
+                                timestamp=timestamp.isoformat(),
+                                role='user',
+                                bot_name='Alice',
+                                platform=None,
+                                user_id=str(index),
+                                user_name=None,
+                                pipeline_id='',
+                                pipeline_name='',
+                            )
+                        ),
+                    )
+                    for index in range(size)
+                ],
+            )
+        steps = [0]
+
+        def tick():
+            steps[0] += 1
+            return 0
+
+        async def handler(callback):
+            async with engine.connect() as connection:
+                await connection.run_sync(
+                    lambda sync: sync.connection.dbapi_connection.run_async(
+                        lambda raw: raw._execute(raw._conn.set_progress_handler, callback, 1),
+                    )
+                )
+
+        await handler(tick)
+        try:
+            async with service._summary_session('workspace') as session:
+                await SessionSummaryWriter('sqlite')._refresh(session, 'workspace', 'a', 'group')
+        finally:
+            await handler(None)
+        measurements.append(steps[0])
+        assert (await service.get_sessions(context()))[0][0]['user_id'] == str(size - 1)
+    assert measurements[1] <= measurements[0] + 30, measurements
+
+
+@pytest.mark.asyncio(loop_scope='module')
+@pytest.mark.postgres
+@pytest.mark.contract('summary_postgres')
+@pytest.mark.contract('summary_retention')
+@pytest.mark.contract('summary_snapshot')
+async def test_postgres_expiry_and_late_link_restore_the_same_snapshot(cloud_database, monkeypatch):
+    """Force a late link to wait on a summary locked by ingress expiry."""
+    import asyncio
+    from tests.integration.persistence.test_monitoring_postgres import WORKSPACE_A, _context, _read
+
+    ap, admin = cloud_database
+    monitoring = ap.monitoring_service
+    ctx = _context(WORKSPACE_A)
+    bot = 'snapshot-expiry-race'
+    for user, pipeline in [('Alice', 'A'), ('Bob', 'B')]:
+        await monitoring.record_ingress_event(
+            ctx,
+            event_id=bot + user,
+            event_type='message.received',
+            bot_id=bot,
+            bot_name=user,
+            platform=user,
+            payload={},
+            conversation_id='group',
+            actor_id=user,
+            actor_name=user,
+            routes=[{'target_uuid': pipeline, 'target_type': pipeline}],
+        )
+    identity = await monitoring.record_message(
+        ctx,
+        bot_id=bot,
+        bot_name='Bob',
+        pipeline_id='B',
+        pipeline_name='B',
+        platform='Bob',
+        user_id='Bob',
+        user_name='Bob',
+        session_id='topic',
+        message_content='hello',
+    )
+    receipt = models.MonitoringSessionReceipt
+    async with admin.begin() as connection:
+        await connection.execute(
+            sa.update(receipt)
+            .where(
+                receipt.workspace_uuid == WORKSPACE_A,
+                receipt.bot_id == bot,
+                receipt.source_key == 'event:' + bot + 'Alice',
+            )
+            .values(timestamp=dt.datetime(2000, 1, 1))
+        )
+    sealed, release, attempted = (asyncio.Event() for _ in range(3))
+    original = SessionSummaryWriter._merge_base
+    cover_task = None
+    cover_pid = None
+
+    async def pause_seal(self, session, workspace, bot_id, owner, snapshot):
+        await original(self, session, workspace, bot_id, owner, snapshot)
+        if bot_id == bot and snapshot['source_key'] == 'event:' + bot + 'Alice':
+            sealed.set()
+            await asyncio.wait_for(release.wait(), 5)
+
+    def observe_sql(connection, cursor, sql, parameters, execution, executemany):
+        nonlocal cover_pid
+        statement = getattr(execution.compiled, 'statement', None)
+        if asyncio.current_task() is cover_task and getattr(statement, '_for_update_arg', None) is not None:
+            cover_pid = connection.connection.driver_connection.get_server_pid()
+            attempted.set()
+
+    monkeypatch.setattr(SessionSummaryWriter, '_merge_base', pause_seal)
+    engine = ap.persistence_mgr.get_db_engine()
+    sa.event.listen(engine.sync_engine, 'before_cursor_execute', observe_sql)
+    expiry_task = asyncio.create_task(monitoring.cleanup_expired_records(ctx, 1, batch_size=1, max_batches_per_table=1))
+    try:
+        await asyncio.wait_for(sealed.wait(), 5)
+        cover_task = asyncio.create_task(monitoring.link_execution_message(ctx, identity, 'run', bot + 'Bob'))
+        await asyncio.wait_for(attempted.wait(), 5)
+        async with asyncio.timeout(5):
+            while True:
+                async with admin.connect() as connection:
+                    if await connection.scalar(
+                        sa.text("SELECT wait_event_type='Lock' FROM pg_stat_activity WHERE pid=:pid"),
+                        {'pid': cover_pid},
+                    ):
+                        break
+                await asyncio.sleep(0.01)
+        release.set()
+        deleted, _ = await asyncio.wait_for(asyncio.gather(expiry_task, cover_task), 10)
+        assert deleted['monitoring_session_receipts'] == 1
+    finally:
+        release.set()
+        sa.event.remove(engine.sync_engine, 'before_cursor_execute', observe_sql)
+        tasks = [task for task in (expiry_task, cover_task) if task is not None]
+        for task in tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+    rows, _ = await _read(monitoring, 'get_sessions', ctx, bot_ids=[bot])
+    rows = {row['session_id']: row for row in rows}
+    assert rows['group']['message_count'] == rows['topic']['message_count'] == 1
+    assert tuple(rows['group'][field] for field in FIELDS) == ('Alice', 'Alice', 'Alice', 'Alice', 'A', 'A')
+
+
+@pytest.mark.asyncio(loop_scope='module')
+@pytest.mark.postgres
+@pytest.mark.contract('summary_postgres')
+@pytest.mark.contract('summary_snapshot')
+async def test_postgres_candidate_query_uses_a_bounded_index_scan(cloud_database):
+    """Explain the Writer's actual candidate query, rather than a test-only query."""
+    ap, admin = cloud_database
+    from tests.integration.persistence.test_monitoring_postgres import WORKSPACE_A, _context
+
+    monitoring = ap.monitoring_service
+    ctx = _context(WORKSPACE_A)
+    bot = 'snapshot-scale'
+    await monitoring.record_ingress_event(
+        ctx,
+        event_id=bot,
+        event_type='message.received',
+        bot_id=bot,
+        bot_name='Bot',
+        platform=None,
+        payload={},
+        conversation_id='group',
+        actor_id='Alice',
+    )
+    receipt = models.MonitoringSessionReceipt
+    captured = []
+
+    def observe_sql(connection, statement, multiparams, params, execution_options):
+        # before_cursor_execute can expose an older statement's bound values
+        # from the compilation cache. Explain this invocation's statement.
+        if getattr(statement, 'is_select', False) and 'candidate_snapshot' in statement.selected_columns:
+            captured.append(statement)
+
+    engine = ap.persistence_mgr.get_db_engine()
+    sa.event.listen(engine.sync_engine, 'before_execute', observe_sql)
+    try:
+        for size in (100, 10000):
+            timestamp = dt.datetime(2026, 1, 1)
+            async with admin.begin() as connection:
+                await connection.execute(
+                    sa.delete(receipt).where(receipt.workspace_uuid == WORKSPACE_A, receipt.bot_id == bot)
+                )
+                await connection.execute(
+                    sa.insert(receipt),
+                    [
+                        dict(
+                            workspace_uuid=WORKSPACE_A,
+                            bot_id=bot,
+                            session_id='group',
+                            source_key=f'event:{i:08}',
+                            role='event',
+                            covered=False,
+                            timestamp=timestamp,
+                            source_timestamp=timestamp,
+                            candidate_snapshot=json.dumps(
+                                dict(
+                                    source_key=f'event:{i:08}',
+                                    timestamp=timestamp.isoformat(),
+                                    role='user',
+                                    bot_name='Bot',
+                                    platform=None,
+                                    user_id='Alice',
+                                    user_name=None,
+                                    pipeline_id='',
+                                    pipeline_name='',
+                                )
+                            ),
+                        )
+                        for i in range(size)
+                    ],
+                )
+                await connection.execute(sa.text('ANALYZE monitoring_session_receipts'))
+            captured.clear()
+            async with monitoring._summary_session(WORKSPACE_A) as session:
+                await monitoring._summary_writer()._refresh(session, WORKSPACE_A, bot, 'group')
+            # The summary read also selects base_snapshot; only receipt candidate
+            # queries have this selected column. A full candidate scan must fail.
+            assert len(captured) == 1
+            sql = str(captured[0].compile(dialect=admin.dialect, compile_kwargs={'literal_binds': True}))
+            async with admin.connect() as connection:
+                explain = (
+                    await connection.execute(sa.text('EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ' + sql))
+                ).scalar_one()
+            plan = (json.loads(explain) if isinstance(explain, str) else explain)[0]['Plan']
+
+            def walk(node):
+                yield node
+                for child in node.get('Plans', []):
+                    yield from walk(child)
+
+            nodes = list(walk(plan))
+            assert any(node.get('Index Name') == 'ix_session_receipts_candidate' for node in nodes), plan
+            assert not any(node['Node Type'] in ('Seq Scan', 'Sort') for node in nodes), plan
+            assert plan['Actual Rows'] == 1
+            assert plan.get('Shared Hit Blocks', 0) + plan.get('Shared Read Blocks', 0) < 25, plan
+    finally:
+        sa.event.remove(engine.sync_engine, 'before_execute', observe_sql)

tests/utils/required_suites.py

@@ -5,6 +5,20 @@ import pytest
 
 # These are behavior obligations, independent of file names and parametrization.
 REQUIRED_CONTRACTS = {
+    'session_summary': {
+        'reader_dependencies',
+        'reader_sql',
+        'reader_http',
+        'reader_scale',
+        'summary_accounting',
+        'summary_fact_equivalence',
+        'summary_snapshot',
+        'summary_atomicity',
+        'summary_backfill',
+        'summary_retention',
+        'summary_postgres',
+        'summary_schema',
+    },
     'postgres': {
         'admission_isolation',
         'rag_identity',