feat(storage): media content-addressable cache and monitoring base64 externalization

- Add MediaCache using xxHash3-128 (with sha256 fallback) content-addressable storage
- Externalize message chain image payloads before recording monitoring and discarded messages
- Strip base64 payloads to null in SQLite monitoring_messages, dropping row size from megabytes to hundreds of bytes
- Add GET /api/v1/files/media/<filename> route with immutable HTTP cache headers to serve cached media
- Integrate age-based retention (default 30 days) and configurable disk quota with MaintenanceService cleanup loop
- Add defensive sanitizer in MonitoringService.record_message against oversized raw base64 payloads
- Add comprehensive unit tests and end-to-end verification covering CAS deduplication, route serving, and LRU pruning

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

@@ -47,6 +47,21 @@ class FilesRouterGroup(group.RouterGroup):
 
             return quart.Response(image_bytes, mimetype=mime_type)
 
+        @self.route(
+            '/media/<filename>',
+            methods=['GET'],
+            auth_type=group.AuthType.NONE,
+        )
+        async def get_media_file(filename: str) -> quart.Response:
+            media = await self.ap.storage_mgr.media_cache.get_media(filename)
+            if media is None:
+                return quart.Response('Media not found or expired', status=404)
+            media_bytes, mime_type = media
+            headers = {
+                'Cache-Control': 'public, max-age=2592000, immutable',
+            }
+            return quart.Response(media_bytes, mimetype=mime_type, headers=headers)
+
         @self.route(
             '/images',
             methods=['POST'],

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

@@ -80,6 +80,25 @@ class MaintenanceService:
             DEFAULT_LOG_RETENTION_DAYS,
             'storage.cleanup.log_retention_days',
         )
+        media_cfg = self.ap.instance_config.data.get('storage', {}).get('media_cache', {})
+        media_retention_days = self._positive_int(
+            media_cfg.get('retention_days'),
+            30,
+            'storage.media_cache.retention_days',
+        )
+        media_max_size_mb = self._non_negative_int(
+            media_cfg.get('max_size_mb'),
+            0,
+            'storage.media_cache.max_size_mb',
+        )
+        media_cleanup = (
+            await self.ap.storage_mgr.media_cache.cleanup(
+                media_retention_days,
+                media_max_size_mb,
+            )
+            if hasattr(self.ap.storage_mgr, 'media_cache') and await self._is_oss_singleton(context)
+            else {}
+        )
 
         return {
             'uploaded_files': await self._cleanup_expired_uploaded_files(context, upload_retention_days),
@@ -89,6 +108,7 @@ class MaintenanceService:
             )
             if await self._is_oss_singleton(context)
             else 0,
+            'media_files': media_cleanup.get('expired_deleted', 0) + media_cleanup.get('size_deleted', 0),
         }
 
     async def get_storage_analysis(self, context: TenantContext) -> dict[str, Any]:
@@ -466,6 +486,17 @@ class MaintenanceService:
             count += len(files)
         return count
 
+    def _non_negative_int(self, value: Any, default: int, name: str) -> int:
+        try:
+            parsed = int(value)
+        except (TypeError, ValueError):
+            self.ap.logger.warning(f'Invalid {name}: {value!r}, using {default}')
+            return default
+        if parsed < 0:
+            self.ap.logger.warning(f'{name} must be non-negative: {value!r}, using {default}')
+            return default
+        return parsed
+
     def _positive_int(self, value: Any, default: int, name: str) -> int:
         try:
             parsed = int(value)

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 import uuid
 import datetime
 import functools
@@ -417,6 +418,32 @@ class MonitoringService:
 
     # ========== Recording Methods ==========
 
+    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):
+            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)
+        except Exception:
+            return re.sub(
+                r'data:image/[a-zA-Z0-9.+_-]+;base64,[\sA-Za-z0-9+/=]{1000,}', '[base64 image omitted]', content
+            )
+
     @_workspace_transaction
     async def record_message(
         self,
@@ -439,6 +466,7 @@ class MonitoringService:
         """Record a message"""
         workspace_uuid = self._require_write_context(context)
         message_id = str(uuid.uuid4())
+        message_content = self._sanitize_message_content(message_content)
         message_data = {
             'id': message_id,
             'workspace_uuid': workspace_uuid,

src/langbot/pkg/pipeline/monitoring_helper.py

@@ -48,7 +48,10 @@ class MonitoringHelper:
             # Try to record message
             # Use JSON serialization to preserve message chain structure (including image URLs, etc.)
             if hasattr(query, 'message_chain') and hasattr(query.message_chain, 'model_dump'):
-                message_content = json.dumps(query.message_chain.model_dump(), ensure_ascii=False)
+                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)
             else:
                 message_content = str(query)
 
@@ -168,7 +171,10 @@ class MonitoringHelper:
                 if hasattr(last_resp, 'get_content_platform_message_chain'):
                     chain = last_resp.get_content_platform_message_chain()
                     if hasattr(chain, 'model_dump'):
-                        message_content = json.dumps(chain.model_dump(), ensure_ascii=False)
+                        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)
                     else:
                         message_content = str(chain)
                 else:

