diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 9ad2a5a9b..e4d12fc51 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -273,6 +273,8 @@ Persistence is centered on `pkg/persistence/mgr.py`. Configuration starts from `src/langbot/templates/config.yaml` and is generated into `data/config.yaml` on first run. Most long-lived managers read from `ap.instance_config.data`. +Storage maintenance applies `storage.media_cache.retention_days` to both CAS files and historical inline images. Inline image expiry uses the owning record's creation time and preserves text, URLs, and record identity. It does not backfill CAS. `pkg/storage/inline_media_cleanup.py` processes one payload row at a time, with a per-table limit of 100 indexed candidates or 8 MiB of JSON characters per maintenance pass (one oversized row may exceed the latter limit). Run events lack a Workspace column, so their candidate cursor also advances past foreign or missing records; payloads are loaded only after checking Workspace ownership. Durable Workspace metadata cursors and timestamp/identity indexes keep completed history out of subsequent scans; newly written history must retain its actual creation timestamp. Updates and cursor advancement commit together. Invalid historical JSON is preserved. SQLite pages become reusable after payload removal; the database file is not automatically vacuumed by this task. + ## Frontend The frontend lives in `web/` and is a Vite SPA using React Router 7, shadcn/ui, Tailwind CSS, and pnpm. It is not Next.js, despite some historical filenames. diff --git a/src/langbot/pkg/api/http/service/maintenance.py b/src/langbot/pkg/api/http/service/maintenance.py index 4fe3cc642..86485efa7 100644 --- a/src/langbot/pkg/api/http/service/maintenance.py +++ b/src/langbot/pkg/api/http/service/maintenance.py @@ -114,6 +114,10 @@ class MaintenanceService: } if media_cache is not None and is_singleton: result['media_files'] = media_cleanup.get('expired_deleted', 0) + media_cleanup.get('size_deleted', 0) + if media_cache is not None: + from ....storage.inline_media_cleanup import cleanup_inline_media + + result['inline_media_records'] = await cleanup_inline_media(self.ap, context, media_retention_days) return result async def get_storage_analysis(self, context: TenantContext) -> dict[str, Any]: diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 97dce0da4..4b4d499ab 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -545,7 +545,7 @@ class Application: if total_deleted > 0: self.logger.info( f'Storage maintenance for Workspace {context.workspace_uuid}: ' - f'deleted expired files: {deleted}' + f'cleaned expired storage data: {deleted}' ) except asyncio.CancelledError: raise diff --git a/src/langbot/pkg/entity/persistence/agent_run.py b/src/langbot/pkg/entity/persistence/agent_run.py index af1f9f6b6..c68ee01f5 100644 --- a/src/langbot/pkg/entity/persistence/agent_run.py +++ b/src/langbot/pkg/entity/persistence/agent_run.py @@ -111,6 +111,7 @@ class AgentRun(Base): """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' ), @@ -202,4 +203,5 @@ class AgentRunEvent(Base): __table_args__ = ( sqlalchemy.UniqueConstraint('run_id', 'sequence', name='uq_agent_run_event_run_sequence'), sqlalchemy.Index('ix_agent_run_event_run_sequence', 'run_id', 'sequence'), + sqlalchemy.Index('ix_agent_run_event_media_ttl', 'created_at', 'id', 'run_id'), ) diff --git a/src/langbot/pkg/entity/persistence/event_log.py b/src/langbot/pkg/entity/persistence/event_log.py index bd730b03c..da16ee1b5 100644 --- a/src/langbot/pkg/entity/persistence/event_log.py +++ b/src/langbot/pkg/entity/persistence/event_log.py @@ -84,3 +84,5 @@ class EventLog(Base): metadata_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) """Additional metadata as JSON string.""" + + __table_args__ = (sqlalchemy.Index('ix_event_log_media_ttl', 'workspace_id', 'created_at', 'id'),) diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index b43ec6f63..5f9fdd257 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -39,6 +39,7 @@ 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' diff --git a/src/langbot/pkg/entity/persistence/transcript.py b/src/langbot/pkg/entity/persistence/transcript.py index f975ab245..f19c87d8a 100644 --- a/src/langbot/pkg/entity/persistence/transcript.py +++ b/src/langbot/pkg/entity/persistence/transcript.py @@ -74,6 +74,7 @@ class Transcript(Base): # Indexes __table_args__ = ( + sqlalchemy.Index('ix_transcript_media_ttl', 'workspace_id', 'created_at', 'id'), sqlalchemy.Index('ix_transcript_conversation_seq', 'conversation_id', 'seq'), sqlalchemy.Index('ix_transcript_conversation_created', 'conversation_id', 'created_at'), sqlalchemy.Index('ix_transcript_scope_seq', 'bot_id', 'workspace_id', 'conversation_id', 'thread_id', 'seq'), diff --git a/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py index 4d66427e4..7b4724e4f 100644 --- a/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py +++ b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py @@ -15,6 +15,8 @@ def upgrade(): ('agent_run', 'workspace_id'), ('monitoring_messages', 'workspace_uuid'), ): + if not inspector.has_table(table): + continue index = f'ix_{table}_workspace_status_lease' # Fresh installations create these indexes from ORM metadata before migrations. if index not in {item['name'] for item in inspector.get_indexes(table)}: @@ -24,6 +26,8 @@ def upgrade(): def downgrade(): inspector = sa.inspect(op.get_bind()) for table in ('agent_run', 'monitoring_messages'): + if not inspector.has_table(table): + continue index = f'ix_{table}_workspace_status_lease' if index in {item['name'] for item in inspector.get_indexes(table)}: op.drop_index(index, table_name=table) diff --git a/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py b/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py index 1b62980cb..85f37f855 100644 --- a/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py +++ b/src/langbot/pkg/persistence/alembic/versions/0040_conversation_page_index.py @@ -13,11 +13,13 @@ INDEX = 'ix_monitoring_messages_conversation_page' def upgrade(): + inspector = sa.inspect(op.get_bind()) # Fresh databases already receive this index from ORM metadata. - if INDEX not in {index['name'] for index in sa.inspect(op.get_bind()).get_indexes(TABLE)}: + if inspector.has_table(TABLE) and INDEX not in {index['name'] for index in inspector.get_indexes(TABLE)}: op.create_index(INDEX, TABLE, ['workspace_uuid', 'bot_id', 'session_id', 'timestamp', 'id']) def downgrade(): - if INDEX in {index['name'] for index in sa.inspect(op.get_bind()).get_indexes(TABLE)}: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(TABLE) and INDEX in {index['name'] for index in inspector.get_indexes(TABLE)}: op.drop_index(INDEX, table_name=TABLE) diff --git a/src/langbot/pkg/persistence/alembic/versions/0041_inline_media_ttl_indexes.py b/src/langbot/pkg/persistence/alembic/versions/0041_inline_media_ttl_indexes.py new file mode 100644 index 000000000..3820766ae --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0041_inline_media_ttl_indexes.py @@ -0,0 +1,37 @@ +"""Index incremental expiry of historical inline images.""" + +from alembic import op +import sqlalchemy as sa + +revision = '0041_inline_media_ttl_indexes' +down_revision = '0040_conversation_page_index' +branch_labels = None +depends_on = None + +INDEXES = ( + ('monitoring_messages', ['workspace_uuid', 'timestamp', 'id']), + ('event_log', ['workspace_id', 'created_at', 'id']), + ('agent_run', ['workspace_id', 'created_at', 'id']), + ('agent_run_event', ['created_at', 'id', 'run_id']), + ('transcript', ['workspace_id', 'created_at', 'id']), +) + + +def upgrade(): + inspector = sa.inspect(op.get_bind()) + for table, columns in INDEXES: + if not inspector.has_table(table): + continue + name = f'ix_{table}_media_ttl' + if name not in {index['name'] for index in inspector.get_indexes(table)}: + op.create_index(name, table, columns) + + +def downgrade(): + inspector = sa.inspect(op.get_bind()) + for table, _ in reversed(INDEXES): + if not inspector.has_table(table): + continue + name = f'ix_{table}_media_ttl' + if name in {index['name'] for index in inspector.get_indexes(table)}: + op.drop_index(name, table_name=table) diff --git a/src/langbot/pkg/storage/inline_media_cleanup.py b/src/langbot/pkg/storage/inline_media_cleanup.py new file mode 100644 index 000000000..3293fcca4 --- /dev/null +++ b/src/langbot/pkg/storage/inline_media_cleanup.py @@ -0,0 +1,147 @@ +"""Incrementally expire historical inline images without rewriting record history.""" + +from __future__ import annotations + +import asyncio +import datetime +import json +from typing import TYPE_CHECKING + +import sqlalchemy as sa + +from ..api.http.authz import WorkspaceRequiredError +from ..api.http.context import ExecutionContext +from ..entity.persistence.agent_run import AgentRun, AgentRunEvent +from ..entity.persistence.event_log import EventLog +from ..entity.persistence.metadata import WorkspaceMetadata +from ..entity.persistence.monitoring import MonitoringMessage +from ..entity.persistence.transcript import Transcript +from ..persistence.datetime_utils import as_naive_utc +from .media import MediaCache + +if TYPE_CHECKING: + from ..core.app import Application + + +ROWS_PER_TABLE = 100 +CHARACTERS_PER_TABLE = 8 * 1024 * 1024 +CURSOR_PREFIX = 'inline_media_ttl_v1:' + + +def _redact_columns(values: dict) -> dict: + updates = {} + for name, raw in values.items(): + if not raw or ('base64' not in raw and 'data:image/' not in raw): + continue + try: + payload = json.loads(raw) + except (ValueError, TypeError, RecursionError): + # Preserve malformed historical text rather than deleting an entire record. + continue + if MediaCache.has_inline_images(payload): + updates[name] = json.dumps(MediaCache.redact_inline_images(payload), ensure_ascii=False) + return updates + + +async def cleanup_inline_media(ap: Application, context: ExecutionContext, retention_days: int) -> int: + """Use the media TTL for immutable history, with durable per-Workspace cursors. + + Bound each table by both candidate count and serialized payload characters. One + oversized row is processed on its own so it cannot permanently block the cursor. + """ + if not isinstance(context, ExecutionContext) or not context.workspace_uuid: + raise WorkspaceRequiredError('Inline media cleanup requires an ExecutionContext') + if retention_days < 1: + raise ValueError('retention_days must be >= 1') + workspace = context.workspace_uuid + cutoff = as_naive_utc(datetime.datetime.now(datetime.timezone.utc)) - datetime.timedelta(days=retention_days) + sources = ( + ( + MonitoringMessage, + MonitoringMessage.timestamp, + MonitoringMessage.workspace_uuid == workspace, + ('message_content', 'variables'), + ), + (EventLog, EventLog.created_at, EventLog.workspace_id == workspace, ('input_json', 'metadata_json')), + (AgentRun, AgentRun.created_at, AgentRun.workspace_id == workspace, ('metadata_json',)), + ( + AgentRunEvent, + AgentRunEvent.created_at, + sa.exists( + sa.select(AgentRun.id).where( + AgentRun.run_id == AgentRunEvent.run_id, AgentRun.workspace_id == workspace + ) + ), + ('data_json', 'metadata_json'), + ), + ( + Transcript, + Transcript.created_at, + Transcript.workspace_id == workspace, + ('content_json', 'attachment_refs_json', 'metadata_json'), + ), + ) + manager = ap.persistence_mgr + cleaned = 0 + for model, timestamp, scope, fields in sources: + consumed = 0 + cursor_key = CURSOR_PREFIX + model.__tablename__ + cursor_filter = (WorkspaceMetadata.workspace_uuid == workspace, WorkspaceMetadata.key == cursor_key) + for _ in range(ROWS_PER_TABLE): + async with manager.tenant_uow(workspace): + cursor_result = await manager.execute_async( + sa.select(WorkspaceMetadata.value).where(*cursor_filter).with_for_update() + ) + cursor_raw = cursor_result.scalar_one_or_none() + # Events have no Workspace column. Limit indexed candidates before + # checking ownership so other Workspaces cannot cause unbounded scans. + query = sa.select(model.id, timestamp).where(timestamp < cutoff) + if model is not AgentRunEvent: + query = query.where(scope) + if cursor_raw is not None: + after_time, after_id = json.loads(cursor_raw) + query = query.where( + sa.tuple_(timestamp, model.id) + > sa.tuple_(datetime.datetime.fromisoformat(after_time), after_id) + ) + result = await manager.execute_async(query.order_by(timestamp, model.id).limit(1)) + row = result.first() + if row is None: + break + payload = ( + await manager.execute_async( + sa.select(*(getattr(model, field) for field in fields)).where(model.id == row[0], scope) + ) + ).first() + # Missing and foreign candidates still consume budget and advance the + # cursor. Their payloads are never returned to the cleanup worker. + values = dict(zip(fields, payload)) if payload is not None else {} + consumed += sum(len(value) for value in values.values() if value) + updates = await asyncio.to_thread(_redact_columns, values) if values else {} + if updates: + if model is AgentRun: + # Media expiry must not appear as new execution activity. + updates['updated_at'] = AgentRun.updated_at + changed = await manager.execute_async( + sa.update(model) + .where( + scope, + model.id == row[0], + *(getattr(model, name) == values[name] for name in fields if name in updates), + ) + .values(**updates) + ) + if not changed.rowcount: + # A concurrent writer changed this payload; retry it next time. + break + cleaned += 1 + cursor = json.dumps([row[1].isoformat(), row[0]]) + if cursor_raw is None: + await manager.execute_async( + sa.insert(WorkspaceMetadata).values(workspace_uuid=workspace, key=cursor_key, value=cursor) + ) + else: + await manager.execute_async(sa.update(WorkspaceMetadata).where(*cursor_filter).values(value=cursor)) + if consumed >= CHARACTERS_PER_TABLE: + break + return cleaned diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index de5092370..97650b8d8 100644 --- a/src/langbot/templates/config.yaml +++ b/src/langbot/templates/config.yaml @@ -224,7 +224,8 @@ storage: max_object_read_bytes: 10485760 # Media content cache (images & attachments externalized from monitoring and pipelines) media_cache: - # Retention period in days for cached media (defaults to 30 days) + # Retention for CAS files and historical inline images (defaults to 30 days). + # Inline images expire by record creation time; message text is retained. retention_days: 30 # Maximum disk storage for cached media in MB (0 means unlimited, defaults to 0) max_size_mb: 0 diff --git a/tests/unit_tests/api/service/test_maintenance_service.py b/tests/unit_tests/api/service/test_maintenance_service.py index 16f27aa38..3c12fabd6 100644 --- a/tests/unit_tests/api/service/test_maintenance_service.py +++ b/tests/unit_tests/api/service/test_maintenance_service.py @@ -396,9 +396,8 @@ class TestMaintenanceServiceGetStorageAnalysis: result = await service.get_storage_analysis(TEST_CONTEXT) assert result['total_size_bytes'] == 125 - assert ( - next(section for section in result['sections'] if section['key'] == 'skills')['path'] - == '/core/skills/tenants/test' + assert next(section for section in result['sections'] if section['key'] == 'skills')['path'] == str( + Path('/core/skills/tenants/test') ) assert result['processes'][1]['directories'][0]['key'] == 'artifacts' assert result['processes'][2]['directories'][1]['key'] == 'mcp' diff --git a/tests/unit_tests/storage/test_inline_media_cleanup.py b/tests/unit_tests/storage/test_inline_media_cleanup.py new file mode 100644 index 000000000..ad36088c8 --- /dev/null +++ b/tests/unit_tests/storage/test_inline_media_cleanup.py @@ -0,0 +1,359 @@ +"""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' + 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.0041_inline_media_ttl_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, columns in migration.INDEXES: + indexes = { + index['name']: index['column_names'] for index in sa.inspect(connection).get_indexes(table) + } + assert indexes[f'ix_{table}_media_ttl'] == columns + orm_index = next( + index for index in Base.metadata.tables[table].indexes if index.name == f'ix_{table}_media_ttl' + ) + 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() + + +@pytest.mark.parametrize( + 'revision', ['0039_execution_lease_indexes', '0040_conversation_page_index', '0041_inline_media_ttl_indexes'] +) +def test_index_migrations_allow_historical_schemas_without_monitoring_tables(revision): + migration = importlib.import_module(f'langbot.pkg.persistence.alembic.versions.{revision}') + 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()