Merge pull request #2706 from langbot-app/fix/monitoring-and-message-handling

perf: bound monitoring workloads and enforce reference-only image persistence
This commit is contained in:
彼方
2026-10-18 13:26:43 +08:00
committed by GitHub
46 changed files with 2904 additions and 163 deletions
+1
View File
@@ -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: `<type>(<scope>): <subject>`.
+3
View File
@@ -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.
+3
View File
@@ -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.
@@ -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',
+12 -4
View File
@@ -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,
@@ -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)
@@ -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,
@@ -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]:
+96 -38
View File
@@ -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,
@@ -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(
+1 -1
View File
@@ -313,7 +313,7 @@ class LangBotMCPServer:
)
@mcp.tool(
description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Each section is independently paginated; follow pages[section].next_offset when has_more is true.'
description='Read execution inputs, generated outputs, delivery records, calls, errors, related executions and conversation history. Use section=overview for inputs, outputs, deliveries and errors without loading history; request other sections on demand. Omit section for the full trace. Each section is independently paginated; follow pages[section].next_offset when has_more is true.'
)
async def get_monitoring_execution_detail(
source: typing.Literal['auto', 'agent', 'pipeline', 'event'],
+1 -1
View File
@@ -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
@@ -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'),
)
@@ -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'),)
@@ -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',
),
)
@@ -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'),
@@ -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)
+46
View File
@@ -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
+4 -2
View File
@@ -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')
+5 -11
View File
@@ -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:
@@ -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']
+1 -4
View File
@@ -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)
+8 -4
View File
@@ -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,
)
@@ -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*(?:(?!<think>).)*?</think>\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()
@@ -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
+99 -27
View File
@@ -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)
+2 -1
View File
@@ -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
@@ -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))
@@ -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
@@ -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)
@@ -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})
@@ -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(
@@ -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'
@@ -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
@@ -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)
@@ -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()
@@ -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()
@@ -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)),), {})
+144
View File
@@ -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
@@ -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()
@@ -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()
+66 -2
View File
@@ -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)
@@ -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
@@ -87,10 +87,14 @@ export default function ExecutionDetailSheet({
const [detail, setDetail] = useState<ExecutionDetail | null>(null);
const [loading, setLoading] = useState(false);
const [error, setError] = useState(false);
const [sectionError, setSectionError] = useState<ExecutionSection | null>(
null,
const [sectionError, setSectionError] = useState<
Partial<Record<ExecutionSection, boolean>>
>({});
const [busy, setBusy] = useState<Partial<Record<ExecutionSection, boolean>>>(
{},
);
const [busy, setBusy] = useState<ExecutionSection | null>(null);
const pendingSections = useRef(new Set<ExecutionSection>());
const expandedSections = useRef(new Set<ExecutionSection>());
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<Record<ExecutionSection, boolean>> = {};
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')}
</p>
)}
{page.items.length === 0 && (
{!page && !sectionError[section] && (
<Skeleton className="h-16 w-full" />
)}
{page?.items.length === 0 && (
<p className="text-sm text-muted-foreground">
{t('monitoring.unified.noRecords')}
</p>
)}
{page.items.map((item, i) => (
{page?.items.map((item, i) => (
<div key={`${item.id}-${i}`}>{renderItem(section, item)}</div>
))}
{sectionError === section && (
{sectionError[section] && (
<LoadErrorState
compact
title={t('monitoring.execution.detail.loadError')}
onRetry={() => loadMore(section)}
/>
)}
{page.has_more && (
{page?.has_more && (
<Button
variant="outline"
size="sm"
disabled={busy !== null}
disabled={!!busy[section]}
onClick={() => loadMore(section)}
>
{t('monitoring.unified.loadMore')}
@@ -337,7 +375,7 @@ export default function ExecutionDetailSheet({
)}
</div>
);
const title = `${t(`monitoring.unified.${section}`)} (${page.items.length}${page.has_more ? '+' : ''})`;
const title = `${t(`monitoring.unified.${section}`)}${page ? ` (${page.items.length}${page.has_more ? '+' : ''})` : ''}`;
return primary ? (
<section key={section} className="space-y-3">
<h3 className="flex items-center gap-2 text-sm font-semibold">
@@ -352,6 +390,12 @@ export default function ExecutionDetailSheet({
<details
key={section}
open={section === 'errors' ? true : undefined}
onToggle={(event) => {
if (event.currentTarget.open) {
expandedSections.current.add(section);
if (!page) void loadMore(section);
} else expandedSections.current.delete(section);
}}
className={`rounded-lg border p-3 ${section === 'errors' ? 'border-red-500/25 bg-red-500/5' : ''}`}
>
<summary className="cursor-pointer text-sm font-medium">
+14
View File
@@ -52,6 +52,7 @@ function SelectContent({
className,
children,
position = 'popper',
onCloseAutoFocus,
...props
}: React.ComponentProps<typeof SelectPrimitive.Content>) {
return (
@@ -66,6 +67,19 @@ function SelectContent({
)}
position={position}
{...props}
onCloseAutoFocus={(event) => {
onCloseAutoFocus?.(event);
if (event.defaultPrevented) return;
const content = event.target as HTMLElement;
const activeElement = content.ownerDocument.activeElement;
// Exit animations may finish after another control has taken focus.
if (
activeElement &&
activeElement !== content.ownerDocument.body &&
!content.contains(activeElement)
)
event.preventDefault();
}}
>
<SelectScrollUpButton />
<SelectPrimitive.Viewport
+201
View File
@@ -24,6 +24,178 @@ const pageOf = (
next_offset = items.length,
) => ({ items, has_more, next_offset });
test('loads collapsed history on demand, retries and pages without reloading it on reopen', async ({
page,
}) => {
await installLangBotApiMocks(page, {
authenticated: true,
monitoringData: monitoring,
});
const requests: string[] = [];
let failHistory = true;
await page.route('**/api/v1/monitoring/executions/*/*?*', async (route) => {
const url = new URL(route.request().url());
const section = url.searchParams.get('section');
requests.push(url.search);
if (section === 'conversation' && failHistory) {
failHistory = false;
await route.fulfill({
status: 500,
json: { code: 500, msg: 'Temporary failure' },
});
return;
}
const offset = Number(url.searchParams.get('offset') || 0);
const pages =
section === 'overview'
? {
inputs: pageOf([{ id: 'input', content: 'Selected message' }]),
outputs: pageOf([]),
deliveries: pageOf([]),
errors: pageOf([]),
}
: section === 'conversation'
? {
conversation: pageOf(
[
{
id: `history-${offset}`,
content: offset ? 'Older history' : 'Recent history',
},
],
offset === 0,
offset + 1,
),
}
: { related: pageOf([]) };
await route.fulfill({
json: { code: 0, data: { source: 'pipeline', row, pages } },
});
});
await page.goto('/home/monitoring');
await page
.getByRole('button')
.filter({ hasText: 'Open this execution' })
.click();
const sheet = page.getByRole('dialog');
await expect(
sheet.getByText('Selected message').filter({ visible: true }),
).toBeVisible();
expect(requests).toEqual(['?section=overview']);
const history = sheet.locator('details').filter({
has: page.locator('summary', { hasText: /^Conversation context/ }),
});
await history.locator(':scope > summary').click();
await expect(history.getByRole('alert')).toBeVisible();
await history.getByRole('button', { name: 'Retry', exact: true }).click();
await expect(
history.getByText('Recent history').filter({ visible: true }),
).toBeVisible();
await history.getByRole('button', { name: 'Load more', exact: true }).click();
await expect(
history.getByText('Older history').filter({ visible: true }),
).toBeVisible();
await expect(
history.getByText('Recent history').filter({ visible: true }),
).toBeVisible();
const count = requests.length;
await history.locator(':scope > summary').click();
await history.locator(':scope > summary').click();
await expect(
history.getByText('Older history').filter({ visible: true }),
).toBeVisible();
expect(requests).toHaveLength(count);
expect(requests.every((query) => !query.includes('section=related'))).toBe(
true,
);
});
test('ignores a late history response after selecting a different execution', async ({
page,
}) => {
await installLangBotApiMocks(page, {
authenticated: true,
monitoringData: monitoring,
});
let releaseHistory: (() => void) | undefined;
const requested: string[] = [];
await page.route('**/api/v1/monitoring/executions/*/*?*', async (route) => {
const url = new URL(route.request().url());
const section = url.searchParams.get('section');
const sibling = url.pathname.endsWith('/sibling');
requested.push(`${sibling ? 'sibling' : 'original'}:${section}`);
if (section === 'conversation')
await new Promise<void>((resolve) => {
releaseHistory = resolve;
});
const pages =
section === 'overview'
? {
inputs: pageOf([
{
id: 'input',
content: sibling ? 'Sibling selection' : 'Original selection',
},
]),
outputs: pageOf([]),
deliveries: pageOf([]),
errors: pageOf([]),
}
: section === 'related'
? {
related: pageOf([
{ ...row, id: 'sibling', target_name: 'Processor B' },
]),
}
: {
conversation: pageOf([
{ id: 'late', content: 'Late old history' },
]),
};
await route.fulfill({
json: {
code: 0,
data: {
source: 'pipeline',
row: sibling ? { ...row, id: 'sibling' } : row,
pages,
},
},
});
});
await page.goto('/home/monitoring');
await page
.getByRole('button')
.filter({ hasText: 'Open this execution' })
.click();
const sheet = page.getByRole('dialog');
await expect(
sheet.getByText('Original selection').filter({ visible: true }),
).toBeVisible();
await sheet
.locator('summary')
.filter({ hasText: /^Conversation context/ })
.click();
await expect.poll(() => !!releaseHistory).toBe(true);
await sheet
.locator('summary')
.filter({ hasText: /^Related executions/ })
.click();
await sheet.getByRole('button', { name: /Processor B/ }).click();
await expect(
sheet.getByText('Sibling selection').filter({ visible: true }),
).toBeVisible();
const response = page.waitForResponse(
(res) => new URL(res.url()).searchParams.get('section') === 'conversation',
);
releaseHistory!();
await response;
await expect(sheet.getByText('Late old history')).toHaveCount(0);
expect(requested.filter((value) => value.startsWith('sibling:'))).toEqual([
'sibling:overview',
]);
});
test('renders typed and custom events without duplicate text or empty attachment messages', async ({
page,
}) => {
@@ -403,6 +575,35 @@ test('shares mode and status filters with dashboard charts and records', async (
}
});
test('restores filter trigger focus when closing with Escape', async ({
page,
}) => {
await installLangBotApiMocks(page, { authenticated: true });
await page.goto('/home/monitoring');
for (const id of ['bot', 'mode', 'status', 'time']) {
const trigger = page.locator(`#monitoring-filter-${id}`);
await trigger.click();
await expect(page.getByRole('listbox')).toBeVisible();
await page.keyboard.press('Escape');
await expect(trigger).toBeFocused();
await expect(page.getByRole('listbox')).toHaveCount(0);
}
});
test('keeps the custom range dialog focused when its select closes', async ({
page,
}) => {
await installLangBotApiMocks(page, { authenticated: true });
await page.goto('/home/monitoring');
await page.locator('#monitoring-filter-time').click();
await page
.getByRole('option', { name: 'Custom time range', exact: true })
.click();
const dialog = page.getByRole('dialog', { name: 'Custom time range' });
await expect(dialog).toBeVisible();
await expect(dialog.locator('input').first()).toBeFocused();
});
test('remembers all filters on reload and persists resetting them', async ({
page,
}) => {