src/langbot/pkg/platform/botmgr.py

@@ -210,7 +210,10 @@ class RuntimeBot:
         """Record a discarded message in the monitoring system."""
         try:
             if hasattr(message_chain, 'model_dump'):
-                message_content = json.dumps(message_chain.model_dump(), ensure_ascii=False)
+                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)
             else:
                 message_content = str(message_chain)
 

src/langbot/pkg/storage/media.py

@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import copy
+import datetime
+import hashlib
+import mimetypes
+import os
+import re
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+try:
+    import xxhash
+except ImportError:
+    xxhash = None
+
+if TYPE_CHECKING:
+    from ...core import app
+    from . import mgr as storage_mgr
+
+DEFAULT_RETENTION_DAYS = 30
+DEFAULT_MAX_SIZE_MB = 0
+MEDIA_DIR = 'media_cache'
+SAFE_MEDIA_FILENAME = re.compile(r'^[a-f0-9]{32,64}(\.[a-zA-Z0-9]{1,10})?$')
+
+
+class MediaCache:
+    """Content-addressable storage cache for images and media attachments.
+
+    Deduplicates media files using xxHash3-128 (with sha256 fallback),
+    offloads payloads from SQLite to StorageProvider, and implements LRU
+    and age-based retention cleanup.
+    """
+
+    def __init__(self, ap: app.Application, storage_mgr: storage_mgr.StorageMgr):
+        self.ap = ap
+        self.storage_mgr = storage_mgr
+
+    @staticmethod
+    def hash_bytes(data: bytes) -> str:
+        """Compute content-addressable hash for binary data."""
+        if xxhash is not None:
+            return xxhash.xxh3_128_hexdigest(data)
+        return hashlib.sha256(data).hexdigest()[:32]
+
+    @staticmethod
+    def parse_data_url(data_url: str) -> tuple[bytes, str] | None:
+        """Parse a data URL or raw base64 string into bytes and mime type."""
+        if not data_url or not isinstance(data_url, str):
+            return None
+        try:
+            if data_url.startswith('data:'):
+                split_index = data_url.find(';base64,')
+                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'
+        except Exception:
+            return None
+        return None
+
+    @staticmethod
+    def guess_extension(mime_type: str | None, default: str = '.jpg') -> str:
+        """Guess appropriate file extension from MIME type."""
+        if not mime_type:
+            return default
+        mime_lower = mime_type.lower()
+        if 'png' in mime_lower:
+            return '.png'
+        if 'webp' in mime_lower:
+            return '.webp'
+        if 'gif' in mime_lower:
+            return '.gif'
+        if 'jpeg' in mime_lower or 'jpg' in mime_lower:
+            return '.jpg'
+        ext = mimetypes.guess_extension(mime_type)
+        if ext == '.jpe':
+            return '.jpg'
+        return ext or default
+
+    async def save_media(self, data: bytes, mime_type: str | None = None) -> tuple[str, str, int]:
+        """Save media bytes into content-addressable storage cache.
+
+        Returns:
+            Tuple of (hash_str, storage_key, byte_size)
+        """
+        hash_str = self.hash_bytes(data)
+        ext = self.guess_extension(mime_type)
+        storage_key = f'{MEDIA_DIR}/{hash_str}{ext}'
+        provider = self.storage_mgr.storage_provider
+
+        if not await provider.exists(storage_key):
+            await provider.save(storage_key, data)
+        else:
+            await self.touch(storage_key)
+
+        return hash_str, storage_key, len(data)
+
+    async def get_media(self, filename_or_key: str) -> tuple[bytes, str] | None:
+        """Retrieve media bytes and mime type by key or filename."""
+        filename = os.path.basename(filename_or_key)
+        if not SAFE_MEDIA_FILENAME.match(filename):
+            return None
+        storage_key = f'{MEDIA_DIR}/{filename}'
+        provider = self.storage_mgr.storage_provider
+
+        if not await provider.exists(storage_key):
+            return None
+
+        data = await self.storage_mgr._load_object_bounded(storage_key)
+        mime_type = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
+        await self.touch(storage_key)
+        return data, mime_type
+
+    async def touch(self, storage_key: str) -> None:
+        """Update access/modified time of a media file for LRU tracking."""
+        provider = getattr(self.storage_mgr, 'storage_provider', None)
+        if provider is not None and provider.__class__.__name__ == 'LocalStorageProvider':
+            full_path = os.path.join('data', 'storage', storage_key)
+            if os.path.exists(full_path):
+                now = datetime.datetime.now().timestamp()
+                try:
+                    await asyncio.to_thread(os.utime, full_path, (now, now))
+                except Exception:
+                    pass
+
+    async def cleanup(
+        self,
+        retention_days: int = DEFAULT_RETENTION_DAYS,
+        max_size_mb: int = DEFAULT_MAX_SIZE_MB,
+    ) -> dict[str, int]:
+        """Perform age-based and LRU size-based cleanup on media cache.
+
+        Args:
+            retention_days: Retain media accessed within this many days (default 30).
+            max_size_mb: Maximum total size in MB (0 means unlimited).
+
+        Returns:
+            Dictionary of cleanup metrics.
+        """
+        provider = getattr(self.storage_mgr, 'storage_provider', None)
+        if provider is None or provider.__class__.__name__ != 'LocalStorageProvider':
+            return {'expired_deleted': 0, 'size_deleted': 0, 'bytes_freed': 0}
+
+        target_dir = Path('data/storage') / MEDIA_DIR
+        if not target_dir.exists() or not target_dir.is_dir():
+            return {'expired_deleted': 0, 'size_deleted': 0, 'bytes_freed': 0}
+
+        now = datetime.datetime.now().timestamp()
+        cutoff = (now - retention_days * 86400) if retention_days > 0 else 0
+
+        expired_deleted = 0
+        size_deleted = 0
+        bytes_freed = 0
+        remaining: list[tuple[Path, int, float]] = []
+
+        for entry in target_dir.iterdir():
+            if not entry.is_file():
+                continue
+            try:
+                stat = entry.stat()
+            except OSError:
+                continue
+
+            if cutoff > 0 and stat.st_mtime < cutoff:
+                try:
+                    entry.unlink(missing_ok=True)
+                    expired_deleted += 1
+                    bytes_freed += stat.st_size
+                except OSError:
+                    pass
+            else:
+                remaining.append((entry, stat.st_size, stat.st_mtime))
+
+        if max_size_mb > 0:
+            max_bytes = max_size_mb * 1024 * 1024
+            total_bytes = sum(item[1] for item in remaining)
+            if total_bytes > max_bytes:
+                remaining.sort(key=lambda item: item[2])
+                for path, size, _ in remaining:
+                    if total_bytes <= max_bytes:
+                        break
+                    try:
+                        path.unlink(missing_ok=True)
+                        size_deleted += 1
+                        bytes_freed += size
+                        total_bytes -= size
+                    except OSError:
+                        pass
+
+        return {
+            'expired_deleted': expired_deleted,
+            'size_deleted': size_deleted,
+            'bytes_freed': bytes_freed,
+        }
+
+    async def externalize_chain_dump(self, chain_dump: Any) -> Any:
+        """Recursively extract raw base64 media into cache and replace with references."""
+        if isinstance(chain_dump, list):
+            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
+            for k, v in list(node.items()):
+                if isinstance(v, (list, dict)):
+                    node[k] = await self.externalize_chain_dump(v)
+            return node
+        return chain_dump

src/langbot/pkg/storage/mgr.py

@@ -34,6 +34,9 @@ class StorageMgr:
 
     def __init__(self, ap: app.Application):
         self.ap = ap
+        from . import media
+
+        self.media_cache = media.MediaCache(ap, self)
 
     def _object_read_limit(self) -> int:
         config = getattr(getattr(self.ap, 'instance_config', None), 'data', {})

src/langbot/templates/config.yaml

@@ -227,6 +227,12 @@ storage:
     # Bound every object materialized into Core memory. Built-in Local/S3
     # providers enforce this while reading (hard cap: 64 MiB).
     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_days: 30
+        # Maximum disk storage for cached media in MB (0 means unlimited, defaults to 0)
+        max_size_mb: 0
     cleanup:
         # Enable periodic cleanup of local/S3 uploaded files and old log files
         enabled: true