diff --git a/AGENTS.md b/AGENTS.md index bc016a26e..96f8c6903 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,6 +88,7 @@ Config keys to verify in `data/config.yaml` / `src/langbot/templates/config.yaml - New schema changes use Alembic under `src/langbot/pkg/persistence/alembic/versions/`. LangBot 4.x does not support upgrading 3.x databases. - 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). - 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: `(): `. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 9ad2a5a9b..f99f926a8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -267,12 +267,15 @@ Persistence is centered on `pkg/persistence/mgr.py`. - SQLite is the default database; PostgreSQL is supported. - Models live under `pkg/entity/persistence/`. +- `pkg/persistence/media.py::MediaReferenceText` validates bound conversation payloads on ORM/Core writes while preserving the existing TEXT schema and string API. It rejects recognized structured inline images: retained images go through `MediaCache.externalize_chain_dump()` before serialization, while metadata-only snapshots and transcripts use `redact_inline_images()`. Reads and comparisons preserve historical values for incremental cleanup. Runtime/adapter messages, arbitrary diagnostic text, and opaque strings are outside this contract. - Timezone-less SQL `DateTime` columns store UTC-naive values on both backends. Normalize aware values with `pkg/persistence/datetime_utils.py::as_naive_utc` at the bind boundary, including deadlines, leases, event times, and query/retention cutoffs; do not merely strip an offset. Existing naive rows represent UTC. Restore UTC awareness for application comparisons and epoch serialization. Runner ledger/event/transcript stores follow this existing schema contract without a data migration or changes to journal/tenant authorization. - Fresh schemas are created from current metadata, then Alembic migrations run to head. LangBot 4.x does not upgrade 3.x databases. - New schema changes use Alembic under `pkg/persistence/alembic/versions/`; there is no legacy migration chain in 4.x. Configuration starts from `src/langbot/templates/config.yaml` and is generated into `data/config.yaml` on first run. Most long-lived managers read from `ap.instance_config.data`. +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. + ## 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. diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index ae73d104c..ec47b8bb1 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -203,6 +203,9 @@ message or event identifier outside the current list page. The detail exposes `inputs`, `outputs` (generated content), `deliveries` (recorded platform sends), `events`, `llm_calls`, `tool_calls`, `errors`, `related`, and `conversation` in `pages`. Follow each section's `has_more` and `next_offset` independently. +Start with `section=overview` to load only inputs, outputs, deliveries and errors. +Request `conversation`, `related`, calls and events separately when needed; +omitting `section` retains the complete trace response for existing consumers. Conversation history supplies context; historical messages without explicit links must not be asserted to belong to the selected execution. Events without a processor run use `source=event`. All lookups remain Workspace-scoped. diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index b74ab8041..b744f865b 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -794,7 +794,9 @@ async def execute_platform_tool( bot_name=str(getattr(getattr(bot, 'bot_entity', None), 'name', bot_id)), pipeline_id=str(authorization.get('processor_id') or ''), pipeline_name=str(authorization.get('processor_id') or authorization.get('runner_id') or ''), - message_content=json.dumps(content, ensure_ascii=False), + message_content=content + if isinstance(content, (list, dict)) + else json.dumps(content, ensure_ascii=False), session_id=str(authorization.get('conversation_id') or ''), role='assistant', status='error' if delivery_error else 'success', diff --git a/src/langbot/pkg/agent/runner/run_journal.py b/src/langbot/pkg/agent/runner/run_journal.py index 3a0217f89..c2ea15df8 100644 --- a/src/langbot/pkg/agent/runner/run_journal.py +++ b/src/langbot/pkg/agent/runner/run_journal.py @@ -5,6 +5,7 @@ from __future__ import annotations import typing from ...core import app +from ...storage.media import MediaCache from .descriptor import RunnerDescriptor from .errors import RunnerProtocolError from .host_models import AgentBinding, AgentEventEnvelope @@ -44,10 +45,10 @@ class AgentRunJournal: if not item: return value item_type = item.get('type') - if item_type == 'image_base64' and item.get('image_base64'): - item['image_base64'] = None + if MediaCache.has_inline_images(item): + item = MediaCache.redact_inline_images(item) item['content_redacted'] = True - elif item_type == 'file_base64' and item.get('file_base64'): + if item_type == 'file_base64' and item.get('file_base64'): item['file_base64'] = None item['content_redacted'] = True return item @@ -77,6 +78,8 @@ class AgentRunJournal: return [cls._sanitize_event_data(item) for item in value] if not isinstance(value, dict): return value + if value.get('type') in ('Image', 'image', 'image_url', 'image_base64'): + value = cls._sanitize_content_item(value) result = {} for key, item in value.items(): if key in ('image_base64', 'file_base64', 'base64') and item: @@ -157,11 +160,16 @@ class AgentRunJournal: usage = result_dict.get('usage') if hasattr(usage, 'model_dump'): usage = usage.model_dump(mode='json') + data = result_dict.get('data') if isinstance(result_dict.get('data'), dict) else {} + if MediaCache.has_inline_images(data): + data = await self.ap.storage_mgr.media_cache.externalize_chain_dump(data) + if MediaCache.has_inline_images(metadata): + metadata = MediaCache.redact_inline_images(metadata) return await self._get_run_ledger_store().append_event( run_id=run_id, sequence=sequence, event_type=str(result_dict.get('type') or 'unknown'), - data=result_dict.get('data') if isinstance(result_dict.get('data'), dict) else {}, + data=data, usage=usage if isinstance(usage, dict) else None, source=source, metadata=metadata, diff --git a/src/langbot/pkg/agent/runner/run_ledger_store.py b/src/langbot/pkg/agent/runner/run_ledger_store.py index 2a8b9c75b..06ec1d969 100644 --- a/src/langbot/pkg/agent/runner/run_ledger_store.py +++ b/src/langbot/pkg/agent/runner/run_ledger_store.py @@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession from sqlalchemy.orm import sessionmaker from ...persistence.datetime_utils import as_naive_utc +from ...storage.media import MediaCache from ...utils.inflight import inflight_hub from ...utils.execution_lease import host_execution_lease @@ -384,7 +385,7 @@ class RunLedgerStore: if run is None or run.workspace_id != workspace_id or run.status in TERMINAL_STATUSES: raise ValueError('Skill activation requires an active run in the current Workspace') original = run.metadata_json - metadata = _json_loads(original, {}) + metadata = MediaCache.redact_inline_images(_json_loads(original, {})) bindings = metadata.setdefault('activated_skills', {}) if name in bindings: return str(bindings[name]) @@ -439,7 +440,8 @@ class RunLedgerStore: if cost is not None: run.cost_json = _json_dumps(cost) if metadata is not None: - existing_metadata = _json_loads(run.metadata_json, {}) + # Discard legacy image bytes before merging; new metadata still uses the write contract. + existing_metadata = MediaCache.redact_inline_images(_json_loads(run.metadata_json, {})) if isinstance(existing_metadata, dict): existing_metadata.update(metadata) run.metadata_json = _json_dumps(existing_metadata) diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 5cbdf56c6..d6feeb2bc 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/src/langbot/pkg/api/http/controller/groups/monitoring.py @@ -129,7 +129,7 @@ class MonitoringRouterGroup(group.RouterGroup): permission=Permission.RESOURCE_VIEW, ) async def get_execution_detail(source: str, execution_id: str, request_context: RequestContext) -> str: - """Full trace of a single execution (agent run or pipeline query).""" + """Read a full trace, its initial overview, or one paginated section.""" try: detail = await self.ap.monitoring_service.get_execution_detail( request_context, diff --git a/src/langbot/pkg/api/http/service/maintenance.py b/src/langbot/pkg/api/http/service/maintenance.py index 4fe3cc642..86485efa7 100644 --- a/src/langbot/pkg/api/http/service/maintenance.py +++ b/src/langbot/pkg/api/http/service/maintenance.py @@ -114,6 +114,10 @@ class MaintenanceService: } if media_cache is not None and is_singleton: result['media_files'] = media_cleanup.get('expired_deleted', 0) + media_cleanup.get('size_deleted', 0) + if media_cache is not None: + from ....storage.inline_media_cleanup import cleanup_inline_media + + result['inline_media_records'] = await cleanup_inline_media(self.ap, context, media_retention_days) return result async def get_storage_analysis(self, context: TenantContext) -> dict[str, Any]: diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 0ca23d7e1..0480a7d1b 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -15,6 +15,7 @@ from ....entity.persistence import monitoring as persistence_monitoring from ....entity.persistence import agent_run as persistence_agent_run from ....entity.persistence import agent_interaction as persistence_agent_interaction from ....entity.persistence.event_log import EventLog +from ....storage.media import MediaCache from ..authz import WorkspaceRequiredError from ..context import ExecutionContext from .tenant import TenantContext, require_workspace_uuid @@ -504,30 +505,31 @@ class MonitoringService(ExecutionDetailsMixin): def _sanitize_message_content(self, content: str) -> str: """Strip raw base64 data to protect database storage from unbounded bloating.""" - if not content or len(content) < 10000 or (';base64,' not in content and 'data:image/' not in content): + if not content or ('base64' not in content and 'data:image/' not in content): return content try: data = json.loads(content) - - def _strip_node(node): - if isinstance(node, list): - return [_strip_node(x) for x in node] - if isinstance(node, dict): - res = dict(node) - if res.get('type') == 'Image' and res.get('base64'): - res['base64'] = None - for k, v in list(res.items()): - if isinstance(v, (list, dict)): - res[k] = _strip_node(v) - return res - return node - - return json.dumps(_strip_node(data), ensure_ascii=False) + return json.dumps(MediaCache.redact_inline_images(data), ensure_ascii=False) except Exception: return re.sub( r'data:image/[a-zA-Z0-9.+_-]+;base64,[\sA-Za-z0-9+/=]{1000,}', '[base64 image omitted]', content ) + async def _externalize_message_content(self, content: str | list | dict) -> str: + """Externalize structured images before allocating their JSON representation.""" + if isinstance(content, str): + if not content or ('base64' not in content and 'data:image/' not in content): + return content + try: + payload = json.loads(content) + except (TypeError, ValueError): + return self._sanitize_message_content(content) + else: + payload = content + if MediaCache.has_inline_images(payload): + payload = await self.ap.storage_mgr.media_cache.externalize_chain_dump(payload) + return json.dumps(MediaCache.redact_inline_images(payload), ensure_ascii=False) + @_workspace_transaction async def record_message( self, @@ -536,7 +538,7 @@ class MonitoringService(ExecutionDetailsMixin): bot_name: str, pipeline_id: str, pipeline_name: str, - message_content: str, + message_content: str | list | dict, session_id: str, status: str = 'success', level: str = 'info', @@ -558,7 +560,7 @@ class MonitoringService(ExecutionDetailsMixin): event_id = event_id or (trace_state.event_id if trace_state else None) run_id = run_id or trace.current_run() or None message_id = str(uuid.uuid4()) - message_content = self._sanitize_message_content(message_content) + message_content = await self._externalize_message_content(message_content) from ....utils.execution_lease import host_execution_lease message_data = { @@ -578,7 +580,7 @@ class MonitoringService(ExecutionDetailsMixin): 'user_id': _normalize_user_id(user_id), 'user_name': user_name, 'runner_name': runner_name, - 'variables': variables, + 'variables': self._sanitize_message_content(variables) if variables is not None else None, 'role': role, 'event_id': event_id, 'run_id': run_id, @@ -886,7 +888,7 @@ class MonitoringService(ExecutionDetailsMixin): if level is not None: update_values['level'] = level if variables is not None: - update_values['variables'] = variables + update_values['variables'] = self._sanitize_message_content(variables) await self.ap.persistence_mgr.execute_async( sqlalchemy.update(persistence_monitoring.MonitoringMessage) @@ -1291,15 +1293,16 @@ class MonitoringService(ExecutionDetailsMixin): 'input_preview': _message_preview(json.dumps(metadata.get('input') or {}, ensure_ascii=False), 200), } - def _serialize_pipeline_execution(self, row) -> dict: + def _serialize_pipeline_execution(self, row, *, summary_only=False) -> dict: created_at_ms = _epoch_ms(row.timestamp) + preview = '' if summary_only else _message_preview(row.message_content) return { 'source': 'pipeline', 'id': row.id, 'event_id': row.event_id, 'status': row.status, 'status_group': _status_group(row.status, 'pipeline'), - 'title': _message_preview(row.message_content), + 'title': row.pipeline_name if summary_only else preview, 'target_kind': 'pipeline', 'target_id': row.pipeline_id, 'target_name': row.pipeline_name, @@ -1319,7 +1322,7 @@ class MonitoringService(ExecutionDetailsMixin): 'queue_name': None, 'debug': row.bot_id == 'websocket-proxy-bot', 'has_error': row.status == 'error', - 'input_preview': _message_preview(row.message_content), + 'input_preview': preview, 'platform': row.platform, 'user_id': row.user_id, 'user_name': row.user_name, @@ -1533,6 +1536,23 @@ class MonitoringService(ExecutionDetailsMixin): 'real': total - debug_count, } + @staticmethod + def _inflight_pipeline_condition(*, pending_only: bool = False): + """Historical pending records without a live Host lease are not tasks.""" + Message = persistence_monitoring.MonitoringMessage + live_lease = sqlalchemy.and_( + Message.execution_owner_id.is_not(None), + Message.execution_lease_expires_at > datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), + ) + # Active-only queries must expose the lease range to the composite index. + # The status OR otherwise reads historical payload pages to check role. + if pending_only: + return live_lease + return sqlalchemy.or_( + Message.status != 'pending', + live_lease, + ) + async def get_executions( self, context: TenantContext, @@ -1548,6 +1568,7 @@ class MonitoringService(ExecutionDetailsMixin): limit: int = 50, offset: int = 0, include_summary: bool = True, + for_inflight: bool = False, ) -> dict: """List executions from both object types (agent runs and pipeline queries).""" workspace_uuid = require_workspace_uuid(context) @@ -1579,15 +1600,18 @@ class MonitoringService(ExecutionDetailsMixin): ).where(*conditions) ) if source in {'all', 'pipeline'} and (not agent_ids or pipeline_ids): + pipeline_statuses = self._raw_pipeline_statuses(statuses) conditions = self._pipeline_execution_conditions( workspace_uuid, bot_ids=bot_ids, pipeline_ids=pipeline_ids, - statuses=self._raw_pipeline_statuses(statuses), + statuses=pipeline_statuses, start_time=start_time, end_time=end_time, debug=debug, ) + if for_inflight: + conditions.append(self._inflight_pipeline_condition(pending_only=pipeline_statuses == ['pending'])) branches.append( sqlalchemy.select( sqlalchemy.literal('pipeline').label('source'), @@ -1595,7 +1619,7 @@ class MonitoringService(ExecutionDetailsMixin): Message.timestamp.label('time'), ).where(*conditions) ) - if source in {'all', 'event'} and mode != 'debug' and not pipeline_ids and not agent_ids: + if source in {'all', 'event'} and mode != 'debug' and not pipeline_ids and not agent_ids and not for_inflight: conditions = self._unhandled_event_conditions(workspace_uuid, bot_ids, start_time, end_time) if statuses: conditions.append(self._execution_json_text(EventLog.metadata_json, 'status').in_(statuses)) @@ -1634,20 +1658,23 @@ class MonitoringService(ExecutionDetailsMixin): workspace_column = model.workspace_uuid if kind == 'pipeline' else model.workspace_id records = await self._execution_rows( workspace_uuid, - sqlalchemy.select(model).where(workspace_column == workspace_uuid, identity_column.in_(ids)), + self._execution_statement(model, summary_only=for_inflight).where( + workspace_column == workspace_uuid, identity_column.in_(ids) + ), + summary_only=for_inflight, ) names = await self._agent_names(workspace_uuid, records) if kind == 'agent' else {} for record in records: item = ( self._serialize_agent_execution(record, names) if kind == 'agent' - else self._serialize_pipeline_execution(record) + else self._serialize_pipeline_execution(record, summary_only=for_inflight) if kind == 'pipeline' else self._serialize_event_execution(record) ) by_key[(kind, item['id'])] = item page = [by_key[(ref.source, ref.id)] for ref in refs if (ref.source, ref.id) in by_key] - await self.enrich_execution_rows(workspace_uuid, page) + await self.enrich_execution_rows(workspace_uuid, page, summary_only=for_inflight) if not include_summary: return {'items': page, 'total': total, 'has_more': total > offset + len(page)} return { @@ -1692,7 +1719,9 @@ class MonitoringService(ExecutionDetailsMixin): from ....utils.inflight import inflight_hub workspace = require_workspace_uuid(context) - active = await self.get_executions(context, statuses=['running', 'queued'], limit=100, include_summary=False) + active = await self.get_executions( + context, statuses=['running', 'queued'], limit=100, include_summary=False, for_inflight=True + ) # Execution timestamps are stored as UTC without a timezone. asyncpg # rejects aware datetimes for these columns, even though SQLite accepts them. recent_start = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta( @@ -1703,6 +1732,7 @@ class MonitoringService(ExecutionDetailsMixin): start_time=recent_start, limit=100, include_summary=False, + for_inflight=True, ) items = {(r['source'], r['id']): r for r in recent['items'] + active['items'] if r['source'] != 'event'} # Keep the actual terminal state of older runs that just left the active set. @@ -1722,28 +1752,56 @@ class MonitoringService(ExecutionDetailsMixin): ]: missing = [r['id'] for r in (previous or [])[:100] if r['source'] == kind and (kind, r['id']) not in items] if missing: - records = await self._execution_rows( - workspace, sqlalchemy.select(model).where(scope == workspace, column.in_(missing)) + statement = self._execution_statement(model, summary_only=True).where( + scope == workspace, column.in_(missing) ) + if kind == 'pipeline': + statement = statement.where(self._inflight_pipeline_condition()) + records = await self._execution_rows(workspace, statement, summary_only=True) names = await self._agent_names(workspace, records) if kind == 'agent' else {} for record in records: row = ( self._serialize_agent_execution(record, names) if kind == 'agent' - else self._serialize_pipeline_execution(record) + else self._serialize_pipeline_execution(record, summary_only=True) ) items[(kind, row['id'])] = row rows = list(items.values()) + snapshot = { + 'items': rows, + 'active_total': active['total'], + 'truncated': active['has_more'] or recent['has_more'], + } + if not rows: + inflight_hub.track(workspace, []) + return snapshot Run = persistence_agent_run.AgentRun Message = persistence_monitoring.MonitoringMessage agent_ids = [r['id'] for r in rows if r['source'] == 'agent'] pipeline_ids = [r['id'] for r in rows if r['source'] == 'pipeline'] - linked = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(Run.run_id, Message.id.label('message_id')) - .outerjoin(Message, sqlalchemy.and_(Message.workspace_uuid == workspace, Message.run_id == Run.run_id)) - .where( - Run.workspace_id == workspace, sqlalchemy.or_(Run.run_id.in_(agent_ids), Message.id.in_(pipeline_ids)) + links = [] + if agent_ids: + links.append( + sqlalchemy.select(Run.run_id, sqlalchemy.literal(None).label('message_id')).where( + Run.workspace_id == workspace, Run.run_id.in_(agent_ids) + ) ) + if pipeline_ids: + # Start from the selected message IDs. A cross-table OR after a + # LEFT JOIN makes SQLite scan the attachment-heavy message ledger + # whenever the snapshot contains even one Pipeline execution. + selected_messages = sqlalchemy.union_all( + *[sqlalchemy.select(sqlalchemy.literal(message_id).label('message_id')) for message_id in pipeline_ids] + ).subquery('selected_messages') + links.append( + sqlalchemy.select(Run.run_id, Message.id.label('message_id')) + .select_from(selected_messages) + .join(Message, Message.id == selected_messages.c.message_id) + .join(Run, sqlalchemy.and_(Run.workspace_id == workspace, Run.run_id == Message.run_id)) + .where(Message.workspace_uuid == workspace) + ) + linked = await self.ap.persistence_mgr.execute_async( + links[0] if len(links) == 1 else sqlalchemy.union_all(*links) ) identities = { r.run_id: ('pipeline', r.message_id) if r.message_id in pipeline_ids else ('agent', r.run_id) @@ -1766,7 +1824,7 @@ class MonitoringService(ExecutionDetailsMixin): row = items.get(identities[run_id]) if row is not None: row['progress_event'] = event_type - return {'items': rows, 'active_total': active['total'], 'truncated': active['has_more'] or recent['has_more']} + return snapshot async def get_token_statistics( self, diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py index 467a8026f..9c65f3f2c 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -4,6 +4,7 @@ from __future__ import annotations import datetime import json +from types import SimpleNamespace from langbot_plugin.api.entities.builtin.platform.events import event_summary import sqlalchemy as sa @@ -99,7 +100,7 @@ def transcript_item(row): class ExecutionDetailsMixin: - def _session_message_source(self, workspace, bot_ids): + def _session_message_source(self, workspace, bot_ids, *, summary_only=False): """Combine delivered messages with ingress missing from the legacy ledger.""" message = models.MonitoringMessage conversation = sa.func.coalesce( @@ -147,7 +148,11 @@ class ExecutionDetailsMixin: 'event_id': EventLog.event_id, 'run_id': EventLog.run_id, } - columns = list(message.__table__.columns) + columns = [ + column + for column in message.__table__.columns + if not summary_only or column.name not in {'message_content', 'variables'} + ] stored = sa.select(*columns).where(message.workspace_uuid == workspace, message.bot_id.in_(bot_ids)) incoming = sa.select( *[values.get(column.name, sa.literal(None)).label(column.name) for column in columns] @@ -183,7 +188,9 @@ class ExecutionDetailsMixin: self, workspace, bot_ids, start_time, end_time, user_query, is_active, limit, offset ): """Read session summaries across processor types without mutating history.""" - messages = self._session_message_source(workspace, bot_ids) + # Window functions sort and retain their input rows. Carrying historical + # inline media through that sort can consume gigabytes for a summary. + messages = self._session_message_source(workspace, bot_ids, summary_only=True) key = [messages.c.workspace_uuid, messages.c.bot_id, messages.c.session_id] ranked = ( sa.select( @@ -354,8 +361,44 @@ class ExecutionDetailsMixin: async with AsyncSession(self.ap.persistence_mgr.get_db_engine()) as session: return await session.execute(statement) - async def _execution_rows(self, workspace, statement): - return list((await self._execution_query(workspace, statement)).scalars().all()) + async def _execution_rows(self, workspace, statement, *, summary_only=False): + result = await self._execution_query(workspace, statement) + if not summary_only: + return list(result.scalars().all()) + records = [] + for row in result.mappings(): + fields = dict(row) + metadata = { + key.removeprefix('display_metadata_'): fields.pop(key) + for key in row + if key.startswith('display_metadata_') + } + if metadata: + fields['metadata_json'] = json.dumps(metadata, ensure_ascii=False) + records.append(SimpleNamespace(**fields)) + return records + + def _execution_statement(self, model, *, summary_only=False): + """Keep periodic snapshots independent of message and attachment sizes.""" + if not summary_only: + return sa.select(model) + columns = [ + column + for column in model.__table__.columns + if column.name not in {'message_content', 'variables', 'input_json', 'authorization_json', 'metadata_json'} + ] + if model is models.MonitoringMessage: + return sa.select(*columns) + keys = ( + ('processor_id', 'processor_type', 'event_type', 'agent_name', 'target_name', 'bot_name') + if model is AgentRun + else ('platform', 'bot_name') + ) + # Metadata also contains complete inputs. Project only the small display + # fields in SQL, rather than deserializing an attachment-bearing object. + for key in keys: + columns.append(self._execution_json_text(model.metadata_json, key).label(f'display_metadata_{key}')) + return sa.select(*columns) async def _execution_page(self, workspace, statement, convert, offset, limit): rows = await self._execution_rows(workspace, statement.offset(offset).limit(limit + 1)) @@ -365,6 +408,35 @@ class ExecutionDetailsMixin: 'next_offset': offset + min(len(rows), limit), } + async def _conversation_message_page(self, workspace, statement, offset, limit): + """Select a page from the covering index before loading its payloads.""" + message = models.MonitoringMessage + identities = list( + ( + await self._execution_query( + workspace, statement.with_only_columns(message.id).offset(offset).limit(limit + 1) + ) + ).scalars() + ) + selected = identities[:limit] + records = {} + if selected: + # Equality joins keep SQLite on the primary key even for multiple IDs. + page = sa.union_all(*[sa.select(sa.literal(identity).label('id')) for identity in selected]).subquery() + rows = await self._execution_rows( + workspace, + sa.select(message) + .select_from(page) + .join(message, message.id == page.c.id) + .where(message.workspace_uuid == workspace), + ) + records = {row.id: row for row in rows} + return { + 'items': [message_item(records[identity]) for identity in selected if identity in records], + 'has_more': len(identities) > limit, + 'next_offset': offset + len(selected), + } + def _unhandled_event_conditions(self, workspace, bot_ids=None, start_time=None, end_time=None): # Absorbed steering events belong to an existing run, too. conditions = [ @@ -445,14 +517,21 @@ class ExecutionDetailsMixin: or None, } - async def enrich_execution_rows(self, workspace, items): + async def enrich_execution_rows(self, workspace, items, *, summary_only=False): pipeline_items = {item['id']: item for item in items if item.get('source') == 'pipeline'} if pipeline_items: # Pipeline rows own the UI identity; their linked Runner ledger owns # timing and usage. Read them in one batch for both list and detail. linked = await self._execution_query( workspace, - sa.select(models.MonitoringMessage.id, AgentRun) + sa.select( + models.MonitoringMessage.id, + AgentRun.started_at, + AgentRun.finished_at, + AgentRun.usage_json, + AgentRun.cost_json, + ) + .select_from(models.MonitoringMessage) .join( AgentRun, sa.and_( @@ -465,10 +544,15 @@ class ExecutionDetailsMixin: models.MonitoringMessage.id.in_(pipeline_items), ), ) - for message_id, run in linked.all(): - metrics = self._serialize_agent_execution(run) - pipeline_items[message_id].update( - {key: metrics[key] for key in ('started_at_ms', 'finished_at_ms', 'duration_ms', 'usage', 'cost')} + for run in linked.all(): + pipeline_items[run.id].update( + started_at_ms=epoch(run.started_at), + finished_at_ms=epoch(run.finished_at), + duration_ms=int(round((run.finished_at - run.started_at).total_seconds() * 1000)) + if run.started_at and run.finished_at + else None, + usage=self._deserialize_json(run.usage_json) or None, + cost=self._deserialize_json(run.cost_json) or None, ) calls = models.MonitoringLLMCall @@ -498,10 +582,11 @@ class ExecutionDetailsMixin: return events = await self._execution_rows( workspace, - sa.select(EventLog).where( + self._execution_statement(EventLog, summary_only=summary_only).where( EventLog.workspace_id == workspace, EventLog.event_id.in_(ids), ), + summary_only=summary_only, ) by_id = {row.event_id: row for row in events} for item in items: @@ -510,7 +595,9 @@ class ExecutionDetailsMixin: meta = parsed(event.metadata_json, {}) item.update( event_type=event.event_type, - input_preview=event_preview(parsed(event.input_json), event.input_summary, event.event_type), + input_preview=event_preview( + None if summary_only else parsed(event.input_json), event.input_summary, event.event_type + ), user_id=event.actor_id, user_name=event.actor_name, platform=meta.get('platform') or event.source, @@ -532,8 +619,13 @@ class ExecutionDetailsMixin: 'tool_calls', 'errors', } - if section is not None and section not in sections: + if section is not None and section not in sections | {'overview'}: raise ValueError('Unknown execution section') + requested = ( + sections + if section is None + else ({'inputs', 'outputs', 'deliveries', 'errors'} if section == 'overview' else {section}) + ) run = message = event = None if source in {'agent', 'auto'}: @@ -700,8 +792,12 @@ class ExecutionDetailsMixin: pages = {} for name, (query, convert) in statements.items(): - if section is None or section == name: - pages[name] = await self._execution_page(workspace, query, convert, offset, limit) + if name in requested: + pages[name] = ( + await self._conversation_message_page(workspace, query, offset, limit) + if name == 'conversation' and message + else await self._execution_page(workspace, query, convert, offset, limit) + ) if run: meta = parsed(run.metadata_json, {}) for item in pages.get('inputs', {}).get('items', []): @@ -709,17 +805,12 @@ class ExecutionDetailsMixin: # Do not apply the root snapshot to later steering events. if item['id'] == run.event_id and not item.get('metadata', {}).get('input_event'): item['content'] = event_content(item['content'], meta.get('input_event')) - if message and (section is None or section == 'inputs'): + if message and 'inputs' in requested: # The original rich message is authoritative even when the event # log is absent or only retained a compact event summary. if offset == 0: pages['inputs']['items'].insert(0, message_item(message)) - if ( - run - and not pages.get('inputs', {}).get('items') - and offset == 0 - and (section is None or section == 'inputs') - ): + if run and not pages.get('inputs', {}).get('items') and offset == 0 and 'inputs' in requested: meta = parsed(run.metadata_json, {}) content = event_content(meta.get('input'), meta.get('input_event')) if content: @@ -727,7 +818,7 @@ class ExecutionDetailsMixin: {'id': run.event_id or run.run_id, 'content': content, 'timestamp_ms': epoch(run.created_at)} ] - if section is None or section == 'related': + if 'related' in requested: # Union of ledger executions and legacy pipeline roots. SQL paging # avoids silently dropping siblings on highly fanned-out events. related_runs = sa.select( diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index d7ad628d6..527bd00d8 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -313,7 +313,7 @@ class LangBotMCPServer: ) @mcp.tool( - description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Each section is independently paginated; follow pages[section].next_offset when has_more is true.' + description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Use section=overview for inputs, outputs, deliveries and errors without loading history; request other sections on demand. Omit section for the full trace. Each section is independently paginated; follow pages[section].next_offset when has_more is true.' ) async def get_monitoring_execution_detail( source: typing.Literal['auto', 'agent', 'pipeline', 'event'], diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 97dce0da4..4b4d499ab 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -545,7 +545,7 @@ class Application: if total_deleted > 0: self.logger.info( f'Storage maintenance for Workspace {context.workspace_uuid}: ' - f'deleted expired files: {deleted}' + f'cleaned expired storage data: {deleted}' ) except asyncio.CancelledError: raise diff --git a/src/langbot/pkg/entity/persistence/agent_run.py b/src/langbot/pkg/entity/persistence/agent_run.py index 7d5dc0a8c..05d5c2b00 100644 --- a/src/langbot/pkg/entity/persistence/agent_run.py +++ b/src/langbot/pkg/entity/persistence/agent_run.py @@ -6,6 +6,7 @@ import datetime import sqlalchemy +from ...persistence.media import MediaReferenceText from .base import Base @@ -107,15 +108,17 @@ class AgentRun(Base): authorization_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) """Run-scoped authorization snapshot JSON.""" - metadata_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + metadata_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Additional metadata JSON.""" __table_args__ = ( + 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' ), sqlalchemy.Index('ix_agent_run_runner_status', 'runner_id', 'status'), sqlalchemy.Index('ix_agent_run_queue_claim', 'queue_name', 'status', 'priority', 'id'), + sqlalchemy.Index('ix_agent_run_workspace_status_lease', 'workspace_id', 'status', 'execution_lease_expires_at'), ) @@ -183,7 +186,7 @@ class AgentRunEvent(Base): type = sqlalchemy.Column(sqlalchemy.String(100), nullable=False, index=True) """Result event type.""" - data_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + data_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Result event payload JSON.""" usage_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) @@ -195,10 +198,11 @@ class AgentRunEvent(Base): source = sqlalchemy.Column(sqlalchemy.String(50), nullable=True) """Source that appended the event.""" - metadata_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + metadata_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Additional metadata JSON.""" __table_args__ = ( sqlalchemy.UniqueConstraint('run_id', 'sequence', name='uq_agent_run_event_run_sequence'), sqlalchemy.Index('ix_agent_run_event_run_sequence', 'run_id', 'sequence'), + sqlalchemy.Index('ix_agent_run_event_media_ttl', 'created_at', 'id', 'run_id'), ) diff --git a/src/langbot/pkg/entity/persistence/event_log.py b/src/langbot/pkg/entity/persistence/event_log.py index bd730b03c..8132d7ce9 100644 --- a/src/langbot/pkg/entity/persistence/event_log.py +++ b/src/langbot/pkg/entity/persistence/event_log.py @@ -6,6 +6,7 @@ import sqlalchemy import datetime from .base import Base +from ...persistence.media import MediaReferenceText class EventLog(Base): @@ -66,7 +67,7 @@ class EventLog(Base): input_summary = sqlalchemy.Column(sqlalchemy.Text, nullable=True) """Brief summary of input (truncated text, max 1000 chars).""" - input_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + input_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Full input JSON if reasonably sized (AgentInput as JSON string).""" # Raw event reference @@ -82,5 +83,7 @@ class EventLog(Base): created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, default=datetime.datetime.utcnow) """When this record was created.""" - metadata_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + metadata_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Additional metadata as JSON string.""" + + __table_args__ = (sqlalchemy.Index('ix_event_log_media_ttl', 'workspace_id', 'created_at', 'id'),) diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index ecf032986..358c1791e 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -1,5 +1,6 @@ import sqlalchemy +from ...persistence.media import MediaReferenceText from .base import Base @@ -19,7 +20,7 @@ class MonitoringMessage(Base): bot_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) pipeline_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True) pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) - message_content = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + message_content = sqlalchemy.Column(MediaReferenceText, nullable=False) session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True) execution_owner_id = sqlalchemy.Column(sqlalchemy.String(36), nullable=True, index=True) execution_lease_expires_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True, index=True) @@ -31,7 +32,7 @@ class MonitoringMessage(Base): user_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True) user_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True) # User display name runner_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True) # Runner name for this query - variables = sqlalchemy.Column(sqlalchemy.Text, nullable=True) # Query variables as JSON string + variables = sqlalchemy.Column(MediaReferenceText, nullable=True) # Query variables as JSON string role = sqlalchemy.Column(sqlalchemy.String(50), nullable=True, default='user') # user, assistant event_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) run_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) @@ -39,7 +40,17 @@ class MonitoringMessage(Base): __table_args__ = ( 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'), + sqlalchemy.Index( + 'ix_monitoring_messages_conversation_page', 'workspace_uuid', 'bot_id', 'session_id', 'timestamp', 'id' + ), + sqlalchemy.Index( + 'ix_monitoring_messages_workspace_status_lease', + 'workspace_uuid', + 'status', + 'execution_lease_expires_at', + ), ) diff --git a/src/langbot/pkg/entity/persistence/transcript.py b/src/langbot/pkg/entity/persistence/transcript.py index f975ab245..d61be2e3b 100644 --- a/src/langbot/pkg/entity/persistence/transcript.py +++ b/src/langbot/pkg/entity/persistence/transcript.py @@ -6,6 +6,7 @@ import sqlalchemy import datetime from .base import Base +from ...persistence.media import MediaReferenceText class Transcript(Base): @@ -48,11 +49,11 @@ class Transcript(Base): content = sqlalchemy.Column(sqlalchemy.Text, nullable=True) """Text content summary (may be truncated for large messages, max 4000 chars).""" - content_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + content_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Full structured content as JSON string (Message model dump).""" # Attachment references - attachment_refs_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + attachment_refs_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Attachment references as JSON string.""" # Sequence for cursor-based pagination @@ -69,11 +70,12 @@ class Transcript(Base): created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, default=datetime.datetime.utcnow) """When this item was created.""" - metadata_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + metadata_json = sqlalchemy.Column(MediaReferenceText, nullable=True) """Additional metadata as JSON string (sender_id, platform, etc.).""" # Indexes __table_args__ = ( + sqlalchemy.Index('ix_transcript_media_ttl', 'workspace_id', 'created_at', 'id'), sqlalchemy.Index('ix_transcript_conversation_seq', 'conversation_id', 'seq'), sqlalchemy.Index('ix_transcript_conversation_created', 'conversation_id', 'created_at'), sqlalchemy.Index('ix_transcript_scope_seq', 'bot_id', 'workspace_id', 'conversation_id', 'thread_id', 'seq'), diff --git a/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py new file mode 100644 index 000000000..eba466381 --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py @@ -0,0 +1,47 @@ +"""Index execution leases, conversation paging, and historical inline image expiry.""" + +from alembic import op +import sqlalchemy as sa + +revision = '0039_execution_lease_indexes' +down_revision = '0038_llm_call_details' +branch_labels = None +depends_on = None + +INDEXES = ( + ('agent_run', 'ix_agent_run_workspace_status_lease', ['workspace_id', 'status', 'execution_lease_expires_at']), + ( + 'monitoring_messages', + 'ix_monitoring_messages_workspace_status_lease', + ['workspace_uuid', 'status', 'execution_lease_expires_at'], + ), + ( + 'monitoring_messages', + 'ix_monitoring_messages_conversation_page', + ['workspace_uuid', 'bot_id', 'session_id', 'timestamp', 'id'], + ), + ('monitoring_messages', 'ix_monitoring_messages_media_ttl', ['workspace_uuid', 'timestamp', 'id']), + ('event_log', 'ix_event_log_media_ttl', ['workspace_id', 'created_at', 'id']), + ('agent_run', 'ix_agent_run_media_ttl', ['workspace_id', 'created_at', 'id']), + ('agent_run_event', 'ix_agent_run_event_media_ttl', ['created_at', 'id', 'run_id']), + ('transcript', 'ix_transcript_media_ttl', ['workspace_id', 'created_at', 'id']), +) + + +def upgrade(): + inspector = sa.inspect(op.get_bind()) + for table, index, columns in INDEXES: + if not inspector.has_table(table): + continue + # Fresh installations create these indexes from ORM metadata before migrations. + if index not in {item['name'] for item in inspector.get_indexes(table)}: + op.create_index(index, table, columns) + + +def downgrade(): + inspector = sa.inspect(op.get_bind()) + for table, index, _ in reversed(INDEXES): + if not inspector.has_table(table): + continue + if index in {item['name'] for item in inspector.get_indexes(table)}: + op.drop_index(index, table_name=table) diff --git a/src/langbot/pkg/persistence/media.py b/src/langbot/pkg/persistence/media.py new file mode 100644 index 000000000..337c63834 --- /dev/null +++ b/src/langbot/pkg/persistence/media.py @@ -0,0 +1,46 @@ +"""The storage contract for conversation payloads containing image references.""" + +from __future__ import annotations + +import json +from typing import Any + +from sqlalchemy import Text +from sqlalchemy.engine import Dialect +from sqlalchemy.sql.operators import OperatorType +from sqlalchemy.types import TypeDecorator + + +class MediaReferenceText(TypeDecorator[str]): + """Keep the existing TEXT format while rejecting structured inline images. + + Writers must externalize retained images through MediaCache or redact + metadata-only images before binding a value. Both ORM and Core writes use + this contract; reads preserve historical values for incremental cleanup. + Plain text remains valid for fields such as monitoring message content. + """ + + impl = Text + cache_ok = True + + def process_bind_param(self, value: str | None, dialect: Dialect) -> str | None: + if value is None: + return None + try: + payload = json.loads(value) + except json.JSONDecodeError: + return value + + # Model imports must not initialize the media cache's runtime dependencies. + from ..storage.media import MediaCache + + if MediaCache.has_inline_images(payload): + raise ValueError( + 'Inline images cannot be persisted; use MediaCache.externalize_chain_dump() ' + 'for retained content or MediaCache.redact_inline_images() for metadata' + ) + return value + + def coerce_compared_value(self, op: OperatorType | None, value: Any) -> Text: + # Legacy payloads must remain usable in searches and compare-and-swap updates. + return self.impl diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index 396b20103..c55b7a252 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/src/langbot/pkg/persistence/tenant_uow.py @@ -19,6 +19,8 @@ from sqlalchemy.dialects.postgresql.dml import OnConflictDoUpdate as PostgreSQLO from sqlalchemy.dialects.sqlite.dml import OnConflictDoNothing as SQLiteOnConflictDoNothing from sqlalchemy.dialects.sqlite.dml import OnConflictDoUpdate as SQLiteOnConflictDoUpdate +from .media import MediaReferenceText + TENANT_SETTING = 'langbot.workspace_uuid' ACCOUNT_SETTING = 'langbot.account_uuid' @@ -292,7 +294,7 @@ def _validate_scoped_sql_type( if type(sql_type) in {Vector, HALFVEC}: return - if not type(sql_type).__module__.startswith('sqlalchemy.'): + if type(sql_type) is not MediaReferenceText and not type(sql_type).__module__.startswith('sqlalchemy.'): raise ScopedSessionTransactionError('TenantUnitOfWork does not allow custom SQL types in public statements') for identifier_attribute in ('name', 'schema', 'collation'): @@ -486,7 +488,7 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s monitoring_metadata = ( type(element.type) is sqlalchemy.JSON and isinstance(source, sqlalchemy.Column) - and type(source.type) is sqlalchemy.Text + and type(source.type) in {sqlalchemy.Text, MediaReferenceText} and getattr(getattr(source, 'table', None), 'name', None) in {'event_log', 'agent_run'} and ( source.name == 'metadata_json' or (source.table.name == 'event_log' and source.name == 'input_json') diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index 4efb21498..25b58ec8b 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -143,12 +143,9 @@ class MonitoringHelper: sender_name = query.message_event.sender.member_name # Try to record message - # Use JSON serialization to preserve message chain structure (including image URLs, etc.) + # Let persistence externalize images before serializing the message chain. if hasattr(query, 'message_chain') and hasattr(query.message_chain, 'model_dump'): - chain_dump = query.message_chain.model_dump() - if hasattr(ap, 'storage_mgr') and hasattr(ap.storage_mgr, 'media_cache'): - chain_dump = await ap.storage_mgr.media_cache.externalize_chain_dump(chain_dump) - message_content = json.dumps(chain_dump, ensure_ascii=False) + message_content = query.message_chain.model_dump() else: message_content = str(query) @@ -257,10 +254,10 @@ class MonitoringHelper: # Extract response content from resp_message_chain if hasattr(query, 'resp_message_chain') and query.resp_message_chain: - # Serialize the last response message chain + # Preserve the structure until persistence has externalized images. last_resp = query.resp_message_chain[-1] if hasattr(last_resp, 'model_dump'): - message_content = json.dumps(last_resp.model_dump(), ensure_ascii=False) + message_content = last_resp.model_dump() else: message_content = str(last_resp) elif hasattr(query, 'resp_messages') and query.resp_messages: @@ -268,10 +265,7 @@ class MonitoringHelper: if hasattr(last_resp, 'get_content_platform_message_chain'): chain = last_resp.get_content_platform_message_chain() if hasattr(chain, 'model_dump'): - chain_dump = chain.model_dump() - if hasattr(ap, 'storage_mgr') and hasattr(ap.storage_mgr, 'media_cache'): - chain_dump = await ap.storage_mgr.media_cache.externalize_chain_dump(chain_dump) - message_content = json.dumps(chain_dump, ensure_ascii=False) + message_content = chain.model_dump() else: message_content = str(chain) else: diff --git a/src/langbot/pkg/pipeline/resprule/resprule.py b/src/langbot/pkg/pipeline/resprule/resprule.py index 6e7fc369e..3e8c16a96 100644 --- a/src/langbot/pkg/pipeline/resprule/resprule.py +++ b/src/langbot/pkg/pipeline/resprule/resprule.py @@ -37,8 +37,13 @@ class GroupRespondRuleCheckStage(stage.PipelineStage): if query.launcher_type.value != 'group': # 只处理群消息 return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) - # 通过路由规则明确指定的流水线，跳过群响应规则检查 - if query.variables and query.variables.get('_routed_by_rule', False): + # Interaction submissions and routed platform form callbacks resume the + # original run without another mention or prefix. Ordinary routing only + # selects a pipeline and must still obey its group response rules. + if query.variables and ( + query.variables.get('_interaction_submission') + or (query.variables.get('_routed_by_rule') and query.variables.get('_dify_form_action')) + ): return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query) rules = query.pipeline_config['trigger']['group-respond-rules'] diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index ae0ecdc4e..ac425a9c4 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -1356,10 +1356,7 @@ class RuntimeBot: """Record a discarded message in the monitoring system.""" try: if hasattr(message_chain, 'model_dump'): - chain_dump = message_chain.model_dump() - if hasattr(self.ap, 'storage_mgr') and hasattr(self.ap.storage_mgr, 'media_cache'): - chain_dump = await self.ap.storage_mgr.media_cache.externalize_chain_dump(chain_dump) - message_content = json.dumps(chain_dump, ensure_ascii=False) + message_content = message_chain.model_dump() else: message_content = str(message_chain) diff --git a/src/langbot/pkg/plugin/runner_actions.py b/src/langbot/pkg/plugin/runner_actions.py index bca1a6c07..659ac2f4c 100644 --- a/src/langbot/pkg/plugin/runner_actions.py +++ b/src/langbot/pkg/plugin/runner_actions.py @@ -469,12 +469,16 @@ def register(h): if payload_error: return handler.ActionResponse.error(message=payload_error) - event = await store.append_event( + from ..agent.runner.run_journal import AgentRunJournal + + event = await AgentRunJournal(h.ap).append_run_result( run_id=str(target_run_id), sequence=sequence, - event_type=str(event_type), - data=event_payload, - usage=usage if isinstance(usage, dict) else None, + result_dict={ + 'type': str(event_type), + 'data': event_payload, + 'usage': usage if isinstance(usage, dict) else None, + }, source=str(data.get('source') or result.get('source') or 'runner'), metadata=metadata, ) diff --git a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py index 300fb7551..6f1de0e86 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py +++ b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py @@ -836,6 +836,13 @@ class LiteLLMRequester(requester.ProviderAPIRequester): req_messages = [] for m in messages: msg_dict = m.dict(exclude_none=True) + provider_fields = msg_dict.get('provider_specific_fields') + if isinstance(provider_fields, dict): + # Stream completion metadata belongs to the response, never to + # replayed chat messages. Keep provider signatures for LiteLLM. + provider_fields.pop('finish_reason', None) + if not provider_fields: + msg_dict.pop('provider_specific_fields', None) content = msg_dict.get('content') if msg_dict.get('role') == 'assistant' and reasoning_family: @@ -908,6 +915,17 @@ class LiteLLMRequester(requester.ProviderAPIRequester): return req_messages + @staticmethod + def _adapt_messages_for_provider(messages: list[dict], chat_provider: str | None) -> None: + """Adapt serialized request messages in place for the chat transport.""" + if chat_provider == 'groq': + # Groq rejects LiteLLM's metadata envelope on messages and tool calls. + for message in messages: + message.pop('provider_specific_fields', None) + for tool_call in message.get('tool_calls') or []: + tool_call.pop('provider_specific_fields', None) + tool_call.get('function', {}).pop('provider_specific_fields', None) + _THINK_PATTERNS: tuple[str, ...] = ( r'^\s*(?:(?!).)*?\s*', r'^\s*(?:(?!CRETIRE_REASONING_BEGINk).)*?CRETIRE_REASONING_ENDk\s*', @@ -1324,6 +1342,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester): # OpenRouter shares requester metadata with OpenAI-compatible non-chat # operations, but chat needs its own adapter to preserve cache markers. chat_provider = 'openrouter' if self._uses_openrouter_protocol() else self._get_custom_llm_provider() + self._adapt_messages_for_provider(req_messages, chat_provider) model_name = self._build_litellm_model_name(model.model_entity.name, chat_provider) api_key = model.provider.token_mgr.get_token() diff --git a/src/langbot/pkg/storage/inline_media_cleanup.py b/src/langbot/pkg/storage/inline_media_cleanup.py new file mode 100644 index 000000000..3293fcca4 --- /dev/null +++ b/src/langbot/pkg/storage/inline_media_cleanup.py @@ -0,0 +1,147 @@ +"""Incrementally expire historical inline images without rewriting record history.""" + +from __future__ import annotations + +import asyncio +import datetime +import json +from typing import TYPE_CHECKING + +import sqlalchemy as sa + +from ..api.http.authz import WorkspaceRequiredError +from ..api.http.context import ExecutionContext +from ..entity.persistence.agent_run import AgentRun, AgentRunEvent +from ..entity.persistence.event_log import EventLog +from ..entity.persistence.metadata import WorkspaceMetadata +from ..entity.persistence.monitoring import MonitoringMessage +from ..entity.persistence.transcript import Transcript +from ..persistence.datetime_utils import as_naive_utc +from .media import MediaCache + +if TYPE_CHECKING: + from ..core.app import Application + + +ROWS_PER_TABLE = 100 +CHARACTERS_PER_TABLE = 8 * 1024 * 1024 +CURSOR_PREFIX = 'inline_media_ttl_v1:' + + +def _redact_columns(values: dict) -> dict: + updates = {} + for name, raw in values.items(): + if not raw or ('base64' not in raw and 'data:image/' not in raw): + continue + try: + payload = json.loads(raw) + except (ValueError, TypeError, RecursionError): + # Preserve malformed historical text rather than deleting an entire record. + continue + if MediaCache.has_inline_images(payload): + updates[name] = json.dumps(MediaCache.redact_inline_images(payload), ensure_ascii=False) + return updates + + +async def cleanup_inline_media(ap: Application, context: ExecutionContext, retention_days: int) -> int: + """Use the media TTL for immutable history, with durable per-Workspace cursors. + + Bound each table by both candidate count and serialized payload characters. One + oversized row is processed on its own so it cannot permanently block the cursor. + """ + if not isinstance(context, ExecutionContext) or not context.workspace_uuid: + raise WorkspaceRequiredError('Inline media cleanup requires an ExecutionContext') + if retention_days < 1: + raise ValueError('retention_days must be >= 1') + workspace = context.workspace_uuid + cutoff = as_naive_utc(datetime.datetime.now(datetime.timezone.utc)) - datetime.timedelta(days=retention_days) + sources = ( + ( + MonitoringMessage, + MonitoringMessage.timestamp, + MonitoringMessage.workspace_uuid == workspace, + ('message_content', 'variables'), + ), + (EventLog, EventLog.created_at, EventLog.workspace_id == workspace, ('input_json', 'metadata_json')), + (AgentRun, AgentRun.created_at, AgentRun.workspace_id == workspace, ('metadata_json',)), + ( + AgentRunEvent, + AgentRunEvent.created_at, + sa.exists( + sa.select(AgentRun.id).where( + AgentRun.run_id == AgentRunEvent.run_id, AgentRun.workspace_id == workspace + ) + ), + ('data_json', 'metadata_json'), + ), + ( + Transcript, + Transcript.created_at, + Transcript.workspace_id == workspace, + ('content_json', 'attachment_refs_json', 'metadata_json'), + ), + ) + manager = ap.persistence_mgr + cleaned = 0 + for model, timestamp, scope, fields in sources: + consumed = 0 + cursor_key = CURSOR_PREFIX + model.__tablename__ + cursor_filter = (WorkspaceMetadata.workspace_uuid == workspace, WorkspaceMetadata.key == cursor_key) + for _ in range(ROWS_PER_TABLE): + async with manager.tenant_uow(workspace): + cursor_result = await manager.execute_async( + sa.select(WorkspaceMetadata.value).where(*cursor_filter).with_for_update() + ) + cursor_raw = cursor_result.scalar_one_or_none() + # Events have no Workspace column. Limit indexed candidates before + # checking ownership so other Workspaces cannot cause unbounded scans. + query = sa.select(model.id, timestamp).where(timestamp < cutoff) + if model is not AgentRunEvent: + query = query.where(scope) + if cursor_raw is not None: + after_time, after_id = json.loads(cursor_raw) + query = query.where( + sa.tuple_(timestamp, model.id) + > sa.tuple_(datetime.datetime.fromisoformat(after_time), after_id) + ) + result = await manager.execute_async(query.order_by(timestamp, model.id).limit(1)) + row = result.first() + if row is None: + break + payload = ( + await manager.execute_async( + sa.select(*(getattr(model, field) for field in fields)).where(model.id == row[0], scope) + ) + ).first() + # Missing and foreign candidates still consume budget and advance the + # cursor. Their payloads are never returned to the cleanup worker. + values = dict(zip(fields, payload)) if payload is not None else {} + consumed += sum(len(value) for value in values.values() if value) + updates = await asyncio.to_thread(_redact_columns, values) if values else {} + if updates: + if model is AgentRun: + # Media expiry must not appear as new execution activity. + updates['updated_at'] = AgentRun.updated_at + changed = await manager.execute_async( + sa.update(model) + .where( + scope, + model.id == row[0], + *(getattr(model, name) == values[name] for name in fields if name in updates), + ) + .values(**updates) + ) + if not changed.rowcount: + # A concurrent writer changed this payload; retry it next time. + break + cleaned += 1 + cursor = json.dumps([row[1].isoformat(), row[0]]) + if cursor_raw is None: + await manager.execute_async( + sa.insert(WorkspaceMetadata).values(workspace_uuid=workspace, key=cursor_key, value=cursor) + ) + else: + await manager.execute_async(sa.update(WorkspaceMetadata).where(*cursor_filter).values(value=cursor)) + if consumed >= CHARACTERS_PER_TABLE: + break + return cleaned diff --git a/src/langbot/pkg/storage/media.py b/src/langbot/pkg/storage/media.py index 8f5537fb3..4a615abc1 100644 --- a/src/langbot/pkg/storage/media.py +++ b/src/langbot/pkg/storage/media.py @@ -11,6 +11,8 @@ import xxhash from pathlib import Path from typing import TYPE_CHECKING, Any +from ..utils.image import DEFAULT_BASE64_MEDIA_LIMIT + if TYPE_CHECKING: from ...core import app from . import mgr as storage_mgr @@ -49,14 +51,80 @@ class MediaCache: if split_index != -1: mime_type = data_url[5:split_index] b64_data = data_url[split_index + 8 :] - return base64.b64decode(b64_data), mime_type - # Try raw base64 if sufficiently long - if len(data_url) > 20 and not data_url.startswith(('http://', 'https://', '/')): - return base64.b64decode(data_url), 'application/octet-stream' + return MediaCache._decode_base64(b64_data), mime_type + if not data_url.startswith(('http://', 'https://', '/')): + return MediaCache._decode_base64(data_url), 'application/octet-stream' except Exception: return None return None + @staticmethod + def _decode_base64(value: str) -> bytes: + """Bound decoding before allocating a binary payload for persistence.""" + if len(value) > 4 * ((DEFAULT_BASE64_MEDIA_LIMIT + 2) // 3): + raise ValueError('Inline image exceeds the media size limit') + decoded = base64.b64decode(value, validate=True) + if len(decoded) > DEFAULT_BASE64_MEDIA_LIMIT: + raise ValueError('Inline image exceeds the media size limit') + return decoded + + @staticmethod + def _inline_image_payload(node: dict) -> str | None: + """Recognize platform and provider image payloads without fetching URLs.""" + node_type = node.get('type') + if node_type == 'Image': + payload = node.get('base64') + url = node.get('url') + elif node_type in ('image', 'image_url', 'image_base64'): + payload = node.get('image_base64') + url = node.get('image_url') + if isinstance(url, dict): + url = url.get('url') + else: + return None + if isinstance(payload, str) and payload: + return payload + if isinstance(url, str) and url.startswith('data:image/'): + return url + return None + + @classmethod + def has_inline_images(cls, value: Any) -> bool: + if isinstance(value, list): + return any(cls.has_inline_images(item) for item in value) + if isinstance(value, dict): + return cls._inline_image_payload(value) is not None or any( + cls.has_inline_images(item) for item in value.values() if isinstance(item, (list, dict)) + ) + return False + + @staticmethod + def _clear_inline_image(node: dict) -> None: + field = 'base64' if node.get('type') == 'Image' else 'image_base64' + if node.get(field): + node[field] = None + url_field = 'url' if node.get('type') == 'Image' else 'image_url' + url = node.get(url_field) + if isinstance(url, str) and url.startswith('data:image/'): + node[url_field] = None + elif isinstance(url, dict) and str(url.get('url', '')).startswith('data:image/'): + node[url_field] = {**url, 'url': None} + + @classmethod + def redact_inline_images(cls, value: Any) -> Any: + """Keep metadata-only records binary-free, including short raw base64.""" + if isinstance(value, list): + return [cls.redact_inline_images(item) for item in value] + if isinstance(value, dict): + node = dict(value) + if cls._inline_image_payload(node) is not None: + cls._clear_inline_image(node) + return { + key: cls.redact_inline_images(item) if isinstance(item, (list, dict)) else item + for key, item in node.items() + } + return value + @staticmethod def guess_extension(mime_type: str | None, default: str = '.jpg') -> str: """Guess appropriate file extension from MIME type.""" @@ -201,29 +269,33 @@ class MediaCache: return [await self.externalize_chain_dump(item) for item in chain_dump] if isinstance(chain_dump, dict): node = copy.copy(chain_dump) - node_type = node.get('type') - if node_type == 'Image': - b64 = node.get('base64') - if b64 and isinstance(b64, str): - try: - parsed = self.parse_data_url(b64) - if parsed is not None: - raw_bytes, mime_type = parsed - hash_str, storage_key, size = await self.save_media(raw_bytes, mime_type) - filename = os.path.basename(storage_key) - current_url = node.get('url') or '' - if current_url and not current_url.startswith('data:'): - node['original_url'] = current_url - node['url'] = f'/api/v1/files/media/{filename}' - node['base64'] = None - node['hash'] = hash_str - node['storage_key'] = storage_key - node['size'] = size - node['mime_type'] = mime_type - except Exception as e: - if hasattr(self.ap, 'logger') and self.ap.logger: - self.ap.logger.warning(f'Failed to externalize image to media cache: {e}') - node['base64'] = None + payload = self._inline_image_payload(node) + if payload is not None: + try: + parsed = await asyncio.to_thread(self.parse_data_url, payload) + if parsed is None: + raise ValueError('Invalid inline image payload') + raw_bytes, mime_type = parsed + hash_str, storage_key, size = await self.save_media(raw_bytes, mime_type) + url = f'/api/v1/files/media/{os.path.basename(storage_key)}' + if node.get('type') == 'Image': + current_url = node.get('url') or '' + if current_url and not current_url.startswith('data:'): + node['original_url'] = current_url + node['url'] = url + else: + current_url = node.get('image_url') + if isinstance(current_url, dict): + current_url = current_url.get('url') + if current_url and not current_url.startswith('data:'): + node['original_url'] = current_url + image_url = node.get('image_url') + node['image_url'] = {**image_url, 'url': url} if isinstance(image_url, dict) else {'url': url} + node.update(hash=hash_str, storage_key=storage_key, size=size, mime_type=mime_type) + except Exception as e: + if hasattr(self.ap, 'logger') and self.ap.logger: + self.ap.logger.warning(f'Failed to externalize image to media cache: {e}') + self._clear_inline_image(node) for k, v in list(node.items()): if isinstance(v, (list, dict)): node[k] = await self.externalize_chain_dump(v) diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index de5092370..97650b8d8 100644 --- a/src/langbot/templates/config.yaml +++ b/src/langbot/templates/config.yaml @@ -224,7 +224,8 @@ storage: max_object_read_bytes: 10485760 # Media content cache (images & attachments externalized from monitoring and pipelines) media_cache: - # Retention period in days for cached media (defaults to 30 days) + # Retention for CAS files and historical inline images (defaults to 30 days). + # Inline images expire by record creation time; message text is retained. retention_days: 30 # Maximum disk storage for cached media in MB (0 means unlimited, defaults to 0) max_size_mb: 0 diff --git a/tests/integration/persistence/resource_migration_support.py b/tests/integration/persistence/resource_migration_support.py index c82742acf..47bac85fe 100644 --- a/tests/integration/persistence/resource_migration_support.py +++ b/tests/integration/persistence/resource_migration_support.py @@ -202,6 +202,7 @@ async def create_legacy_resource_schema(engine, *, instance_uuid: str) -> None: table.append_column(sa.Column(name, sa.String(255))) if table_name == 'monitoring_messages': table.append_column(sa.Column('bot_id', sa.String(255), nullable=False, default='bot-1')) + table.append_column(sa.Column('status', sa.String(50), nullable=False, default='success')) table.append_column(sa.Column('role', sa.String(50))) else: table.append_column(sa.Column('message_count', sa.Integer, nullable=False, default=1)) diff --git a/tests/integration/persistence/test_media_reference_postgres.py b/tests/integration/persistence/test_media_reference_postgres.py new file mode 100644 index 000000000..a9bea397c --- /dev/null +++ b/tests/integration/persistence/test_media_reference_postgres.py @@ -0,0 +1,190 @@ +"""Media writes against real asyncpg and existing TEXT columns. + +Run with TEST_POSTGRES_URL pointing to a disposable PostgreSQL database. +Each case creates and removes its own schema. +""" + +import importlib +import json +import os +import uuid +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + +from langbot.pkg.agent.runner.run_journal import AgentRunJournal +from langbot.pkg.agent.runner.run_ledger_store import RunLedgerStore +from langbot.pkg.entity.persistence.agent_run import AgentRun, AgentRunEvent +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.persistence.tenant_uow import _validate_scoped_statement_call +from langbot.pkg.storage.media import MediaCache + +pytestmark = [pytest.mark.integration, pytest.mark.slow, pytest.mark.asyncio] + + +def _published_schema(connection): + with Operations.context(MigrationContext.configure(connection)): + for revision in ( + '58846a8d7a81_add_event_log_and_transcript_tables', + '8d3a1f2c4b6e_add_agent_run_ledger', + '0036_execution_host_leases', + ): + importlib.import_module(f'langbot.pkg.persistence.alembic.versions.{revision}').upgrade() + + +@pytest.fixture(params=['metadata', 'published_migrations']) +async def pg_engine(request): + url = os.environ.get('TEST_POSTGRES_URL') + if not url: + pytest.skip('TEST_POSTGRES_URL is required (disposable PostgreSQL)') + namespace = f'media_contract_{uuid.uuid4().hex}' + admin = create_async_engine(url) + engine = None + try: + async with admin.begin() as conn: + await conn.execute(sa.text(f'CREATE SCHEMA {namespace}')) + engine = create_async_engine(url, connect_args={'server_settings': {'search_path': namespace}}) + async with engine.begin() as conn: + if request.param == 'metadata': + await conn.run_sync( + lambda sync: Base.metadata.create_all(sync, tables=[AgentRun.__table__, AgentRunEvent.__table__]) + ) + else: + await conn.run_sync(_published_schema) + yield engine + finally: + if engine is not None: + await engine.dispose() + async with admin.begin() as conn: + await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS {namespace} CASCADE')) + await admin.dispose() + + +@pytest.fixture +async def store(pg_engine): + store = RunLedgerStore(pg_engine) + await store.create_run( + run_id='run', event_id='event', binding_id=None, runner_id='runner', workspace_id='ws', metadata={'attempt': 0} + ) + return store + + +@pytest.mark.parametrize('writer', ['orm', 'insert', 'update']) +@pytest.mark.parametrize( + 'image', + [ + {'type': 'Image', 'base64': 'eA=='}, + {'type': 'image_base64', 'image_base64': 'eA=='}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA=='}}, + ], +) +async def test_unprocessed_images_roll_back(pg_engine, store, writer, image): + content = json.dumps({'nested': [image]}) + with pytest.raises(sa.exc.StatementError, match='Inline images cannot be persisted'): + if writer == 'orm': + async with AsyncSession(pg_engine) as session: + run = (await session.execute(sa.select(AgentRun).where(AgentRun.run_id == 'run'))).scalar_one() + run.status = 'completed' + run.metadata_json = content + await session.commit() + else: + statement = ( + sa.insert(AgentRun).values(run_id='new', runner_id='runner', status='completed', metadata_json=content) + if writer == 'insert' + else sa.update(AgentRun) + .where(AgentRun.run_id == 'run') + .values(status='completed', metadata_json=content) + ) + async with pg_engine.begin() as conn: + await conn.execute(statement) + run = await store.get_run('run') + assert run['status'] == 'running' + assert run['metadata'] == {'attempt': 0} + assert await store.get_run('new') is None + + +@pytest.mark.parametrize('operation', ['pin', 'finalize', 'status_only']) +async def test_legacy_metadata_updates_preserve_the_write_contract(pg_engine, store, operation): + legacy = {'image': {'type': 'Image', 'base64': 'eA=='}, 'attempt': 0} + original = json.dumps(legacy) + async with pg_engine.begin() as conn: + # Seed the pre-contract format while keeping current model mappings. + await conn.execute( + sa.update(AgentRun) + .where(AgentRun.run_id == 'run') + .values(metadata_json=sa.bindparam('legacy', original, type_=sa.Text())) + ) + assert ( + await conn.execute(sa.select(AgentRun.run_id).where(AgentRun.metadata_json == original)) + ).scalar_one() == 'run' + + with pytest.raises(sa.exc.StatementError, match='Inline images cannot be persisted'): + await store.finalize_run( + run_id='run', status='completed', metadata={'new': {'type': 'Image', 'base64': 'eQ=='}} + ) + assert (await store.get_run('run'))['status'] == 'running' + assert (await store.get_run('run'))['metadata'] == legacy + + if operation == 'pin': + assert await store.pin_skill_revision('run', 'ws', 'demo', 'v1') == 'v1' + assert (await store.get_run('run'))['metadata'] == { + 'image': {'type': 'Image', 'base64': None}, + 'attempt': 0, + 'activated_skills': {'demo': 'v1'}, + } + else: + metadata = {'note': 'done'} if operation == 'finalize' else None + run = await store.finalize_run(run_id='run', status='completed', metadata=metadata) + assert run['status'] == 'completed' + assert run['metadata'] == ( + {'image': {'type': 'Image', 'base64': None}, 'attempt': 0, 'note': 'done'} + if metadata is not None + else legacy + ) + + +async def test_journal_retains_references_and_redacts_metadata(pg_engine, store): + provider = SimpleNamespace(exists=AsyncMock(return_value=False), save=AsyncMock()) + storage = SimpleNamespace(storage_provider=provider) + app = SimpleNamespace(storage_mgr=storage, persistence_mgr=SimpleNamespace(get_db_engine=lambda: pg_engine)) + storage.media_cache = MediaCache(app, storage) + image = {'type': 'Image', 'base64': 'eA=='} + event = await AgentRunJournal(app).append_run_result( + run_id='run', + sequence=1, + result_dict={'type': 'tool.call.completed', 'data': {'image': image}}, + metadata={'preview': image}, + ) + assert event['data']['image']['base64'] is None + assert event['data']['image']['url'].startswith('/api/v1/files/media/') + assert event['metadata'] == {'preview': {'type': 'Image', 'base64': None}} + provider.save.assert_awaited_once() + async with pg_engine.connect() as conn: + row = (await conn.execute(sa.select(AgentRunEvent.data_json, AgentRunEvent.metadata_json))).one() + assert json.loads(row.data_json) == event['data'] + assert json.loads(row.metadata_json) == event['metadata'] + + +async def test_text_schema_and_monitoring_json_casts(pg_engine, store): + statement = sa.select(sa.cast(AgentRun.metadata_json, sa.JSON)['attempt'].as_integer()) + _validate_scoped_statement_call((statement,), {}) + async with pg_engine.connect() as conn: + assert (await conn.execute(statement)).scalar_one() == 0 + columns = ( + ( + await conn.execute( + sa.text( + 'SELECT data_type FROM information_schema.columns WHERE table_schema = current_schema() ' + "AND table_name IN ('agent_run', 'agent_run_event') AND column_name IN ('data_json', 'metadata_json')" + ) + ) + ) + .scalars() + .all() + ) + assert columns == ['text'] * 3 diff --git a/tests/unit_tests/agent/test_run_journal.py b/tests/unit_tests/agent/test_run_journal.py index 1ae134a01..c229318ca 100644 --- a/tests/unit_tests/agent/test_run_journal.py +++ b/tests/unit_tests/agent/test_run_journal.py @@ -6,6 +6,7 @@ from unittest.mock import AsyncMock import pytest from langbot.pkg.agent.runner.run_journal import AgentRunJournal +from langbot.pkg.storage.media import MediaCache @pytest.mark.asyncio @@ -68,3 +69,21 @@ def test_event_snapshot_preserves_custom_values_and_redacts_nested_binary(): assert saved['custom']['contents'][0]['image_base64'] is None assert saved['attachments'][0]['content'] is None assert data['custom']['contents'][0]['image_base64'] == 'secret' + + +@pytest.mark.parametrize( + 'image', + [ + {'type': 'Image', 'base64': 'eA=='}, + {'type': 'image_base64', 'image_base64': 'eA=='}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA=='}}, + ], +) +def test_metadata_and_transcript_redaction_cover_all_image_encodings(image): + content = AgentRunJournal._sanitize_content_item(image) + event = AgentRunJournal._sanitize_event_data({'nested': [image]}) + assert not MediaCache.has_inline_images(content) + assert not MediaCache.has_inline_images(event) + assert content['content_redacted'] is True + assert event['nested'][0]['content_redacted'] is True + assert MediaCache.has_inline_images(image) diff --git a/tests/unit_tests/agent/test_run_ledger_api_auth.py b/tests/unit_tests/agent/test_run_ledger_api_auth.py index e75f67739..62d390326 100644 --- a/tests/unit_tests/agent/test_run_ledger_api_auth.py +++ b/tests/unit_tests/agent/test_run_ledger_api_auth.py @@ -3,8 +3,9 @@ from __future__ import annotations import datetime +import json from types import SimpleNamespace -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest import sqlalchemy @@ -17,6 +18,7 @@ from langbot.pkg.agent.runner.session_registry import AgentRunSessionRegistry from langbot.pkg.entity.persistence import agent_run as agent_run_model from langbot.pkg.entity.persistence.base import Base from langbot.pkg.plugin.handler import RuntimeConnectionHandler +from langbot.pkg.storage.media import MediaCache from langbot_plugin.api.entities.builtin.runner.run_ledger import ( AgentRun, AgentRunEvent, @@ -1064,6 +1066,50 @@ async def test_run_append_result_basic_path(session_registry, db_engine): assert event.usage == {'output_tokens': 1} +@pytest.mark.asyncio +async def test_run_append_result_uses_cas_and_redacts_metadata(session_registry, db_engine): + await _register_session(session_registry, available_apis={'run_append_result': True}) + await _create_run(db_engine) + handler = _handler(db_engine) + objects = {} + + async def save(key, data): + objects[key] = data + + storage = SimpleNamespace( + storage_provider=SimpleNamespace(exists=AsyncMock(return_value=False), save=AsyncMock(side_effect=save)) + ) + storage.media_cache = MediaCache(handler.ap, storage) + handler.ap.storage_mgr = storage + image = {'type': 'Image', 'base64': 'eA=='} + result = await handler.actions[PluginToRuntimeAction.RUN_APPEND_RESULT.value]( + { + 'run_id': 'run_1', + 'caller_plugin_identity': 'test/runner', + 'sequence': 2, + 'event_type': 'tool.call.completed', + 'data': {'tool_call_id': 'call_1', 'tool_name': 'image', 'result': {'images': [image]}}, + 'metadata': {'preview': image}, + } + ) + assert result.code == 0 + event = AgentRunEvent.model_validate(result.data) + saved = event.data['result']['images'][0] + assert saved['base64'] is None + assert saved['url'].startswith('/api/v1/files/media/') + assert objects == {saved['storage_key']: b'x'} + async with db_engine.connect() as conn: + metadata = ( + await conn.execute( + sqlalchemy.select(agent_run_model.AgentRunEvent.metadata_json).where( + agent_run_model.AgentRunEvent.sequence == 2 + ) + ) + ).scalar_one() + assert json.loads(metadata)['preview']['base64'] is None + assert image['base64'] == 'eA==' + + @pytest.mark.asyncio async def test_run_append_result_rejects_side_effecting_result_type(session_registry, db_engine, ledger_snapshot): await _register_session(session_registry, available_apis={'run_append_result': True}) diff --git a/tests/unit_tests/agent/test_run_ledger_store.py b/tests/unit_tests/agent/test_run_ledger_store.py index 22c97baea..2195f86f7 100644 --- a/tests/unit_tests/agent/test_run_ledger_store.py +++ b/tests/unit_tests/agent/test_run_ledger_store.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime +import json import pytest import sqlalchemy @@ -36,6 +37,81 @@ def store(db_engine): return RunLedgerStore(db_engine) +@pytest.fixture +async def legacy_run(store, db_engine): + await store.create_run( + run_id='legacy', + event_id='event', + binding_id='binding', + runner_id='runner', + workspace_id='ws', + ) + metadata = { + 'input': [ + {'type': 'Image', 'base64': 'eA=='}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA==', 'detail': 'low'}}, + {'type': 'image_url', 'image_url': {'url': 'https://example.com/image.png'}}, + {'type': 'text', 'text': 'keep this'}, + ], + 'attempt': 0, + 'activated_skills': {'existing': 'v1'}, + } + async with db_engine.begin() as conn: + # Explicit TEXT binding simulates data persisted before the media contract. + await conn.execute( + sqlalchemy.update(AgentRun) + .where(AgentRun.run_id == 'legacy') + .values( + metadata_json=sqlalchemy.bindparam('legacy_metadata', json.dumps(metadata), type_=sqlalchemy.Text()) + ) + ) + return metadata + + +@pytest.mark.asyncio +@pytest.mark.parametrize('operation', ['pin', 'finalize']) +async def test_metadata_update_redacts_legacy_images(store, legacy_run, operation): + if operation == 'pin': + assert await store.pin_skill_revision('legacy', 'ws', 'demo', 'v2') == 'v2' + else: + await store.finalize_run(run_id='legacy', status='completed', metadata={'note': 'done'}) + + run = await store.get_run('legacy') + expected = { + **legacy_run, + 'input': [ + {'type': 'Image', 'base64': None}, + {'type': 'image_url', 'image_url': {'url': None, 'detail': 'low'}}, + *legacy_run['input'][2:], + ], + } + if operation == 'pin': + expected['activated_skills'] = {'existing': 'v1', 'demo': 'v2'} + assert run['status'] == 'running' + else: + expected['note'] = 'done' + assert run['status'] == 'completed' + assert run['metadata'] == expected + + +@pytest.mark.asyncio +async def test_finalize_without_metadata_preserves_legacy_snapshot(store, legacy_run): + run = await store.finalize_run(run_id='legacy', status='completed') + assert run['status'] == 'completed' + assert run['metadata'] == legacy_run + + +@pytest.mark.asyncio +async def test_metadata_update_still_rejects_new_inline_images(store, legacy_run): + with pytest.raises(sqlalchemy.exc.StatementError, match='Inline images cannot be persisted'): + await store.finalize_run( + run_id='legacy', status='completed', metadata={'new_image': {'type': 'Image', 'base64': 'eQ=='}} + ) + run = await store.get_run('legacy') + assert run['status'] == 'running' + assert run['metadata'] == legacy_run + + @pytest.mark.asyncio async def test_create_queued_run_claim_renew_release(store): run = await store.create_run( diff --git a/tests/unit_tests/api/service/test_maintenance_service.py b/tests/unit_tests/api/service/test_maintenance_service.py index c43ecbff5..77d6dbe90 100644 --- a/tests/unit_tests/api/service/test_maintenance_service.py +++ b/tests/unit_tests/api/service/test_maintenance_service.py @@ -294,7 +294,7 @@ class TestMaintenanceServiceGetStorageAnalysis: assert 'database' in result assert result['cleanup_policy'] == {'uploaded_file_retention_days': 7, 'log_retention_days': 3} assert { - section['key']: (section['path'], section['size_bytes'], section['file_count']) + section['key']: (Path(section['path']).as_posix(), section['size_bytes'], section['file_count']) for section in result['sections'] } == { 'database': ('data/langbot.db', 1000, 5), @@ -426,9 +426,8 @@ class TestMaintenanceServiceGetStorageAnalysis: result = await service.get_storage_analysis(TEST_CONTEXT) assert result['total_size_bytes'] == 125 - assert ( - next(section for section in result['sections'] if section['key'] == 'skills')['path'] - == '/core/skills/tenants/test' + assert next(section for section in result['sections'] if section['key'] == 'skills')['path'] == str( + Path('/core/skills/tenants/test') ) assert result['processes'][1]['directories'][0]['key'] == 'artifacts' assert result['processes'][2]['directories'][1]['key'] == 'mcp' diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index 63fbd32b5..39587f245 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -423,6 +423,99 @@ async def test_execution_detail_for_agent_and_pipeline(engine, service): await service.get_execution_detail(WORKSPACE, 'pipeline', 'missing-message') +@pytest.mark.asyncio +async def test_detail_overview_does_not_query_collapsed_sections(engine, service, monkeypatch): + await _seed(engine) + page = service._execution_page + converters = [] + + async def capture(workspace, statement, convert, offset, limit): + converters.append(convert.__name__) + return await page(workspace, statement, convert, offset, limit) + + async def reject_history(*args, **kwargs): + pytest.fail('Collapsed conversation history must not be loaded') + + monkeypatch.setattr(service, '_execution_page', capture) + monkeypatch.setattr(service, '_conversation_message_page', reject_history) + detail = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user', section='overview') + + assert set(detail['pages']) == {'inputs', 'outputs', 'deliveries', 'errors'} + assert detail['pages']['inputs']['items'][0]['id'] == 'msg-user' + assert len(converters) == 4 + assert 'events' not in detail + + +@pytest.mark.asyncio +async def test_conversation_pages_identities_from_covering_index_before_payloads(engine, service): + import sqlalchemy as sa + + await _seed(engine) + message = persistence_monitoring.MonitoringMessage + payload = 'historical inline media ' * 4096 + async with engine.begin() as connection: + await connection.execute( + sa.insert(message), + [ + dict( + id=f'history-{index:02d}', + workspace_uuid=workspace, + bot_id=bot, + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session-1', + timestamp=_dt(12), + role='user', + message_content=payload, + variables=payload, + status='success', + level='info', + ) + for index, (workspace, bot) in enumerate( + [(WORKSPACE, 'bot-1')] * 5 + [(WORKSPACE, 'other-bot'), ('other-workspace', 'bot-1')] + ) + ], + ) + statements = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + if statement.startswith('SELECT'): + statements.append((statement, parameters)) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture) + try: + first = await service.get_execution_detail(WORKSPACE, 'pipeline', 'msg-user', section='conversation', limit=2) + second = await service.get_execution_detail( + WORKSPACE, 'pipeline', 'msg-user', section='conversation', limit=2, offset=2 + ) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture) + + page = first['pages']['conversation'] + assert [row['id'] for row in page['items']] == ['history-04', 'history-03'] + assert page['has_more'] and page['next_offset'] == 2 + assert all(row['content'] == payload for row in page['items']) + assert [row['id'] for row in second['pages']['conversation']['items']] == ['history-02', 'history-01'] + identities = [(sql, params) for sql, params in statements if sql.startswith('SELECT monitoring_messages.id \n')] + assert len(identities) == 2 + async with engine.connect() as connection: + for sql, params in identities: + assert 'message_content' not in sql and 'variables' not in sql + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + sql, params)).all() + assert any('COVERING INDEX ix_monitoring_messages_conversation_page' in row[3] for row in plan), plan + assert not any('TEMP B-TREE' in row[3] for row in plan), plan + payload_queries = [(sql, params) for sql, params in statements if 'JOIN monitoring_messages ON' in sql] + assert len(payload_queries) == 2 + assert {value for value in payload_queries[0][1] if str(value).startswith('history-')} == { + 'history-04', + 'history-03', + } + for sql, params in payload_queries: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + sql, params)).all() + assert any('sqlite_autoindex_monitoring_messages_1 (id=?)' in row[3] for row in plan), plan + + @pytest.mark.asyncio async def test_explicit_deliveries_context_and_tenant_isolation(engine, service): import sqlalchemy as sa @@ -739,6 +832,217 @@ async def test_pipeline_debug_is_consistent_across_rows_rollups_and_telemetry(en assert sum(p['llm_calls'] for p in traffic['points']) == 1 +@pytest.mark.asyncio +async def test_inflight_excludes_unowned_and_expired_messages_without_rewriting_history(engine, service): + import sqlalchemy as sa + + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + Message = persistence_monitoring.MonitoringMessage + async with AsyncSession(engine) as session: + for message_id, owner, expiry, timestamp in [ + ('legacy', None, None, _dt(1)), + ('recent-unowned', None, now + datetime.timedelta(minutes=1), now), + ('expired', 'stopped-host', now - datetime.timedelta(seconds=1), _dt(1)), + ('long-running', 'live-host', now + datetime.timedelta(minutes=1), _dt(1)), + ]: + session.add( + Message( + id=message_id, + workspace_uuid=WORKSPACE, + timestamp=timestamp, + bot_id='bot', + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session', + message_content='historical message', + status='pending', + level='info', + execution_owner_id=owner, + execution_lease_expires_at=expiry, + ) + ) + session.add( + persistence_agent_run.AgentRun( + run_id='remote-queue', + workspace_id=WORKSPACE, + runner_id='runner', + status='queued', + queue_name='remote', + created_at=_dt(1), + ) + ) + await session.commit() + # A previous connection must not reintroduce unowned pending rows either. + snapshot = await service.get_inflight_snapshot(WORKSPACE, [{'source': 'pipeline', 'id': 'legacy'}]) + assert snapshot['active_total'] == 2 + assert {row['id'] for row in snapshot['items']} == {'long-running', 'remote-queue'} + async with AsyncSession(engine) as session: + statuses = (await session.execute(sa.select(Message.status))).scalars().all() + assert statuses == ['pending'] * 4 + + +@pytest.mark.asyncio +@pytest.mark.parametrize('source', ['all', 'pipeline']) +@pytest.mark.parametrize('statuses', [['running'], ['running', 'queued']]) +async def test_active_inflight_count_and_page_use_the_indexed_lease_range(engine, service, source, statuses): + import sqlalchemy as sa + + await _seed(engine) + async with engine.begin() as connection: + await connection.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-user') + .values(status='pending', message_content='historical payload ' * 8192) + ) + statements = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + if statement.startswith('SELECT '): + statements.append((statement, parameters)) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture) + try: + result = await service.get_executions( + WORKSPACE, statuses=statuses, source=source, include_summary=False, for_inflight=True + ) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture) + + assert result == {'items': [], 'total': 0, 'has_more': False} + assert len(statements) == 2 # Both COUNT and the sorted page must skip historical payloads. + async with engine.connect() as connection: + for statement, parameters in statements: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters)).all() + assert any( + 'ix_monitoring_messages_workspace_status_lease' in row[3] and 'execution_lease_expires_at>?' in row[3] + for row in plan + ), plan + + +@pytest.mark.asyncio +async def test_empty_inflight_snapshot_skips_linked_history_and_clears_tracking(engine, service, monkeypatch): + import sqlalchemy as sa + from langbot.pkg.utils.inflight import inflight_hub + + await _seed(engine) + monkeypatch.setattr(inflight_hub, 'run_workspaces', {'finished-run': WORKSPACE, 'other-run': 'other-workspace'}) + statements = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + statements.append(statement) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture) + try: + snapshot = await service.get_inflight_snapshot(WORKSPACE) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture) + + assert snapshot == {'items': [], 'active_total': 0, 'truncated': False} + assert inflight_hub.run_workspaces == {'other-run': 'other-workspace'} + assert not any('JOIN monitoring_messages ON' in query or 'FROM agent_run_event' in query for query in statements) + + +@pytest.mark.asyncio +async def test_inflight_never_loads_payloads_in_active_recent_or_terminal_rows(engine, service, monkeypatch): + import json + import sqlalchemy as sa + from langbot.pkg.entity.persistence.event_log import EventLog + + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + payload = json.dumps({'text': 'full input', 'attachment': 'x' * (2 * 1024 * 1024)}) + async with AsyncSession(engine) as session: + for message_id, status, timestamp in [ + ('active-message', 'pending', _dt(1)), + ('recent-message', 'success', now), + ('finished-message', 'success', _dt(1)), + ]: + session.add( + persistence_monitoring.MonitoringMessage( + id=message_id, + workspace_uuid=WORKSPACE, + timestamp=timestamp, + bot_id='bot', + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session', + message_content=payload, + variables=payload, + status=status, + level='info', + event_id=message_id, + run_id='linked-run' if status == 'pending' else None, + execution_owner_id='host' if status == 'pending' else None, + execution_lease_expires_at=now + datetime.timedelta(minutes=1) if status == 'pending' else None, + ) + ) + session.add( + EventLog( + event_id=message_id, + workspace_id=WORKSPACE, + event_type='message.received', + source='platform', + input_summary='small summary', + input_json=payload, + metadata_json=json.dumps({'platform': 'test', 'unused_payload': payload}), + ) + ) + for run_id in ['linked-run', 'standalone-run']: + session.add( + persistence_agent_run.AgentRun( + run_id=run_id, + workspace_id=WORKSPACE, + runner_id='runner', + status='running', + created_at=_dt(1), + authorization_json=payload, + metadata_json=json.dumps( + {'event_type': 'message.received', 'agent_name': 'Agent', 'input': {'text': payload}} + ), + ) + ) + await session.commit() + + linkage_queries = [] + + def reject_payload_reads(connection, cursor, statement, parameters, context, executemany): + for column in [ + 'monitoring_messages.message_content', + 'monitoring_messages.variables', + 'agent_run.authorization_json', + 'event_log.input_json', + ]: + assert column not in statement + if statement.startswith('SELECT agent_run.run_id,') and 'monitoring_messages.id AS message_id' in statement: + linkage_queries.append((statement, parameters)) + + deserialize = service._deserialize_json + + def bounded_metadata(raw): + assert not raw or len(raw) < 10000 + return deserialize(raw) + + monkeypatch.setattr(service, '_deserialize_json', bounded_metadata) + sa.event.listen(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + try: + snapshot = await service.get_inflight_snapshot(WORKSPACE, [{'source': 'pipeline', 'id': 'finished-message'}]) + rows = {row['id']: row for row in snapshot['items']} + assert rows['active-message']['input_preview'] == 'small summary' + assert rows['recent-message']['status_group'] == 'completed' + assert rows['finished-message']['status_group'] == 'completed' + assert rows['standalone-run']['target_name'] == 'Agent' + assert snapshot['active_total'] == 2 + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + assert linkage_queries + async with engine.connect() as connection: + for statement, parameters in linkage_queries: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters)).all() + message_lookups = [row[3] for row in plan if 'SEARCH monitoring_messages' in row[3]] + assert message_lookups and all('(id=?)' in lookup for lookup in message_lookups), plan + + @pytest.mark.asyncio async def test_inflight_snapshot_is_scoped_and_keeps_terminal_transition(engine, service): import sqlalchemy as sa diff --git a/tests/unit_tests/api/service/test_monitoring_sessions.py b/tests/unit_tests/api/service/test_monitoring_sessions.py index 891171ba3..5750e466e 100644 --- a/tests/unit_tests/api/service/test_monitoring_sessions.py +++ b/tests/unit_tests/api/service/test_monitoring_sessions.py @@ -245,3 +245,49 @@ async def test_old_active_sessions_are_listed_exported_and_not_cleaned(service): assert count == 1 rows, total = await service.get_sessions(context()) assert total == 1 and rows[0]['bot_id'] == 'b' + + +async def test_bot_session_summary_never_sorts_inline_message_payloads(service): + engine = service.ap.persistence_mgr.get_db_engine() + payload = 'historical inline media ' * 16384 + now = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None) + async with engine.begin() as connection: + await connection.execute( + sa.insert(models.MonitoringMessage), + [ + dict( + id=f'message-{index}', + workspace_uuid='workspace', + session_id='group_42', + **resource('a'), + timestamp=now + dt.timedelta(seconds=index), + role='user' if index == 0 else 'assistant', + message_content=payload, + variables=payload, + status='success', + level='info', + user_id='user', + user_name='User', + platform='test', + ) + for index in range(2) + ], + ) + + def reject_payload_reads(connection, cursor, statement, parameters, context, executemany): + assert 'monitoring_messages.message_content' not in statement + assert 'monitoring_messages.variables' not in statement + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + try: + rows, total = await service.get_sessions(context('a'), bot_ids=['a']) + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', reject_payload_reads) + + assert total == 1 + assert rows[0]['session_id'] == 'group_42' + assert rows[0]['message_count'] == 2 + assert rows[0]['user_name'] == 'User' + messages, total = await service.get_messages(context('a'), bot_ids=['a'], session_ids=['group_42']) + assert total == 2 + assert all(message['message_content'] == payload and message['variables'] == payload for message in messages) diff --git a/tests/unit_tests/persistence/test_conversation_page_index.py b/tests/unit_tests/persistence/test_conversation_page_index.py new file mode 100644 index 000000000..5a2b46dc8 --- /dev/null +++ b/tests/unit_tests/persistence/test_conversation_page_index.py @@ -0,0 +1,52 @@ +"""Conversation paging index is safe for existing and fresh databases.""" + +import importlib + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage + + +def test_conversation_index_migration_preserves_history_and_matches_metadata(): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') + index_name = 'ix_monitoring_messages_conversation_page' + engine = sa.create_engine('sqlite:///:memory:') + metadata = sa.MetaData() + table = sa.Table( + 'monitoring_messages', + metadata, + *[ + sa.Column(name, sa.String, primary_key=name == 'id') + for name in ( + 'id', + 'workspace_uuid', + 'bot_id', + 'session_id', + 'timestamp', + 'message_content', + 'status', + 'execution_lease_expires_at', + ) + ], + ) + sa.Index('existing_index', table.c.workspace_uuid) + try: + with engine.begin() as connection: + metadata.create_all(connection) + connection.execute(table.insert().values(id='history', message_content='inline media')) + with Operations.context(MigrationContext.configure(connection)): + migration.upgrade() + migration.upgrade() + indexes = { + index['name']: index['column_names'] for index in sa.inspect(connection).get_indexes(table.name) + } + expected = next(index for index in MonitoringMessage.__table__.indexes if index.name == index_name) + assert indexes[index_name] == [column.name for column in expected.columns] + migration.downgrade() + assert {index['name'] for index in sa.inspect(connection).get_indexes(table.name)} == {'existing_index'} + assert connection.scalar(sa.select(table.c.message_content)) == 'inline media' + migration.upgrade() + finally: + engine.dispose() diff --git a/tests/unit_tests/persistence/test_execution_lease_indexes.py b/tests/unit_tests/persistence/test_execution_lease_indexes.py new file mode 100644 index 000000000..d09e1dbfd --- /dev/null +++ b/tests/unit_tests/persistence/test_execution_lease_indexes.py @@ -0,0 +1,136 @@ +"""Lease maintenance must not scan a Workspace's completed execution history.""" + +import datetime +import importlib + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import create_async_engine + +# Register users and workspaces so create_all can resolve the foreign keys. +from langbot.pkg.entity.persistence import user, workspace # noqa: F401 +from langbot.pkg.entity.persistence.agent_run import AgentRun +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage +from langbot.pkg.utils.execution_lease import reconcile_execution_leases + + +MIGRATION = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') +LEASE_INDEXES = { + 'agent_run': ('ix_agent_run_workspace_status_lease', ['workspace_id', 'status', 'execution_lease_expires_at']), + 'monitoring_messages': ( + 'ix_monitoring_messages_workspace_status_lease', + ['workspace_uuid', 'status', 'execution_lease_expires_at'], + ), +} + + +def _create_schema(connection, source): + Base.metadata.create_all(connection) + if source == 'migration': + for table, (index, _) in LEASE_INDEXES.items(): + next(item for item in Base.metadata.tables[table].indexes if item.name == index).drop(connection) + with Operations.context(MigrationContext.configure(connection)): + MIGRATION.upgrade() + + +@pytest.mark.parametrize('source', ['metadata', 'migration']) +def test_lease_index_migration_preserves_data_and_existing_indexes(source): + engine = sa.create_engine('sqlite:///:memory:') + try: + with engine.begin() as connection: + _create_schema(connection, source) + connection.execute(sa.insert(AgentRun).values(run_id='history', runner_id='runner', status='completed')) + original_indexes = { + table: {index['name'] for index in sa.inspect(connection).get_indexes(table)} for table in LEASE_INDEXES + } + with Operations.context(MigrationContext.configure(connection)): + MIGRATION.upgrade() + for table, (index, columns) in LEASE_INDEXES.items(): + indexes = {item['name']: item['column_names'] for item in sa.inspect(connection).get_indexes(table)} + assert indexes[index] == columns + assert original_indexes[table] <= indexes.keys() + + MIGRATION.downgrade() + for table in LEASE_INDEXES: + remaining = {item['name'] for item in sa.inspect(connection).get_indexes(table)} + removed = {index for index_table, index, _ in MIGRATION.INDEXES if index_table == table} + assert remaining == original_indexes[table] - removed + assert connection.scalar(sa.select(AgentRun.status)) == 'completed' + + MIGRATION.upgrade() + assert connection.scalar(sa.select(AgentRun.status)) == 'completed' + finally: + engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('source', ['metadata', 'migration']) +@pytest.mark.parametrize('model', [AgentRun, MonitoringMessage]) +async def test_lease_queries_use_workspace_status_expiry_index(model, source): + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + now = datetime.datetime(2026, 10, 17) + statements = [] + + def capture_statement(connection, cursor, statement, parameters, context, executemany): + if statement.startswith(('SELECT ', 'UPDATE ')) and model.__tablename__ in statement: + statements.append((statement, parameters)) + + try: + async with engine.begin() as connection: + await connection.run_sync(lambda sync: _create_schema(sync, source)) + common = { + 'execution_owner_id': 'live-host', + 'execution_lease_expires_at': now - datetime.timedelta(seconds=1), + } + if model is AgentRun: + rows = [ + dict( + common, + run_id=str(i), + runner_id='runner', + workspace_id='workspace', + status='completed', + created_at=now, + updated_at=now, + ) + for i in range(2000) + ] + else: + rows = [ + dict( + common, + id=str(i), + workspace_uuid='workspace', + timestamp=now, + bot_id='bot', + bot_name='bot', + pipeline_id='pipeline', + pipeline_name='pipeline', + message_content='Historical message ' * 64, + session_id='session', + status='success', + level='info', + ) + for i in range(2000) + ] + await connection.execute(sa.insert(model), rows) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture_statement) + try: + assert await reconcile_execution_leases(engine, 'workspace', now=now, owner_id='live-host') == 0 + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture_statement) + + assert len(statements) == 2 + async with engine.connect() as connection: + for statement, parameters in statements: + plan = (await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters)).all() + details = ' '.join(row[3] for row in plan) + assert LEASE_INDEXES[model.__tablename__][0] in details, details + assert 'execution_lease_expires_at' in details, details + assert 'USE TEMP B-TREE' not in details, details + finally: + await engine.dispose() diff --git a/tests/unit_tests/persistence/test_media_reference_text.py b/tests/unit_tests/persistence/test_media_reference_text.py new file mode 100644 index 000000000..9d2d7c899 --- /dev/null +++ b/tests/unit_tests/persistence/test_media_reference_text.py @@ -0,0 +1,139 @@ +"""Image reference contracts apply to ORM and Core writes, including new callers.""" + +import json + +import pytest +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql, sqlite +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.orm import DeclarativeBase + +from langbot.pkg.entity.persistence.agent_run import AgentRun, AgentRunEvent +from langbot.pkg.entity.persistence.event_log import EventLog +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage +from langbot.pkg.entity.persistence.transcript import Transcript +from langbot.pkg.persistence.media import MediaReferenceText +from langbot.pkg.persistence.tenant_uow import ScopedSessionTransactionError, _validate_scoped_statement_call + + +class PayloadBase(DeclarativeBase): + pass + + +class Payload(PayloadBase): + __tablename__ = 'payload' + id = sa.Column(sa.Integer, primary_key=True) + content = sa.Column(MediaReferenceText) + + +@pytest.fixture +async def engine(): + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + async with engine.begin() as conn: + await conn.run_sync(PayloadBase.metadata.create_all) + yield engine + await engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('writer', ['orm', 'insert', 'update']) +@pytest.mark.parametrize( + 'image', + [ + {'type': 'Image', 'base64': 'eA=='}, + {'type': 'Image', 'url': 'data:image/png;base64,eA=='}, + {'type': 'image_base64', 'image_base64': 'eA=='}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA=='}}, + {'type': 'image', 'image_url': 'data:image/png;base64,eA=='}, + ], +) +async def test_unprocessed_images_are_rejected_at_the_bind_boundary(engine, writer, image): + content = json.dumps({'nested': [image]}) + async with engine.begin() as conn: + await conn.execute(sa.insert(Payload).values(id=1, content='original')) + with pytest.raises(sa.exc.StatementError, match='use MediaCache.externalize_chain_dump'): + if writer == 'orm': + async with AsyncSession(engine) as session: + session.add(Payload(id=2, content=content)) + await session.commit() + else: + statement = ( + sa.insert(Payload).values(id=2, content=content) + if writer == 'insert' + else sa.update(Payload).where(Payload.id == 1).values(content=content) + ) + async with engine.begin() as conn: + await conn.execute(statement) + async with engine.connect() as conn: + assert (await conn.execute(sa.select(Payload.content))).scalars().all() == ['original'] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'content', + [ + None, + 'plain text', + '{historical invalid JSON', + json.dumps({'type': 'Plain', 'text': 'base64 is a transport encoding'}), + json.dumps({'type': 'Image', 'url': 'https://example.com/image.png'}), + json.dumps({'type': 'Image', 'url': '/api/v1/files/media/hash.png', 'base64': None}), + ], +) +async def test_plain_text_and_references_round_trip_unchanged(engine, content): + async with AsyncSession(engine) as session: + session.add(Payload(id=1, content=content)) + await session.commit() + async with engine.connect() as conn: + assert (await conn.execute(sa.select(Payload.content))).scalar_one() == content + + +@pytest.mark.asyncio +async def test_legacy_images_can_be_read_compared_and_redacted(engine): + content = json.dumps({'type': 'Image', 'base64': 'eA=='}) + async with engine.begin() as conn: + # Simulate a row created before the contract was introduced. + await conn.execute(sa.text('INSERT INTO payload (id, content) VALUES (1, :content)'), {'content': content}) + assert ( + await conn.execute(sa.select(Payload.content).where(Payload.content == content)) + ).scalar_one() == content + assert (await conn.execute(sa.select(Payload.id).where(Payload.content.like('%base64%')))).scalar_one() == 1 + updated = await conn.execute( + sa.update(Payload).where(Payload.content == content).values(content='{"type": "Image", "base64": null}') + ) + assert updated.rowcount == 1 + + +@pytest.mark.parametrize('dialect', [sqlite.dialect(), postgresql.dialect()], ids=['sqlite', 'postgresql']) +def test_contract_keeps_text_schema_and_handles_escaped_json_keys(dialect): + column_type = MediaReferenceText() + assert column_type.compile(dialect=dialect) == 'TEXT' + process = column_type.dialect_impl(dialect).bind_processor(dialect) + with pytest.raises(ValueError, match='Inline images cannot be persisted'): + process(r'{"type": "Image", "base\u00364": "eA=="}') + + +@pytest.mark.parametrize( + 'model, fields', + [ + (MonitoringMessage, ['message_content', 'variables']), + (EventLog, ['input_json', 'metadata_json']), + (AgentRun, ['metadata_json']), + (AgentRunEvent, ['data_json', 'metadata_json']), + (Transcript, ['content_json', 'attachment_refs_json', 'metadata_json']), + ], +) +def test_conversation_payload_columns_declare_the_contract(model, fields): + for field in fields: + assert isinstance(model.__table__.c[field].type, MediaReferenceText), f'{model.__tablename__}.{field}' + + +def test_tenant_queries_admit_only_the_exact_trusted_media_type(): + _validate_scoped_statement_call((sa.select(Payload.content),), {}) + + class UntrustedMediaReferenceText(MediaReferenceText): + pass + + for column_type in (UntrustedMediaReferenceText(), MediaReferenceText(collation=sa.sql.quoted_name('x', False))): + with pytest.raises(ScopedSessionTransactionError): + _validate_scoped_statement_call((sa.select(sa.column('content', column_type)),), {}) diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index 63dfbfd0d..4032ad6ff 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -138,3 +138,147 @@ async def test_atbot_rule_no_match(mock_app, sample_query): result = await atbot_rule.match(str(message_chain), message_chain, {}, sample_query) assert result.matching is False + + +@pytest.mark.asyncio +@pytest.mark.parametrize('routed', [False, True]) +@pytest.mark.parametrize('kind', ['plain', 'mention', 'prefix']) +async def test_routed_group_messages_still_obey_response_rules(mock_app, sample_query, routed, kind): + resprule, entities, _, _ = get_modules() + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + sample_query.variables = {'_routed_by_rule': routed} + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': True, 'prefix': ['!bot '], 'regexp': [], 'random': 0}} + } + parts = [platform_message.Plain(text='!bot hello' if kind == 'prefix' else 'hello')] + if kind == 'mention': + parts.insert(0, platform_message.At(target='999')) + sample_query.message_chain = platform_message.MessageChain(parts) + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + expected = entities.ResultType.INTERRUPT if kind == 'plain' else entities.ResultType.CONTINUE + assert result.result_type == expected + if kind != 'plain': + assert str(sample_query.message_chain) == 'hello' + + +@pytest.mark.asyncio +@pytest.mark.parametrize('routed', [False, True]) +@pytest.mark.parametrize( + 'overrides,text,mention,draw,matches,expected_text', + [ + pytest.param({'regexp': ['^hello$']}, 'hello', False, 0.9, True, 'hello', id='regexp-match'), + pytest.param({'regexp': ['^hello$']}, 'bye', False, 0.9, False, 'bye', id='regexp-miss'), + pytest.param({'regexp': ['^bye$', '^hello$']}, 'hello', False, 0.9, True, 'hello', id='second-regexp'), + pytest.param({'regexp': ['[']}, 'hello', False, 0.9, False, 'hello', id='invalid-regexp'), + pytest.param({'random': 0}, 'hello', False, 0.0, False, 'hello', id='random-disabled'), + pytest.param({'random': 1}, 'hello', False, 0.999, True, 'hello', id='random-always'), + pytest.param({'random': 0.25}, 'hello', False, 0.249, True, 'hello', id='random-hit'), + pytest.param({'random': 0.25}, 'hello', False, 0.25, False, 'hello', id='random-boundary'), + pytest.param({}, 'hello', True, 0.0, False, 'hello', id='mention-disabled'), + pytest.param({}, '!bot hello', False, 0.0, False, '!bot hello', id='prefix-disabled'), + pytest.param({'prefix': ['!other ', '!bot ']}, '!bot hello', False, 0.9, True, 'hello', id='second-prefix'), + pytest.param( + {'at': True, 'prefix': ['!bot '], 'regexp': ['^hello$'], 'random': 0.25}, + 'hello', + False, + 0.9, + True, + 'hello', + id='combined-regexp-match', + ), + pytest.param( + {'at': True, 'prefix': ['!bot '], 'regexp': ['^bye$'], 'random': 0.25}, + 'hello', + False, + 0.1, + True, + 'hello', + id='combined-random-fallback', + ), + pytest.param( + {'at': True, 'prefix': ['!bot '], 'regexp': ['^bye$'], 'random': 0.25}, + 'hello', + False, + 0.9, + False, + 'hello', + id='combined-all-miss', + ), + ], +) +async def test_group_response_rule_configuration( + mock_app, sample_query, monkeypatch, routed, overrides, text, mention, draw, matches, expected_text +): + resprule, entities, _, _ = get_modules() + random_rule = import_module('langbot.pkg.pipeline.resprule.rules.random') + monkeypatch.setattr(random_rule.random, 'random', lambda: draw) + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + sample_query.variables = {'_routed_by_rule': routed} + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': False, 'prefix': [], 'regexp': [], 'random': 0, **overrides}} + } + parts = [platform_message.Plain(text=text)] + if mention: + parts.insert(0, platform_message.At(target='999')) + sample_query.message_chain = platform_message.MessageChain(parts) + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + assert result.result_type == (entities.ResultType.CONTINUE if matches else entities.ResultType.INTERRUPT) + assert str(result.new_query.message_chain) == expected_text + + +@pytest.mark.asyncio +@pytest.mark.parametrize('callback_key', ['_interaction_submission', '_dify_form_action']) +async def test_interaction_callbacks_do_not_require_another_mention(mock_app, sample_query, callback_key): + resprule, entities, _, _ = get_modules() + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + submission = {'action_id': 'approve'} + sample_query.variables = {'_routed_by_rule': True, callback_key: submission} + sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='[Form Action: Approve]')]) + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': True, 'prefix': ['!bot '], 'regexp': [], 'random': 0}} + } + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + assert result.result_type == entities.ResultType.CONTINUE + assert str(result.new_query.message_chain) == '[Form Action: Approve]' + assert result.new_query.variables[callback_key] == submission + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'variables', + [ + {'_routed_by_rule': True, '_dify_form_action': None}, + {'_routed_by_rule': True, '_dify_form_action': {}}, + {'_routed_by_rule': False, '_dify_form_action': {'action_id': 'approve'}}, + ], +) +async def test_incomplete_form_callback_does_not_bypass_response_rules(mock_app, sample_query, variables): + resprule, entities, _, _ = get_modules() + sample_query.launcher_type = provider_session.LauncherTypes.GROUP + sample_query.adapter.bot_account_id = '999' + sample_query.variables = variables + sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='[Form Action: Approve]')]) + sample_query.pipeline_config = { + 'trigger': {'group-respond-rules': {'at': True, 'prefix': ['!bot '], 'regexp': [], 'random': 0}} + } + stage = resprule.GroupRespondRuleCheckStage(mock_app) + await stage.initialize(sample_query.pipeline_config) + + result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') + + assert result.result_type == entities.ResultType.INTERRUPT diff --git a/tests/unit_tests/provider/test_provider_specific_fields.py b/tests/unit_tests/provider/test_provider_specific_fields.py index fe239f3ea..803754066 100644 --- a/tests/unit_tests/provider/test_provider_specific_fields.py +++ b/tests/unit_tests/provider/test_provider_specific_fields.py @@ -4,15 +4,17 @@ This includes provider_specific_fields round-trip coverage for GitHub issue #1899 and token counting preflight behavior for Runner context budgeting. """ +import json from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch +import httpx import pytest import langbot_plugin.api.entities.builtin.provider.message as provider_message import langbot_plugin.api.entities.builtin.resource.tool as resource_tool from langbot.pkg.provider.modelmgr import requester as model_requester -from langbot.pkg.provider.modelmgr.requesters.litellmchat import LiteLLMRequester +from langbot.pkg.provider.modelmgr.requesters.litellmchat import LiteLLMRequester, _LeasedHTTPHandler def _make_requester() -> LiteLLMRequester: @@ -148,6 +150,132 @@ def test_convert_messages_preserves_message_provider_specific_fields(): assert out[0]['provider_specific_fields']['thought_signatures'] == ['sig1', 'sig2'] +@pytest.mark.parametrize('family', ['', 'openai', 'gemini', 'groq']) +def test_response_finish_reason_is_not_replayed(family): + req = _make_requester() + msg = provider_message.Message( + role='assistant', content='Hello', provider_specific_fields={'finish_reason': 'stop'} + ) + + assert req._convert_messages([msg], reasoning_family=family) == [{'role': 'assistant', 'content': 'Hello'}] + assert msg.provider_specific_fields == {'finish_reason': 'stop'} + + +def test_response_metadata_removal_preserves_gemini_signatures(): + req = _make_requester() + msg = provider_message.Message( + role='assistant', + content='Hello', + provider_specific_fields={'finish_reason': 'stop', 'thought_signatures': ['signature']}, + ) + + out = req._convert_messages([msg], reasoning_family='gemini') + + assert out[0]['provider_specific_fields'] == {'thought_signatures': ['signature']} + + +@pytest.fixture +def replayed_tool_message(): + return provider_message.Message( + role='assistant', + content='Hello', + provider_specific_fields={'finish_reason': 'tool_calls', 'other_response_metadata': True}, + tool_calls=[ + provider_message.ToolCall( + id='call-1', + type='function', + function=provider_message.FunctionCall(name='search', arguments='{}'), + provider_specific_fields={'thought_signature': 'signature'}, + ) + ], + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +@pytest.mark.parametrize('chat_provider,family', [('groq', 'openai'), ('gemini', 'groq'), ('openrouter', 'groq')]) +async def test_completion_metadata_adaptation_uses_chat_provider(stream, chat_provider, family, replayed_tool_message): + req = _make_configured_requester() + req.requester_cfg['custom_llm_provider'] = chat_provider + model = _make_runtime_model() + model.model_entity.name = 'openai/gpt-oss-120b' + msg = replayed_tool_message + original = msg.model_dump() + + with patch.object(req, '_reasoning_family', return_value=family): + args = await req._build_completion_args(model, [msg], None, {}, stream=stream) + + assert args['model'] == f'{chat_provider}/openai/gpt-oss-120b' + expected = msg.model_dump(exclude_none=True) + expected['provider_specific_fields'].pop('finish_reason') + if chat_provider == 'groq': + expected.pop('provider_specific_fields') + expected['tool_calls'][0].pop('provider_specific_fields') + assert args['messages'] == [expected] + assert msg.model_dump() == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +async def test_groq_http_request_excludes_replayed_metadata(stream, replayed_tool_message): + req = _make_configured_requester() + req.requester_cfg.update(custom_llm_provider='groq', base_url='https://groq.test/openai/v1') + model = _make_runtime_model() + model.model_entity.name = 'openai/gpt-oss-120b' + msg = replayed_tool_message + original = msg.model_dump() + bodies = [] + + def handle(request): + assert request.url == 'https://groq.test/openai/v1/chat/completions' + bodies.append(json.loads(request.content)) + response = { + 'id': 'completion-1', + 'object': 'chat.completion', + 'created': 1, + 'model': 'openai/gpt-oss-120b', + 'service_tier': 'on_demand', + 'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': 'Done'}, 'finish_reason': 'stop'}], + 'usage': {'prompt_tokens': 1, 'completion_tokens': 1, 'total_tokens': 2}, + } + if stream: + response['object'] = 'chat.completion.chunk' + response['choices'][0]['delta'] = response['choices'][0].pop('message') + return httpx.Response( + 200, + headers={'content-type': 'text/event-stream'}, + content=f'data: {json.dumps(response)}\n\ndata: [DONE]\n\n', + ) + return httpx.Response(200, json=response) + + messages = [ + provider_message.Message(role='user', content='Search'), + msg, + provider_message.Message(role='tool', tool_call_id='call-1', content='Found'), + ] + async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client: + extra_args = {'client': _LeasedHTTPHandler(client, 10)} + if stream: + chunks = [chunk async for chunk in req.invoke_llm_stream(None, model, messages, extra_args=extra_args)] + assert ''.join(chunk.content or '' for chunk in chunks) == 'Done' + else: + result, _ = await req.invoke_llm(None, model, messages, extra_args=extra_args) + assert result.content == 'Done' + + assert len(bodies) == 1 + assert bodies[0]['model'] == 'openai/gpt-oss-120b' + assert bodies[0]['messages'] == [ + {'role': 'user', 'content': 'Search'}, + { + 'role': 'assistant', + 'content': 'Hello', + 'tool_calls': [{'id': 'call-1', 'type': 'function', 'function': {'name': 'search', 'arguments': '{}'}}], + }, + {'role': 'tool', 'tool_call_id': 'call-1', 'content': 'Found'}, + ] + assert msg.model_dump() == original + + def test_normalize_stream_tool_calls_preserves_provider_specific_fields(): """Streaming tool calls should retain provider_specific_fields.""" req = _make_requester() diff --git a/tests/unit_tests/storage/test_inline_media_cleanup.py b/tests/unit_tests/storage/test_inline_media_cleanup.py new file mode 100644 index 000000000..e3aa1bafe --- /dev/null +++ b/tests/unit_tests/storage/test_inline_media_cleanup.py @@ -0,0 +1,356 @@ +"""Historical image TTL preserves records and makes bounded, durable progress.""" + +import datetime as dt +import importlib +import json +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.api.http.context import ExecutionContext +from langbot.pkg.api.http.service.maintenance import MaintenanceService +from langbot.pkg.entity.persistence.agent_run import AgentRun, AgentRunEvent +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence.event_log import EventLog +from langbot.pkg.entity.persistence.metadata import WorkspaceMetadata +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage +from langbot.pkg.entity.persistence.transcript import Transcript +from langbot.pkg.entity.persistence.workspace import Workspace +from langbot.pkg.persistence.mgr import PersistenceManager +from langbot.pkg.storage import inline_media_cleanup as cleanup +from langbot.pkg.storage.media import MediaCache + + +WORKSPACE_A = '00000000-0000-0000-0000-00000000000a' +WORKSPACE_B = '00000000-0000-0000-0000-00000000000b' +CONTEXT = ExecutionContext(instance_uuid='instance', workspace_uuid=WORKSPACE_A, placement_generation=1) +NOW = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None) +OLD = NOW - dt.timedelta(days=40) +PAYLOAD = json.dumps( + { + 'content': [ + {'type': 'Plain', 'text': 'Keep the message'}, + {'type': 'Image', 'base64': 'eA==', 'url': 'https://upstream/image'}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA=='}}, + {'type': 'image_base64', 'image_base64': 'eA=='}, + {'type': 'Image', 'url': '/api/v1/files/media/already-externalized.png'}, + ], + 'custom': {'count': 0, 'enabled': False}, + } +) +SOURCES = [MonitoringMessage, EventLog, AgentRun, AgentRunEvent, Transcript] + + +@pytest.fixture +async def application(tmp_path): + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "inline-media.db"}') + ap = SimpleNamespace(logger=logging.getLogger(__name__), instance_config=SimpleNamespace(data={})) + ap.persistence_mgr = PersistenceManager(ap) + ap.persistence_mgr.db = SimpleNamespace(get_engine=lambda: engine) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + await conn.execute( + sa.insert(Workspace), + [ + { + 'uuid': workspace, + 'instance_uuid': 'instance', + 'name': workspace, + 'slug': workspace, + 'source': 'cloud_projection', + } + for workspace in (WORKSPACE_A, WORKSPACE_B) + ], + ) + yield ap + await engine.dispose() + + +async def insert_record(ap, model, number, workspace=WORKSPACE_A, timestamp=OLD, payload=PAYLOAD): + if model is MonitoringMessage: + values = dict( + id=str(number), + workspace_uuid=workspace, + timestamp=timestamp, + message_content=payload, + bot_id='bot', + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + session_id='session', + status='success', + level='info', + ) + field = 'message_content' + elif model is AgentRunEvent: + run_id = f'parent-{number}' + await ap.persistence_mgr.execute_async( + sa.insert(AgentRun).values( + run_id=run_id, runner_id='runner', workspace_id=workspace, status='completed', created_at=timestamp + ) + ) + values = dict( + id=number, run_id=run_id, sequence=1, type='message.completed', created_at=timestamp, data_json=payload + ) + field = 'data_json' + else: + values = dict(id=number, workspace_id=workspace, created_at=timestamp) + if model is EventLog: + values.update( + event_id=f'event-{number}', event_type='message.received', source='platform', input_json=payload + ) + field = 'input_json' + elif model is AgentRun: + values.update( + run_id=f'run-{number}', + runner_id='runner', + status='completed', + metadata_json=payload, + updated_at=timestamp, + ) + field = 'metadata_json' + else: + values.update( + transcript_id=f'transcript-{number}', + event_id=f'event-{number}', + conversation_id='conversation', + role='user', + item_type='message', + seq=number, + content_json=payload, + ) + field = 'content_json' + # Seed the pre-contract database format without applying today's write policy. + values[field] = sa.bindparam('historical_payload', payload, type_=sa.Text()) + await ap.persistence_mgr.execute_async(sa.insert(model).values(**values)) + return field + + +async def read_record(ap, model, number, field): + return ( + await ap.persistence_mgr.execute_async(sa.select(getattr(model, field)).where(model.id == number)) + ).scalar_one() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('model', SOURCES, ids=lambda model: model.__tablename__) +async def test_ttl_expires_only_old_images_in_the_current_workspace(application, model): + field = await insert_record(application, model, 1) + await insert_record(application, model, 2, timestamp=NOW) + await insert_record(application, model, 3, workspace=WORKSPACE_B) + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + saved = json.loads(await read_record(application, model, 1, field)) + assert not MediaCache.has_inline_images(saved) + assert saved['content'][0]['text'] == 'Keep the message' + assert saved['content'][1]['url'] == 'https://upstream/image' + assert saved['content'][-1]['url'].startswith('/api/v1/files/media/') + assert saved['custom'] == {'count': 0, 'enabled': False} + assert await read_record(application, model, 2, field) == PAYLOAD + assert await read_record(application, model, 3, field) == PAYLOAD + if model is AgentRun: + assert await read_record(application, model, 1, 'updated_at') == OLD + + +@pytest.mark.asyncio +async def test_row_budget_resumes_persisted_progress_and_later_expiry(application, monkeypatch): + monkeypatch.setattr(cleanup, 'ROWS_PER_TABLE', 1) + for number in (1, 2): + await insert_record(application, MonitoringMessage, number) + await insert_record(application, MonitoringMessage, 3, timestamp=NOW - dt.timedelta(days=5)) + for number in (1, 2): + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + assert not MediaCache.has_inline_images( + json.loads(await read_record(application, MonitoringMessage, number, 'message_content')) + ) + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 0 + assert await cleanup.cleanup_inline_media(application, CONTEXT, 1) == 1 + assert (await application.persistence_mgr.execute_async(sa.select(WorkspaceMetadata.value))).first() + + +@pytest.mark.asyncio +async def test_event_scan_budget_advances_past_foreign_events_without_reading_payloads(application, monkeypatch): + monkeypatch.setattr(cleanup, 'ROWS_PER_TABLE', 2) + for number in range(1, 5): + await insert_record(application, AgentRunEvent, number, workspace=WORKSPACE_B) + await application.persistence_mgr.execute_async(sa.delete(AgentRun).where(AgentRun.run_id == 'parent-3')) + await insert_record(application, AgentRunEvent, 5) + await insert_record(application, AgentRunEvent, 6, timestamp=NOW) + engine = application.persistence_mgr.get_db_engine() + statements = [] + + def capture_statement(connection, cursor, statement, parameters, context, executemany): + if statement.startswith('SELECT ') and 'FROM agent_run_event' in statement: + statements.append((statement, parameters)) + + sa.event.listen(engine.sync_engine, 'before_cursor_execute', capture_statement) + try: + for expected_id in (2, 4): + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 0 + cursor = ( + await application.persistence_mgr.execute_async( + sa.select(WorkspaceMetadata.value).where( + WorkspaceMetadata.workspace_uuid == WORKSPACE_A, + WorkspaceMetadata.key == cleanup.CURSOR_PREFIX + 'agent_run_event', + ) + ) + ).scalar_one() + assert json.loads(cursor) == [OLD.isoformat(), expected_id] + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 0 + finally: + sa.event.remove(engine.sync_engine, 'before_cursor_execute', capture_statement) + + assert not MediaCache.has_inline_images(json.loads(await read_record(application, AgentRunEvent, 5, 'data_json'))) + for number in (1, 2, 3, 4, 6): + assert await read_record(application, AgentRunEvent, number, 'data_json') == PAYLOAD + async with engine.connect() as connection: + for statement, parameters in statements: + plan = ' '.join( + row[3] for row in await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters) + ) + if 'CORRELATED' in plan: + assert 'SEARCH agent_run_event USING INTEGER PRIMARY KEY' in plan, plan + else: + assert 'COVERING INDEX ix_agent_run_event_media_ttl' in plan, plan + assert 'TEMP B-TREE' not in plan, plan + + +@pytest.mark.asyncio +async def test_payload_budget_processes_one_oversized_row_then_yields(application, monkeypatch): + monkeypatch.setattr(cleanup, 'CHARACTERS_PER_TABLE', 1) + await insert_record(application, MonitoringMessage, 1) + await insert_record(application, MonitoringMessage, 2) + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + assert await read_record(application, MonitoringMessage, 2, 'message_content') == PAYLOAD + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + + +@pytest.mark.asyncio +async def test_payload_update_and_cursor_roll_back_together(application, monkeypatch): + await insert_record(application, MonitoringMessage, 1) + execute = application.persistence_mgr.execute_async + + async def fail_checkpoint(statement, *args, **kwargs): + if isinstance(statement, sa.sql.dml.Insert) and statement.table.name == 'workspace_metadata': + raise RuntimeError('Checkpoint failed') + return await execute(statement, *args, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(application.persistence_mgr, 'execute_async', fail_checkpoint) + with pytest.raises(RuntimeError, match='Checkpoint failed'): + await cleanup.cleanup_inline_media(application, CONTEXT, 30) + assert await read_record(application, MonitoringMessage, 1, 'message_content') == PAYLOAD + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + + +@pytest.mark.asyncio +async def test_unchanged_or_invalid_records_are_not_rewritten_or_revisited(application): + values = [ + 'plain text', + '{invalid base64 JSON', + '{"count": 0, "enabled": false}', + json.dumps({'type': 'Image', 'url': '/api/v1/files/media/existing.png'}), + ] + for number, payload in enumerate(values, 1): + await insert_record(application, MonitoringMessage, number, payload=payload) + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 0 + for number, payload in enumerate(values, 1): + assert await read_record(application, MonitoringMessage, number, 'message_content') == payload + before = (await application.persistence_mgr.execute_async(sa.select(WorkspaceMetadata.value))).scalars().all() + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 0 + after = (await application.persistence_mgr.execute_async(sa.select(WorkspaceMetadata.value))).scalars().all() + assert before == after + + +@pytest.mark.asyncio +@pytest.mark.parametrize('model', [MonitoringMessage, AgentRunEvent]) +async def test_concurrent_payload_change_does_not_advance_cursor(application, monkeypatch, model): + field = await insert_record(application, model, 1) + execute = application.persistence_mgr.execute_async + + async def lose_update(statement, *args, **kwargs): + if isinstance(statement, sa.sql.dml.Update) and statement.table.name == model.__tablename__: + return SimpleNamespace(rowcount=0) + return await execute(statement, *args, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(application.persistence_mgr, 'execute_async', lose_update) + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 0 + assert ( + await execute( + sa.select(WorkspaceMetadata.value).where( + WorkspaceMetadata.workspace_uuid == WORKSPACE_A, + WorkspaceMetadata.key == cleanup.CURSOR_PREFIX + model.__tablename__, + ) + ) + ).first() is None + assert await read_record(application, model, 1, field) == PAYLOAD + assert await cleanup.cleanup_inline_media(application, CONTEXT, 30) == 1 + + +@pytest.mark.asyncio +async def test_storage_maintenance_uses_media_ttl_for_inline_records(application): + await insert_record(application, MonitoringMessage, 1, timestamp=NOW - dt.timedelta(days=5)) + application.instance_config.data = {'storage': {'media_cache': {'retention_days': 2}}} + cache = SimpleNamespace(cleanup=AsyncMock(return_value={})) + application.storage_mgr = SimpleNamespace(media_cache=cache) + service = MaintenanceService(application) + service._is_oss_singleton = AsyncMock(return_value=False) + service._cleanup_expired_uploaded_files = AsyncMock(return_value=0) + assert (await service.cleanup_expired_files(CONTEXT))['inline_media_records'] == 1 + cache.cleanup.assert_not_awaited() + + +def test_ttl_indexes_migrate_idempotently_and_support_cursor_queries(): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') + engine = sa.create_engine('sqlite:///:memory:') + try: + with engine.begin() as connection: + Base.metadata.create_all(connection) + with Operations.context(MigrationContext.configure(connection)): + migration.downgrade() + migration.upgrade() + migration.upgrade() + for table, name, columns in migration.INDEXES: + indexes = { + index['name']: index['column_names'] for index in sa.inspect(connection).get_indexes(table) + } + assert indexes[name] == columns + orm_index = next(index for index in Base.metadata.tables[table].indexes if index.name == name) + assert [column.name for column in orm_index.columns] == columns + query = ( + sa.select(MonitoringMessage.id, MonitoringMessage.message_content) + .where( + MonitoringMessage.workspace_uuid == WORKSPACE_A, + MonitoringMessage.timestamp < NOW, + sa.tuple_(MonitoringMessage.timestamp, MonitoringMessage.id) > sa.tuple_(OLD, '0'), + ) + .order_by(MonitoringMessage.timestamp, MonitoringMessage.id) + .limit(1) + ) + sql = str(query.compile(dialect=engine.dialect, compile_kwargs={'literal_binds': True})) + plan = ' '.join(str(row) for row in connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + sql)) + assert 'ix_monitoring_messages_media_ttl' in plan + assert 'TEMP B-TREE' not in plan + migration.downgrade() + finally: + engine.dispose() + + +def test_index_migration_allows_historical_schemas_without_monitoring_tables(): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') + engine = sa.create_engine('sqlite:///:memory:') + try: + with engine.begin() as connection: + with Operations.context(MigrationContext.configure(connection)): + migration.upgrade() + migration.downgrade() + assert not sa.inspect(connection).get_table_names() + finally: + engine.dispose() diff --git a/tests/unit_tests/storage/test_media_cache.py b/tests/unit_tests/storage/test_media_cache.py index 054ee725b..5ba502ad3 100644 --- a/tests/unit_tests/storage/test_media_cache.py +++ b/tests/unit_tests/storage/test_media_cache.py @@ -1,5 +1,6 @@ import base64 import datetime +import json import os import tempfile from pathlib import Path @@ -127,6 +128,51 @@ class TestMediaCache: # Should not raise; base64 should be stripped as fallback assert result[0]['base64'] is None + @pytest.mark.asyncio + @pytest.mark.parametrize( + 'image', + [ + {'type': 'Image', 'base64': 'eA=='}, + {'type': 'image_base64', 'image_base64': 'eA=='}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA==', 'detail': 'low'}}, + {'type': 'image', 'image_url': 'data:image/png;base64,eA=='}, + {'type': 'Image', 'url': 'data:image/png;base64,eA=='}, + ], + ) + async def test_externalize_image_encodings_without_mutating_runtime_input(self, image): + original = json.dumps(image) + result = await self.media_cache.externalize_chain_dump({'nested': [image]}) + saved = result['nested'][0] + assert saved['storage_key'].startswith('media_cache/') + self.mock_provider.save.assert_awaited_once_with(saved['storage_key'], b'x') + assert not MediaCache.has_inline_images(result) + assert json.dumps(image) == original + if image.get('type') == 'image_url': + assert saved['image_url']['detail'] == 'low' + + @pytest.mark.asyncio + @pytest.mark.parametrize('payload', ['invalid!!!', 'https://example.com/image.png']) + async def test_invalid_base64_is_removed_without_saving_or_downloading(self, payload): + result = await self.media_cache.externalize_chain_dump({'type': 'Image', 'base64': payload}) + assert result['base64'] is None + self.mock_provider.save.assert_not_awaited() + + @pytest.mark.asyncio + async def test_url_only_images_do_not_enter_cas(self): + image = {'type': 'Image', 'url': 'https://example.com/image.png'} + assert not MediaCache.has_inline_images(image) + assert await self.media_cache.externalize_chain_dump(image) == image + self.mock_provider.exists.assert_not_awaited() + self.mock_provider.save.assert_not_awaited() + + @pytest.mark.asyncio + async def test_oversized_inline_image_is_removed_before_decode(self): + image = {'type': 'Image', 'base64': 'data:image/png;base64,' + 'A' * 100} + with patch('langbot.pkg.storage.media.DEFAULT_BASE64_MEDIA_LIMIT', 16): + result = await self.media_cache.externalize_chain_dump(image) + assert result['base64'] is None + self.mock_provider.save.assert_not_awaited() + @pytest.mark.asyncio async def test_cleanup_retention_and_max_size(self): with tempfile.TemporaryDirectory() as temp_dir: @@ -197,8 +243,8 @@ class TestMonitoringServiceSanitization: def test_sanitize_oversized_base64_payload(self): svc = MonitoringService.__new__(MonitoringService) small_content = '{"type": "Image", "base64": "data:image/png;base64,tiny"}' - # Should leave small contents untouched - assert svc._sanitize_message_content(small_content) == small_content + # Metadata-only records must not retain even small inline images. + assert json.loads(svc._sanitize_message_content(small_content))['base64'] is None # Large content with base64 data URL huge_b64 = 'A' * 60000 @@ -213,3 +259,21 @@ class TestMonitoringServiceSanitization: sanitized_raw = svc._sanitize_message_content(raw_corrupted) assert '[base64 image omitted]' in sanitized_raw assert multiline_b64 not in sanitized_raw + + @pytest.mark.parametrize( + 'image', + [ + {'type': 'Image', 'base64': 'eA=='}, + {'type': 'Image', 'base64': 'A' * 20000}, + {'type': 'image_base64', 'image_base64': 'eA=='}, + {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,eA=='}}, + ], + ) + def test_redaction_covers_raw_and_provider_images_at_any_size(self, image): + svc = MonitoringService.__new__(MonitoringService) + payload = {'event': {'images': [image]}, 'count': 0, 'enabled': False} + saved = json.loads(svc._sanitize_message_content(json.dumps(payload))) + assert not MediaCache.has_inline_images(saved) + assert saved['count'] == 0 + assert saved['enabled'] is False + assert MediaCache.has_inline_images(payload) diff --git a/tests/unit_tests/storage/test_media_persistence.py b/tests/unit_tests/storage/test_media_persistence.py new file mode 100644 index 000000000..0b1fb9318 --- /dev/null +++ b/tests/unit_tests/storage/test_media_persistence.py @@ -0,0 +1,256 @@ +"""Persist media references without changing live messages or metadata-only policy.""" + +import base64 +import copy +import json +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import sqlalchemy as sa + +from langbot.pkg.agent.runner.run_journal import AgentRunJournal +from langbot.pkg.agent.runner.run_ledger_store import RunLedgerStore +from langbot.pkg.entity.persistence.agent_run import AgentRunEvent +from langbot.pkg.entity.persistence.event_log import EventLog +from langbot.pkg.entity.persistence.monitoring import MonitoringMessage +from langbot.pkg.entity.persistence.transcript import Transcript +from langbot.pkg.pipeline.monitoring_helper import MonitoringHelper +from langbot.pkg.storage.media import MediaCache +from tests.unit_tests.api.service.test_monitoring_tenancy import ( + WORKSPACE_A, + _context, + _record_message, + service as base_service, # noqa: F401 +) + + +@pytest.fixture +def media_service(base_service): # noqa: F811 + service = base_service + objects = {} + + async def save(key, data): + objects[key] = data + + provider = SimpleNamespace( + exists=AsyncMock(side_effect=lambda key: key in objects), + save=AsyncMock(side_effect=save), + ) + service.ap.logger = logging.getLogger(__name__) + service.ap.storage_mgr = SimpleNamespace(storage_provider=provider) + cache = MediaCache(service.ap, service.ap.storage_mgr) + service.ap.storage_mgr.media_cache = cache + service.ap.monitoring_service = service + return service, objects, cache + + +def image(raw=b'tiny adapter image', *, encoded=True): + payload = base64.b64encode(raw).decode() + return {'type': 'Image', 'base64': f'data:image/png;base64,{payload}' if encoded else payload} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'response_kind', + ['input', 'resp_message_chain', 'resp_messages', 'direct', 'structured', 'structured_dict', 'discarded'], +) +async def test_all_retained_monitoring_messages_use_cas(media_service, response_kind, monkeypatch): + service, objects, _ = media_service + context = _context(WORKSPACE_A) + chain = [image()] + original = copy.deepcopy(chain) + legacy_content = json.dumps(chain) + json_dumps = json.dumps + + def serialize_references(value, *args, **kwargs): + assert not MediaCache.has_inline_images(value), 'Inline images must reach CAS before JSON serialization' + return json_dumps(value, *args, **kwargs) + + monkeypatch.setattr(json, 'dumps', serialize_references) + query = SimpleNamespace( + _execution_context=context, + launcher_type='person', + launcher_id='user', + sender_id='user', + message_chain=SimpleNamespace(model_dump=lambda: chain), + variables={}, + ) + resource = dict(bot_id='same-bot', bot_name='Bot', pipeline_id='pipeline', pipeline_name='Pipeline') + if response_kind == 'input': + await MonitoringHelper.record_query_start(service.ap, query, **resource) + elif response_kind == 'direct': + await _record_message(service, context, legacy_content) + elif response_kind in ('structured', 'structured_dict'): + content = {'chain': chain} if response_kind == 'structured_dict' else chain + await _record_message(service, context, content) + elif response_kind == 'discarded': + from langbot.pkg.platform.botmgr import RuntimeBot + + bot = SimpleNamespace( + ap=service.ap, + execution_context=context, + bot_entity=SimpleNamespace(uuid='same-bot', name='Bot'), + PIPELINE_DISCARD=RuntimeBot.PIPELINE_DISCARD, + PIPELINE_DISCARD_DISPLAY_NAME=RuntimeBot.PIPELINE_DISCARD_DISPLAY_NAME, + logger=SimpleNamespace(error=AsyncMock()), + ) + await RuntimeBot._record_discarded_message( + bot, 'person', 'user', 'user', SimpleNamespace(), query.message_chain + ) + else: + response = SimpleNamespace(model_dump=lambda: chain) + if response_kind == 'resp_message_chain': + query.resp_message_chain = [response] + else: + query.resp_messages = [SimpleNamespace(get_content_platform_message_chain=lambda: response)] + await MonitoringHelper.record_query_response(service.ap, query, **resource) + + async with service.ap.persistence_mgr.get_db_engine().connect() as conn: + stored = (await conn.execute(sa.select(MonitoringMessage.message_content))).scalar_one() + saved_chain = json.loads(stored) + saved = (saved_chain['chain'] if isinstance(saved_chain, dict) else saved_chain)[0] + assert saved['base64'] is None + assert saved['url'].startswith('/api/v1/files/media/') + assert objects[saved['storage_key']] == b'tiny adapter image' + assert chain == original + + +@pytest.mark.asyncio +async def test_metadata_only_ingress_discards_images_without_populating_cas(media_service): + service, objects, _ = media_service + payload = {'type': 'message.received', 'message_chain': [image(encoded=False)], 'count': 0} + original = copy.deepcopy(payload) + await service.record_ingress_event( + _context(WORKSPACE_A), + event_id='ingress', + event_type='message.received', + bot_id='same-bot', + bot_name='Bot', + platform='slack', + payload=payload, + ) + async with service.ap.persistence_mgr.get_db_engine().connect() as conn: + stored = (await conn.execute(sa.select(EventLog.input_json))).scalar_one() + assert json.loads(stored)['message_chain'][0]['base64'] is None + assert not objects + assert payload == original + + +@pytest.mark.asyncio +async def test_repeated_images_deduplicate_across_messages_and_run_events(media_service): + service, objects, cache = media_service + raw = b'shared image' + payload = image(raw) + await _record_message(service, _context(WORKSPACE_A), json.dumps([payload])) + journal = AgentRunJournal(service.ap) + result = { + 'type': 'message.completed', + 'data': { + 'message': {'content': [payload]}, + 'provider_image': {'type': 'image_base64', 'image_base64': payload['base64']}, + }, + } + original = copy.deepcopy(result) + metadata = {'diagnostic_image': image(b'metadata-only image')} + await journal.append_run_result(result_dict=result, run_id='run', sequence=1, metadata=metadata) + async with service.ap.persistence_mgr.get_db_engine().connect() as conn: + stored = (await conn.execute(sa.select(AgentRunEvent.data_json))).scalar_one() + saved_metadata = (await conn.execute(sa.select(AgentRunEvent.metadata_json))).scalar_one() + saved = json.loads(stored) + assert saved['message']['content'][0]['hash'] == cache.hash_bytes(raw) + assert saved['provider_image']['image_base64'] is None + assert saved['provider_image']['image_url']['url'].startswith('/api/v1/files/media/') + assert len(objects) == 1 + service.ap.storage_mgr.storage_provider.save.assert_awaited_once() + assert result == original + assert not MediaCache.has_inline_images(json.loads(saved_metadata)) + assert MediaCache.has_inline_images(metadata) + + +@pytest.mark.asyncio +async def test_cas_failure_does_not_write_inline_payload_to_database(media_service): + service, objects, _ = media_service + service.ap.storage_mgr.storage_provider.save.side_effect = OSError('Disk full') + message_id = await _record_message(service, _context(WORKSPACE_A), json.dumps([image()])) + async with service.ap.persistence_mgr.get_db_engine().connect() as conn: + stored = ( + await conn.execute(sa.select(MonitoringMessage.message_content).where(MonitoringMessage.id == message_id)) + ).scalar_one() + assert json.loads(stored)[0]['base64'] is None + assert not objects + + +@pytest.mark.asyncio +@pytest.mark.parametrize('field', ['data', 'metadata']) +async def test_direct_ledger_writes_cannot_bypass_image_persistence(media_service, field): + service, objects, _ = media_service + engine = service.ap.persistence_mgr.get_db_engine() + with pytest.raises(sa.exc.StatementError, match='Inline images cannot be persisted'): + await RunLedgerStore(engine).append_event( + run_id='run', sequence=1, event_type='tool.call.completed', **{field: {'image': image()}} + ) + async with engine.connect() as conn: + assert (await conn.execute(sa.select(sa.func.count()).select_from(AgentRunEvent))).scalar_one() == 0 + assert not objects + + +@pytest.mark.asyncio +@pytest.mark.parametrize('write_kind', ['create', 'update']) +async def test_monitoring_variables_follow_metadata_only_policy(media_service, write_kind): + service, objects, _ = media_service + variables = json.dumps({'preview': image(), 'count': 0}) + context = _context(WORKSPACE_A) + if write_kind == 'create': + message_id = await service.record_message( + context, + bot_id='same-bot', + bot_name='Bot', + pipeline_id='pipeline', + pipeline_name='Pipeline', + message_content='message', + session_id='session', + variables=variables, + ) + else: + message_id = await _record_message(service, context, 'message') + await service.update_message_status(context, message_id, 'success', variables=variables) + async with service.ap.persistence_mgr.get_db_engine().connect() as conn: + stored = ( + await conn.execute(sa.select(MonitoringMessage.variables).where(MonitoringMessage.id == message_id)) + ).scalar_one() + assert json.loads(stored) == {'preview': {'type': 'Image', 'base64': None}, 'count': 0} + assert not objects + + +@pytest.mark.asyncio +@pytest.mark.parametrize('role', ['user', 'assistant']) +async def test_transcript_remains_metadata_only_without_creating_cas_objects(media_service, role): + service, objects, _ = media_service + contents = [image(), {'type': 'image_base64', 'image_base64': 'eA=='}] + original = copy.deepcopy(contents) + event = SimpleNamespace( + event_id='event', + conversation_id='conversation', + bot_id='same-bot', + workspace_id=WORKSPACE_A, + thread_id=None, + actor=None, + input=SimpleNamespace(text='image', contents=contents, attachments=[]), + ) + journal = AgentRunJournal(service.ap) + if role == 'user': + await journal.write_user_transcript(event, 'event') + else: + await journal.write_assistant_transcript( + result_dict={'data': {'message': {'role': 'assistant', 'content': contents}}}, + event=event, + run_id='run', + runner_id='runner', + ) + async with service.ap.persistence_mgr.get_db_engine().connect() as conn: + stored = (await conn.execute(sa.select(Transcript.content_json))).scalar_one() + assert not MediaCache.has_inline_images(json.loads(stored)) + assert not objects + assert contents == original diff --git a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx index 70d226487..bba19cb48 100644 --- a/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx +++ b/web/src/app/home/monitoring/components/executions/ExecutionDetailSheet.tsx @@ -87,10 +87,14 @@ export default function ExecutionDetailSheet({ const [detail, setDetail] = useState(null); const [loading, setLoading] = useState(false); const [error, setError] = useState(false); - const [sectionError, setSectionError] = useState( - null, + const [sectionError, setSectionError] = useState< + Partial> + >({}); + const [busy, setBusy] = useState>>( + {}, ); - const [busy, setBusy] = useState(null); + const pendingSections = useRef(new Set()); + const expandedSections = useRef(new Set()); const [refresh, setRefresh] = useState(0); const request = useRef(0); const identity = `${workspace}:${row?.source}:${row?.id}`; @@ -106,20 +110,43 @@ export default function ExecutionDetailSheet({ if (requestedIdentity.current !== identity) { setDetail(null); requestedIdentity.current = identity; + expandedSections.current.clear(); } setError(false); - setSectionError(null); - setBusy(null); + setSectionError({}); + setBusy({}); + pendingSections.current.clear(); if (!id || !source) { setLoading(false); return; } setLoading(true); - backendClient - .getExecutionDetail(source, id) - .then((result) => { + const expanded = [...expandedSections.current].filter( + (section) => section !== 'errors', + ); + Promise.all([ + backendClient.getExecutionDetail(source, id, { section: 'overview' }), + ...expanded.map((section) => + backendClient + .getExecutionDetail(source, id, { section }) + .catch(() => null), + ), + ]) + .then(([result, ...sections]) => { if (version !== request.current) return; - setDetail(result); + if (!result) return; + const pages = { ...result.pages }; + const failures: Partial> = {}; + sections.forEach((loaded, index) => { + const section = expanded[index]; + if (loaded) Object.assign(pages, loaded.pages); + else failures[section] = true; + }); + setDetail((previous) => ({ + ...result, + pages: { ...previous?.pages, ...pages }, + })); + setSectionError((previous) => ({ ...previous, ...failures })); setLoadedIdentity(identity); }) .catch(() => { @@ -144,10 +171,11 @@ export default function ExecutionDetailSheet({ return () => clearInterval(timer); }, [status, visible]); const loadMore = async (section: ExecutionSection) => { - if (!visible || !row || busy) return; + if (!visible || !row || pendingSections.current.has(section)) return; const version = request.current; - setBusy(section); - setSectionError(null); + pendingSections.current.add(section); + setBusy((previous) => ({ ...previous, [section]: true })); + setSectionError((previous) => ({ ...previous, [section]: false })); try { const result = await backendClient.getExecutionDetail( row.source, @@ -174,9 +202,13 @@ export default function ExecutionDetailSheet({ }, ); } catch { - if (version === request.current) setSectionError(section); + if (version === request.current) + setSectionError((previous) => ({ ...previous, [section]: true })); } finally { - if (version === request.current) setBusy(null); + if (version === request.current) { + pendingSections.current.delete(section); + setBusy((previous) => ({ ...previous, [section]: false })); + } } }; const renderItem = (section: ExecutionSection, item: ExecutionDetailItem) => { @@ -298,9 +330,12 @@ export default function ExecutionDetailSheet({ }; const renderSection = (section: ExecutionSection, primary = false) => { const page = visible?.pages[section]; + if (primary && !page) return null; if ( - !page || - (!primary && page.items.length === 0 && section !== 'conversation') + !primary && + page?.items.length === 0 && + section !== 'conversation' && + !sectionError[section] ) return null; const body = ( @@ -310,26 +345,29 @@ export default function ExecutionDetailSheet({ {t('monitoring.unified.contextHint')}

)} - {page.items.length === 0 && ( + {!page && !sectionError[section] && ( + + )} + {page?.items.length === 0 && (

{t('monitoring.unified.noRecords')}

)} - {page.items.map((item, i) => ( + {page?.items.map((item, i) => (
{renderItem(section, item)}
))} - {sectionError === section && ( + {sectionError[section] && ( loadMore(section)} /> )} - {page.has_more && ( + {page?.has_more && (