diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 8a6444da0b..fb38e27c48 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -958,6 +958,11 @@ def cache_image_from_bytes(data: bytes, ext: str = ".jpg") -> str: return str(filepath) +async def cache_image_from_bytes_async(data: bytes, ext: str = ".jpg") -> str: + """Cache image bytes without blocking the caller's event loop.""" + return await asyncio.to_thread(cache_image_from_bytes, data, ext) + + async def cache_image_from_url(url: str, ext: str = ".jpg", retries: int = 2) -> str: """ Download an image from a URL and save it to the local cache. @@ -1002,7 +1007,7 @@ async def cache_image_from_url(url: str, ext: str = ".jpg", retries: int = 2) -> content = await _read_httpx_body_with_limit( response, media_type="image", ) - return cache_image_from_bytes(content, ext) + return await cache_image_from_bytes_async(content, ext) except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 429: raise @@ -1100,6 +1105,11 @@ def cache_audio_from_bytes(data: bytes, ext: str = ".ogg") -> str: return str(filepath) +async def cache_audio_from_bytes_async(data: bytes, ext: str = ".ogg") -> str: + """Cache audio bytes without blocking the caller's event loop.""" + return await asyncio.to_thread(cache_audio_from_bytes, data, ext) + + async def cache_audio_from_url(url: str, ext: str = ".ogg", retries: int = 2) -> str: """ Download an audio file from a URL and save it to the local cache. @@ -1144,7 +1154,7 @@ async def cache_audio_from_url(url: str, ext: str = ".ogg", retries: int = 2) -> content = await _read_httpx_body_with_limit( response, media_type="audio", ) - return cache_audio_from_bytes(content, ext) + return await cache_audio_from_bytes_async(content, ext) except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 429: raise @@ -1207,6 +1217,11 @@ def cache_video_from_bytes(data: bytes, ext: str = ".mp4") -> str: return str(filepath) +async def cache_video_from_bytes_async(data: bytes, ext: str = ".mp4") -> str: + """Cache video bytes without blocking the caller's event loop.""" + return await asyncio.to_thread(cache_video_from_bytes, data, ext) + + def cleanup_video_cache(max_age_hours: int = 24) -> int: """ Delete cached videos older than *max_age_hours*. @@ -2332,6 +2347,11 @@ def cache_document_from_bytes(data: bytes, filename: str) -> str: return str(filepath) +async def cache_document_from_bytes_async(data: bytes, filename: str) -> str: + """Cache document bytes without blocking the caller's event loop.""" + return await asyncio.to_thread(cache_document_from_bytes, data, filename) + + def cleanup_document_cache(max_age_hours: int = 24) -> int: """ Delete cached documents older than *max_age_hours*. @@ -2452,6 +2472,23 @@ def cache_media_bytes( return CachedMedia(to_agent_visible_cache_path(path), out_mime, "document", display or fallback_name) +async def cache_media_bytes_async( + data: bytes, + *, + filename: str = "", + mime_type: str = "", + default_kind: Optional[str] = None, +) -> Optional[CachedMedia]: + """Classify and cache attachment bytes without blocking the event loop.""" + return await asyncio.to_thread( + cache_media_bytes, + data, + filename=filename, + mime_type=mime_type, + default_kind=default_kind, + ) + + class MessageType(Enum): """Types of incoming messages.""" TEXT = "text" diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index 6306c92b9e..49d1170b0c 100644 --- a/gateway/platforms/bluebubbles.py +++ b/gateway/platforms/bluebubbles.py @@ -28,9 +28,9 @@ from gateway.platforms.base import ( MessageEvent, MessageType, SendResult, - cache_image_from_bytes, - cache_audio_from_bytes, - cache_document_from_bytes, + cache_image_from_bytes_async, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, ) from .media_cache import ext_for_mime from gateway.platforms.helpers import compile_mention_patterns, strip_markdown @@ -848,7 +848,7 @@ class BlueBubblesAdapter(BasePlatformAdapter): use_mimetypes=False, fallback=".jpg", ) or ".jpg" - return cache_image_from_bytes(data, ext) + return await cache_image_from_bytes_async(data, ext) if mime.startswith("audio/"): ext = ext_for_mime( @@ -860,11 +860,11 @@ class BlueBubblesAdapter(BasePlatformAdapter): use_mimetypes=False, fallback=".mp3", ) or ".mp3" - return cache_audio_from_bytes(data, ext) + return await cache_audio_from_bytes_async(data, ext) # Videos, documents, and everything else filename = transfer_name or f"file_{uuid.uuid4().hex[:8]}" - return cache_document_from_bytes(data, filename) + return await cache_document_from_bytes_async(data, filename) except Exception as exc: logger.warning( diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index b8a9470817..bebf86b0c2 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -68,8 +68,8 @@ from gateway.platforms.base import ( MessageType, SendResult, _ssrf_redirect_guard, - cache_document_from_bytes, - cache_image_from_bytes, + cache_document_from_bytes_async, + cache_image_from_bytes_async, ) from gateway.platforms.helpers import strip_markdown from gateway.platforms.media_cache import ext_for_mime @@ -1830,7 +1830,7 @@ class QQAdapter(BasePlatformAdapter): use_mimetypes=True, fallback=".jpg", ) or ".jpg" - return cache_image_from_bytes(data, ext) + return await cache_image_from_bytes_async(data, ext) elif content_type == "voice" or content_type.startswith("audio/"): # QQ voice messages are typically .amr or .silk format. # Convert to .wav using ffmpeg so STT engines can process it. @@ -1841,7 +1841,7 @@ class QQAdapter(BasePlatformAdapter): or Path(urlparse(url).path).name or "qq_attachment" ) - return cache_document_from_bytes(data, filename) + return await cache_document_from_bytes_async(data, filename) @staticmethod def _is_voice_content_type(content_type: str, filename: str) -> bool: @@ -2341,9 +2341,9 @@ class QQAdapter(BasePlatformAdapter): source_url[:60], ext, ) - return cache_document_from_bytes(audio_data, f"qq_voice{ext}") + return await cache_document_from_bytes_async(audio_data, f"qq_voice{ext}") except Exception: - return cache_document_from_bytes(audio_data, f"qq_voice{ext}") + return await cache_document_from_bytes_async(audio_data, f"qq_voice{ext}") finally: try: os.unlink(src_path) @@ -2354,7 +2354,7 @@ class QQAdapter(BasePlatformAdapter): try: wav_data = Path(wav_path).read_bytes() os.unlink(wav_path) - return cache_document_from_bytes(wav_data, "qq_voice.wav") + return await cache_document_from_bytes_async(wav_data, "qq_voice.wav") except Exception as exc: logger.debug("[%s] Failed to read converted wav: %s", self._log_tag, exc) return None diff --git a/gateway/platforms/signal.py b/gateway/platforms/signal.py index 4e46f2b2b2..b41c3aeac7 100644 --- a/gateway/platforms/signal.py +++ b/gateway/platforms/signal.py @@ -37,9 +37,9 @@ from gateway.platforms.base import ( MessageType, ProcessingOutcome, SendResult, - cache_image_from_bytes, - cache_audio_from_bytes, - cache_document_from_bytes, + cache_image_from_bytes_async, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, cache_image_from_url, utf16_len, ) @@ -942,11 +942,11 @@ class SignalAdapter(BasePlatformAdapter): raw_data, ext = remuxed if _is_image_ext(ext): - path = cache_image_from_bytes(raw_data, ext) + path = await cache_image_from_bytes_async(raw_data, ext) elif _is_audio_ext(ext): - path = cache_audio_from_bytes(raw_data, ext) + path = await cache_audio_from_bytes_async(raw_data, ext) else: - path = cache_document_from_bytes(raw_data, ext) + path = await cache_document_from_bytes_async(raw_data, ext) return path, ext diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index ccf610fc7a..73b8dcad55 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -63,9 +63,9 @@ from gateway.platforms.base import ( MessageEvent, MessageType, SendResult, - cache_audio_from_bytes, - cache_document_from_bytes, - cache_image_from_bytes, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, + cache_image_from_bytes_async, ) from hermes_constants import get_hermes_home from utils import atomic_json_write @@ -1673,7 +1673,7 @@ class WeixinAdapter(BasePlatformAdapter): full_url=media.get("full_url"), timeout_seconds=30.0, ) - return cache_image_from_bytes(data, ".jpg") + return await cache_image_from_bytes_async(data, ".jpg") except Exception as exc: logger.warning("[%s] image download failed: %s", self.name, exc) return None @@ -1689,7 +1689,7 @@ class WeixinAdapter(BasePlatformAdapter): full_url=media.get("full_url"), timeout_seconds=120.0, ) - return cache_document_from_bytes(data, "video.mp4") + return await cache_document_from_bytes_async(data, "video.mp4") except Exception as exc: logger.warning("[%s] video download failed: %s", self.name, exc) return None @@ -1708,7 +1708,7 @@ class WeixinAdapter(BasePlatformAdapter): full_url=media.get("full_url"), timeout_seconds=60.0, ) - return cache_document_from_bytes(data, filename), mime + return await cache_document_from_bytes_async(data, filename), mime except Exception as exc: logger.warning("[%s] file download failed: %s", self.name, exc) return None, mime @@ -1732,7 +1732,7 @@ class WeixinAdapter(BasePlatformAdapter): full_url=media.get("full_url"), timeout_seconds=60.0, ) - return cache_audio_from_bytes(data, ".silk") + return await cache_audio_from_bytes_async(data, ".silk") except Exception as exc: logger.warning("[%s] voice download failed: %s", self.name, exc) return None diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py index d36c177794..4ec49d50f8 100644 --- a/gateway/platforms/yuanbao.py +++ b/gateway/platforms/yuanbao.py @@ -56,9 +56,9 @@ from gateway.platforms.base import ( MessageEvent, MessageType, SendResult, - cache_document_from_bytes, - cache_image_from_bytes, - cache_video_from_bytes, + cache_document_from_bytes_async, + cache_image_from_bytes_async, + cache_video_from_bytes_async, ) from gateway.platforms import helpers as _mdchunk from gateway.platforms.helpers import MessageDeduplicator @@ -2522,7 +2522,7 @@ class MediaResolveMiddleware(InboundMiddleware): if kind == "image": ext = cls._guess_image_ext_from_url(fetch_url) try: - local_path = cache_image_from_bytes(file_bytes, ext=ext) + local_path = await cache_image_from_bytes_async(file_bytes, ext=ext) except ValueError as exc: logger.warning( "[%s] inbound image cache rejected: %s err=%s", @@ -2537,7 +2537,7 @@ class MediaResolveMiddleware(InboundMiddleware): if kind == "video": # Yuanbao video resources carry no reliable extension; default to mp4. - local_path = cache_video_from_bytes(file_bytes) + local_path = await cache_video_from_bytes_async(file_bytes) mime = guess_mime_type(local_path) or ( content_type if content_type.startswith("video/") else "video/mp4" ) @@ -2549,7 +2549,7 @@ class MediaResolveMiddleware(InboundMiddleware): parsed = urllib.parse.urlparse(fetch_url) file_name = os.path.basename(parsed.path) or "file" try: - local_path = cache_document_from_bytes(file_bytes, file_name) + local_path = await cache_document_from_bytes_async(file_bytes, file_name) except Exception as exc: logger.warning( "[%s] inbound file cache failed: %s err=%s", diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 1839d57dd0..2b53778d01 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -203,10 +203,10 @@ from gateway.platforms.base import ( ProcessingOutcome, SendResult, cache_image_from_url, - cache_image_from_bytes, + cache_image_from_bytes_async, cache_audio_from_url, - cache_audio_from_bytes, - cache_document_from_bytes, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, SUPPORTED_DOCUMENT_TYPES, _TEXT_INJECT_EXTENSIONS, _prefix_within_utf16_limit, @@ -8108,7 +8108,7 @@ class DiscordAdapter(BasePlatformAdapter): raw_bytes = await self._read_attachment_bytes(att, media_type="image") if raw_bytes is not None: try: - return cache_image_from_bytes(raw_bytes, ext=ext) + return await cache_image_from_bytes_async(raw_bytes, ext=ext) except Exception as e: logger.debug( "[Discord] cache_image_from_bytes rejected att.read() data; falling back to URL: %s", @@ -8127,7 +8127,7 @@ class DiscordAdapter(BasePlatformAdapter): raw_bytes = await self._read_attachment_bytes(att, media_type="audio") if raw_bytes is not None: try: - return cache_audio_from_bytes(raw_bytes, ext=ext) + return await cache_audio_from_bytes_async(raw_bytes, ext=ext) except Exception as e: logger.debug( "[Discord] cache_audio_from_bytes failed; falling back to URL: %s", @@ -8454,7 +8454,7 @@ class DiscordAdapter(BasePlatformAdapter): else: try: raw_bytes = await self._cache_discord_document(att, ext) - cached_path = cache_document_from_bytes( + cached_path = await cache_document_from_bytes_async( raw_bytes, att.filename or f"document{ext or '.bin'}" ) if in_allowlist: diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py index d0546f6339..0c7bb89039 100644 --- a/plugins/platforms/feishu/adapter.py +++ b/plugins/platforms/feishu/adapter.py @@ -126,10 +126,10 @@ from gateway.platforms.base import ( ProcessingOutcome, SendResult, SUPPORTED_DOCUMENT_TYPES, - cache_document_from_bytes, + cache_document_from_bytes_async, cache_image_from_url, - cache_audio_from_bytes, - cache_image_from_bytes, + cache_audio_from_bytes_async, + cache_image_from_bytes_async, ) from gateway.status import acquire_scoped_lock, release_scoped_lock from hermes_constants import get_hermes_home @@ -3656,7 +3656,7 @@ class FeishuAdapter(BasePlatformAdapter): default_name=preferred_name, default_ext=default_ext, ) - cached_path = cache_document_from_bytes(body, filename) + cached_path = await cache_document_from_bytes_async(body, filename) return cached_path, filename @staticmethod @@ -4107,7 +4107,7 @@ class FeishuAdapter(BasePlatformAdapter): content_type = self._get_response_header(response, "Content-Type") filename = getattr(response, "file_name", None) or f"{image_key}.jpg" ext = self._guess_extension(filename, content_type, ".jpg", allowed=_IMAGE_EXTENSIONS) - cached_path = cache_image_from_bytes(raw_bytes, ext=ext) + cached_path = await cache_image_from_bytes_async(raw_bytes, ext=ext) media_type = self._normalize_media_type(content_type, default=self._default_image_media_type(ext)) return cached_path, media_type except Exception: @@ -4161,26 +4161,26 @@ class FeishuAdapter(BasePlatformAdapter): if media_type.startswith("image/"): ext = self._guess_extension(filename, content_type, ".jpg", allowed=_IMAGE_EXTENSIONS) - cached_path = cache_image_from_bytes(raw_bytes, ext=ext) + cached_path = await cache_image_from_bytes_async(raw_bytes, ext=ext) logger.info("[Feishu] Cached message image resource at %s", cached_path) return cached_path, media_type or self._default_image_media_type(ext) if request_type == "audio" or media_type.startswith("audio/"): ext = self._guess_extension(filename, content_type, ".ogg", allowed=_AUDIO_EXTENSIONS) - cached_path = cache_audio_from_bytes(raw_bytes, ext=ext) + cached_path = await cache_audio_from_bytes_async(raw_bytes, ext=ext) logger.info("[Feishu] Cached message audio resource at %s", cached_path) return cached_path, (media_type or f"audio/{ext.lstrip('.') or 'ogg'}") if media_type.startswith("video/"): if not Path(filename).suffix: filename = f"{filename}.mp4" - cached_path = cache_document_from_bytes(raw_bytes, filename) + cached_path = await cache_document_from_bytes_async(raw_bytes, filename) logger.info("[Feishu] Cached message video resource at %s", cached_path) return cached_path, media_type if not Path(filename).suffix and media_type in _DOCUMENT_MIME_TO_EXT: filename = f"{filename}{_DOCUMENT_MIME_TO_EXT[media_type]}" - cached_path = cache_document_from_bytes(raw_bytes, filename) + cached_path = await cache_document_from_bytes_async(raw_bytes, filename) logger.info("[Feishu] Cached message document resource at %s", cached_path) return cached_path, (media_type or self._guess_document_media_type(filename)) except Exception: diff --git a/plugins/platforms/google_chat/adapter.py b/plugins/platforms/google_chat/adapter.py index 7be34ceee2..693068d60c 100644 --- a/plugins/platforms/google_chat/adapter.py +++ b/plugins/platforms/google_chat/adapter.py @@ -229,10 +229,10 @@ from gateway.platforms.base import ( MessageType, ProcessingOutcome, SendResult, - cache_audio_from_bytes, - cache_document_from_bytes, - cache_image_from_bytes, - cache_video_from_bytes, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, + cache_image_from_bytes_async, + cache_video_from_bytes_async, ) @@ -2115,13 +2115,13 @@ class GoogleChatAdapter(BasePlatformAdapter): else: ext = "" if mime.startswith("image/"): - local = cache_image_from_bytes(data, ext=ext or ".jpg") + local = await cache_image_from_bytes_async(data, ext=ext or ".jpg") elif mime.startswith("audio/"): - local = cache_audio_from_bytes(data, ext=ext or ".ogg") + local = await cache_audio_from_bytes_async(data, ext=ext or ".ogg") elif mime.startswith("video/"): - local = cache_video_from_bytes(data, ext=ext or ".mp4") + local = await cache_video_from_bytes_async(data, ext=ext or ".mp4") else: - local = cache_document_from_bytes(data, filename) + local = await cache_document_from_bytes_async(data, filename) return local, mime # ------------------------------------------------------------------ diff --git a/plugins/platforms/line/adapter.py b/plugins/platforms/line/adapter.py index 1150556b4e..e253e87840 100644 --- a/plugins/platforms/line/adapter.py +++ b/plugins/platforms/line/adapter.py @@ -118,10 +118,10 @@ from gateway.platforms.base import ( MessageEvent, MessageType, SendResult, - cache_audio_from_bytes, - cache_document_from_bytes, - cache_image_from_bytes, - cache_video_from_bytes, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, + cache_image_from_bytes_async, + cache_video_from_bytes_async, ) from gateway.config import Platform @@ -1154,16 +1154,16 @@ class LineAdapter(BasePlatformAdapter): }.get(msg_type, ".bin") try: if msg_type == "image": - return cache_image_from_bytes(data, ext=ext), "image/jpeg" + return await cache_image_from_bytes_async(data, ext=ext), "image/jpeg" if msg_type == "audio": media_type = mimetypes.guess_type(f"audio{ext}")[0] or "audio/mp4" - return cache_audio_from_bytes(data, ext=ext), media_type + return await cache_audio_from_bytes_async(data, ext=ext), media_type if msg_type == "video": media_type = mimetypes.guess_type(f"video{ext}")[0] or "video/mp4" - return cache_video_from_bytes(data, ext=ext), media_type + return await cache_video_from_bytes_async(data, ext=ext), media_type document_name = filename or f"line_file{ext}" return ( - cache_document_from_bytes(data, document_name), + await cache_document_from_bytes_async(data, document_name), mimetypes.guess_type(document_name)[0] or "application/octet-stream", ) except Exception as exc: diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 6a2eb362ff..a09f715c90 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -3702,9 +3702,9 @@ class MatrixAdapter(BasePlatformAdapter): if file_bytes is not None: from gateway.platforms.base import ( - cache_audio_from_bytes, - cache_document_from_bytes, - cache_image_from_bytes, + cache_audio_from_bytes_async, + cache_document_from_bytes_async, + cache_image_from_bytes_async, ) if msg_type == MessageType.PHOTO: @@ -3715,7 +3715,7 @@ class MatrixAdapter(BasePlatformAdapter): "image/webp": ".webp", } ext = ext_map.get(media_type, ".jpg") - cached_path = cache_image_from_bytes(file_bytes, ext=ext) + cached_path = await cache_image_from_bytes_async(file_bytes, ext=ext) logger.info("[Matrix] Cached user image at %s", cached_path) elif msg_type in {MessageType.AUDIO, MessageType.VOICE}: ext = ( @@ -3727,14 +3727,14 @@ class MatrixAdapter(BasePlatformAdapter): ).suffix or ".ogg" ) - cached_path = cache_audio_from_bytes(file_bytes, ext=ext) + cached_path = await cache_audio_from_bytes_async(file_bytes, ext=ext) else: filename = body or ( "video.mp4" if msg_type == MessageType.VIDEO else "document" ) - cached_path = cache_document_from_bytes( + cached_path = await cache_document_from_bytes_async( file_bytes, filename ) except Exception as e: diff --git a/plugins/platforms/mattermost/adapter.py b/plugins/platforms/mattermost/adapter.py index 6f5172bb01..45797f0268 100644 --- a/plugins/platforms/mattermost/adapter.py +++ b/plugins/platforms/mattermost/adapter.py @@ -966,18 +966,18 @@ class MattermostAdapter(BasePlatformAdapter): ) as resp: if resp.status < 400: file_data = await resp.read() - from gateway.platforms.base import cache_image_from_bytes, cache_document_from_bytes + from gateway.platforms.base import cache_image_from_bytes_async, cache_document_from_bytes_async if mime.startswith("image/"): - local_path = cache_image_from_bytes(file_data, ext or ".png") + local_path = await cache_image_from_bytes_async(file_data, ext or ".png") media_urls.append(local_path) media_types.append(mime) elif mime.startswith("audio/"): - from gateway.platforms.base import cache_audio_from_bytes - local_path = cache_audio_from_bytes(file_data, ext or ".ogg") + from gateway.platforms.base import cache_audio_from_bytes_async + local_path = await cache_audio_from_bytes_async(file_data, ext or ".ogg") media_urls.append(local_path) media_types.append(mime) else: - local_path = cache_document_from_bytes(file_data, fname) + local_path = await cache_document_from_bytes_async(file_data, fname) media_urls.append(local_path) media_types.append(mime) else: diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index b4cd21c143..32ea3c2aa6 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -56,8 +56,8 @@ from gateway.platforms.base import ( resolve_proxy_url, safe_url_for_log, _ssrf_redirect_guard, - cache_document_from_bytes, - cache_video_from_bytes, + cache_document_from_bytes_async, + cache_video_from_bytes_async, ) try: # sibling module; support both package and flat plugin-dir import @@ -6931,7 +6931,7 @@ class SlackAdapter(BasePlatformAdapter): raw_bytes = await self._download_slack_file_bytes( url, team_id=team_id ) - cached_path = cache_video_from_bytes(raw_bytes, ext=ext) + cached_path = await cache_video_from_bytes_async(raw_bytes, ext=ext) media_urls.append(cached_path) media_types.append( SUPPORTED_VIDEO_TYPES.get(ext, mimetype or "video/mp4") @@ -6985,7 +6985,7 @@ class SlackAdapter(BasePlatformAdapter): raw_bytes = await self._download_slack_file_bytes( url, team_id=team_id ) - cached_path = cache_document_from_bytes( + cached_path = await cache_document_from_bytes_async( raw_bytes, original_filename or f"document{ext or '.bin'}" ) if in_allowlist: @@ -9004,13 +9004,13 @@ class SlackAdapter(BasePlatformAdapter): ) if audio: - from gateway.platforms.base import cache_audio_from_bytes + from gateway.platforms.base import cache_audio_from_bytes_async - return cache_audio_from_bytes(response.content, ext) + return await cache_audio_from_bytes_async(response.content, ext) else: - from gateway.platforms.base import cache_image_from_bytes + from gateway.platforms.base import cache_image_from_bytes_async - return cache_image_from_bytes(response.content, ext) + return await cache_image_from_bytes_async(response.content, ext) except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: if ( isinstance(exc, httpx.HTTPStatusError) diff --git a/plugins/platforms/teams/adapter.py b/plugins/platforms/teams/adapter.py index 172d89d946..b0182381ff 100644 --- a/plugins/platforms/teams/adapter.py +++ b/plugins/platforms/teams/adapter.py @@ -109,7 +109,7 @@ from gateway.platforms.base import ( MessageType, SendResult, cache_image_from_url, - cache_media_bytes, + cache_media_bytes_async, ) from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError @@ -1096,7 +1096,7 @@ class TeamsAdapter(BasePlatformAdapter): filename = att_name or (f"document.{file_type}" if file_type else "document") try: data = await self._fetch_attachment_bytes(download_url) - cached = cache_media_bytes(data, filename=filename, mime_type="") + cached = await cache_media_bytes_async(data, filename=filename, mime_type="") if cached: media_urls.append(cached.path) media_types.append(cached.media_type) @@ -1145,7 +1145,7 @@ class TeamsAdapter(BasePlatformAdapter): # Direct-URL non-image attachment (video/audio/document). try: data = await self._fetch_attachment_bytes(content_url) - cached = cache_media_bytes( + cached = await cache_media_bytes_async( data, filename=att_name, mime_type=content_type ) if cached: diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 15d2541d30..a2ba4e7b51 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -223,9 +223,9 @@ from gateway.platforms.base import ( ProcessingOutcome, SendResult, classify_send_error, - cache_image_from_bytes, - cache_audio_from_bytes, - cache_video_from_bytes, + cache_image_from_bytes_async, + cache_audio_from_bytes_async, + cache_video_from_bytes_async, cache_document_from_bytes, resolve_proxy_url, SUPPORTED_VIDEO_TYPES, @@ -9623,7 +9623,7 @@ class TelegramAdapter(BasePlatformAdapter): ``_max_doc_bytes`` limit as the addressed document path. Oversized or unsupported attachments are noted in the transcript without downloading. """ - from gateway.platforms.base import cache_media_bytes + from gateway.platforms.base import cache_media_bytes_async source, filename, mime, kind = self._observed_media_source(msg) if source is None: @@ -9649,7 +9649,7 @@ class TelegramAdapter(BasePlatformAdapter): data = bytes(await file_obj.download_as_bytearray()) if not filename: filename = os.path.basename(getattr(file_obj, "file_path", "") or "") - cached = cache_media_bytes(data, filename=filename, mime_type=mime, default_kind=kind) + cached = await cache_media_bytes_async(data, filename=filename, mime_type=mime, default_kind=kind) except Exception as exc: logger.warning("[Telegram] Failed to cache observed group media: %s", _redact_telegram_error_text(exc), exc_info=True) return @@ -9676,7 +9676,7 @@ class TelegramAdapter(BasePlatformAdapter): async def _cache_replied_media(self, msg: Any, event: MessageEvent) -> None: """Cache media from the message this turn replies to, if any.""" - from gateway.platforms.base import cache_media_bytes + from gateway.platforms.base import cache_media_bytes_async reply_msg = getattr(msg, "reply_to_message", None) if reply_msg is None: @@ -9699,7 +9699,7 @@ class TelegramAdapter(BasePlatformAdapter): data = bytes(await file_obj.download_as_bytearray()) if not filename: filename = os.path.basename(getattr(file_obj, "file_path", "") or "") - cached = cache_media_bytes(data, filename=filename, mime_type=mime, default_kind=kind) + cached = await cache_media_bytes_async(data, filename=filename, mime_type=mime, default_kind=kind) except Exception as exc: logger.warning("[Telegram] Failed to cache replied-to media: %s", _redact_telegram_error_text(exc), exc_info=True) return @@ -10317,7 +10317,7 @@ class TelegramAdapter(BasePlatformAdapter): ext = candidate break # Save to local cache (for vision tool access) - cached_path = cache_image_from_bytes(bytes(image_bytes), ext=ext) + cached_path = await cache_image_from_bytes_async(bytes(image_bytes), ext=ext) event.media_urls = [cached_path] event.media_types = [f"image/{ext.lstrip('.')}" ] logger.info("[Telegram] Cached user photo at %s", cached_path) @@ -10344,7 +10344,7 @@ class TelegramAdapter(BasePlatformAdapter): return file_obj = await msg.voice.get_file() audio_bytes = await file_obj.download_as_bytearray() - cached_path = cache_audio_from_bytes(bytes(audio_bytes), ext=".ogg") + cached_path = await cache_audio_from_bytes_async(bytes(audio_bytes), ext=".ogg") event.media_urls = [cached_path] event.media_types = ["audio/ogg"] logger.info("[Telegram] Cached user voice at %s", cached_path) @@ -10361,7 +10361,7 @@ class TelegramAdapter(BasePlatformAdapter): return file_obj = await msg.audio.get_file() audio_bytes = await file_obj.download_as_bytearray() - cached_path = cache_audio_from_bytes(bytes(audio_bytes), ext=".mp3") + cached_path = await cache_audio_from_bytes_async(bytes(audio_bytes), ext=".mp3") event.media_urls = [cached_path] event.media_types = ["audio/mp3"] logger.info("[Telegram] Cached user audio at %s", cached_path) @@ -10385,7 +10385,7 @@ class TelegramAdapter(BasePlatformAdapter): if file_obj.file_path.lower().endswith(candidate): ext = candidate break - cached_path = cache_video_from_bytes(bytes(video_bytes), ext=ext) + cached_path = await cache_video_from_bytes_async(bytes(video_bytes), ext=ext) event.media_urls = [cached_path] event.media_types = [SUPPORTED_VIDEO_TYPES.get(ext, "video/mp4")] logger.info("[Telegram] Cached user video at %s", cached_path) @@ -10435,7 +10435,7 @@ class TelegramAdapter(BasePlatformAdapter): image_bytes = await file_obj.download_as_bytearray() image_ext = ext if ext in _TELEGRAM_IMAGE_EXTENSIONS else _TELEGRAM_IMAGE_MIME_TO_EXT.get(doc_mime, ".jpg") try: - cached_path = cache_image_from_bytes(bytes(image_bytes), ext=image_ext) + cached_path = await cache_image_from_bytes_async(bytes(image_bytes), ext=image_ext) except ValueError as e: logger.warning("[Telegram] Failed to cache image document: %s", _redact_telegram_error_text(e), exc_info=True) event.text = ( @@ -10473,7 +10473,7 @@ class TelegramAdapter(BasePlatformAdapter): if ext in SUPPORTED_VIDEO_TYPES: file_obj = await doc.get_file() video_bytes = await file_obj.download_as_bytearray() - cached_path = cache_video_from_bytes(bytes(video_bytes), ext=ext) + cached_path = await cache_video_from_bytes_async(bytes(video_bytes), ext=ext) event.media_urls = [cached_path] event.media_types = [SUPPORTED_VIDEO_TYPES[ext]] event.message_type = MessageType.VIDEO @@ -10494,9 +10494,9 @@ class TelegramAdapter(BasePlatformAdapter): file_obj = await doc.get_file() doc_bytes = await file_obj.download_as_bytearray() raw_bytes = bytes(doc_bytes) - from gateway.platforms.base import cache_media_bytes + from gateway.platforms.base import cache_media_bytes_async - cached = cache_media_bytes( + cached = await cache_media_bytes_async( raw_bytes, filename=original_filename or f"document{ext or '.bin'}", mime_type=doc_mime, @@ -10645,7 +10645,7 @@ class TelegramAdapter(BasePlatformAdapter): try: file_obj = await sticker.get_file() image_bytes = await file_obj.download_as_bytearray() - cached_path = cache_image_from_bytes(bytes(image_bytes), ext=".webp") + cached_path = await cache_image_from_bytes_async(bytes(image_bytes), ext=".webp") logger.info("[Telegram] Analyzing sticker at %s", cached_path) from tools.vision_tools import vision_analyze_tool diff --git a/plugins/platforms/wecom/adapter.py b/plugins/platforms/wecom/adapter.py index 27fbf52e2b..71a6b843d3 100644 --- a/plugins/platforms/wecom/adapter.py +++ b/plugins/platforms/wecom/adapter.py @@ -68,8 +68,8 @@ from gateway.platforms.base import ( MessageEvent, MessageType, SendResult, - cache_document_from_bytes, - cache_image_from_bytes, + cache_document_from_bytes_async, + cache_image_from_bytes_async, ) from utils import env_float @@ -1612,13 +1612,13 @@ class WeComAdapter(BasePlatformAdapter): if kind == "image": ext = self._detect_image_ext(raw) try: - return cache_image_from_bytes(raw, ext), self._mime_for_ext(ext, fallback="image/jpeg") + return await cache_image_from_bytes_async(raw, ext), self._mime_for_ext(ext, fallback="image/jpeg") except ValueError as exc: logger.warning("[%s] Rejected non-image bytes: %s", self.name, exc) return None filename = str(media.get("filename") or media.get("name") or "wecom_file") - return cache_document_from_bytes(raw, filename), mimetypes.guess_type(filename)[0] or "application/octet-stream" + return await cache_document_from_bytes_async(raw, filename), mimetypes.guess_type(filename)[0] or "application/octet-stream" url = str(media.get("url") or "").strip() if not url: @@ -1642,13 +1642,13 @@ class WeComAdapter(BasePlatformAdapter): if kind == "image": ext = self._guess_extension(url, content_type, fallback=self._detect_image_ext(raw)) try: - return cache_image_from_bytes(raw, ext), content_type or self._mime_for_ext(ext, fallback="image/jpeg") + return await cache_image_from_bytes_async(raw, ext), content_type or self._mime_for_ext(ext, fallback="image/jpeg") except ValueError as exc: logger.warning("[%s] Rejected non-image bytes from %s: %s", self.name, url, exc) return None filename = self._guess_filename(url, headers.get("content-disposition"), content_type) - return cache_document_from_bytes(raw, filename), content_type + return await cache_document_from_bytes_async(raw, filename), content_type @staticmethod def _decode_base64(data: str) -> bytes: diff --git a/tests/gateway/test_async_media_cache.py b/tests/gateway/test_async_media_cache.py new file mode 100644 index 0000000000..e28b8e096e --- /dev/null +++ b/tests/gateway/test_async_media_cache.py @@ -0,0 +1,78 @@ +import asyncio +import threading +from pathlib import Path + +import pytest + +import gateway.platforms.base as base + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("async_name", "sync_name", "args"), + [ + ("cache_image_from_bytes_async", "cache_image_from_bytes", (b"data", ".png")), + ("cache_audio_from_bytes_async", "cache_audio_from_bytes", (b"data", ".ogg")), + ("cache_video_from_bytes_async", "cache_video_from_bytes", (b"data", ".mp4")), + ( + "cache_document_from_bytes_async", + "cache_document_from_bytes", + (b"data", "report.pdf"), + ), + ], +) +async def test_async_cache_wrappers_keep_event_loop_responsive( + monkeypatch, async_name, sync_name, args +): + loop_thread = threading.get_ident() + cache_started = threading.Event() + release_cache = threading.Event() + observed = {} + + def blocking_cache(*call_args): + observed["thread"] = threading.get_ident() + observed["args"] = call_args + cache_started.set() + observed["ticker_ran_during_cache"] = release_cache.wait(timeout=1) + return "cached" + + monkeypatch.setattr(base, sync_name, blocking_cache) + + async def ticker(): + while not cache_started.is_set(): + await asyncio.sleep(0) + release_cache.set() + + ticker_task = asyncio.create_task(ticker()) + result = await getattr(base, async_name)(*args) + await ticker_task + + assert result == "cached" + assert observed["args"] == args + assert observed["thread"] != loop_thread + assert observed["ticker_ran_during_cache"] is True + + +@pytest.mark.asyncio +async def test_async_cache_wrapper_propagates_validation_errors(monkeypatch): + def reject_image(data, ext): + raise ValueError("invalid image") + + monkeypatch.setattr(base, "cache_image_from_bytes", reject_image) + + with pytest.raises(ValueError, match="invalid image"): + await base.cache_image_from_bytes_async(b"not-an-image", ".png") + + +@pytest.mark.asyncio +async def test_async_cache_wrapper_uses_active_profile_home(monkeypatch, tmp_path): + profile_home = tmp_path / "profile" + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + + cached = await base.cache_image_from_bytes_async( + b"\x89PNG\r\n\x1a\nminimal", ".png" + ) + + cached_path = Path(cached) + assert cached_path.parent == profile_home / "cache" / "images" + assert cached_path.read_bytes() == b"\x89PNG\r\n\x1a\nminimal" diff --git a/tests/gateway/test_bluebubbles.py b/tests/gateway/test_bluebubbles.py index 695a26297d..7d95190cd0 100644 --- a/tests/gateway/test_bluebubbles.py +++ b/tests/gateway/test_bluebubbles.py @@ -259,13 +259,13 @@ class TestBlueBubblesAttachmentDownload: cached_path = None - def mock_cache_image(data, ext): + async def mock_cache_image(data, ext): nonlocal cached_path cached_path = f"/tmp/test_image{ext}" return cached_path monkeypatch.setattr( - "gateway.platforms.bluebubbles.cache_image_from_bytes", + "gateway.platforms.bluebubbles.cache_image_from_bytes_async", mock_cache_image, ) diff --git a/tests/gateway/test_discord_attachment_download.py b/tests/gateway/test_discord_attachment_download.py index a97632aa15..5b9575ee91 100644 --- a/tests/gateway/test_discord_attachment_download.py +++ b/tests/gateway/test_discord_attachment_download.py @@ -132,8 +132,8 @@ class TestCacheDiscordImage: att = _make_attachment_with_read(b"forbidden") with patch( - "plugins.platforms.discord.adapter.cache_image_from_bytes", - side_effect=ValueError("not a valid image"), + "plugins.platforms.discord.adapter.cache_image_from_bytes_async", + new=AsyncMock(side_effect=ValueError("not a valid image")), ), patch( "plugins.platforms.discord.adapter.cache_image_from_url", new_callable=AsyncMock, @@ -156,8 +156,8 @@ class TestCacheDiscordAudio: att = _make_attachment_with_read(_OGG_BYTES) with patch( - "plugins.platforms.discord.adapter.cache_audio_from_bytes", - return_value="/tmp/voice.ogg", + "plugins.platforms.discord.adapter.cache_audio_from_bytes_async", + new=AsyncMock(return_value="/tmp/voice.ogg"), ) as mock_bytes, patch( "plugins.platforms.discord.adapter.cache_audio_from_url", new_callable=AsyncMock, @@ -165,7 +165,7 @@ class TestCacheDiscordAudio: result = await adapter._cache_discord_audio(att, ".ogg") assert result == "/tmp/voice.ogg" - mock_bytes.assert_called_once_with(_OGG_BYTES, ext=".ogg") + mock_bytes.assert_awaited_once_with(_OGG_BYTES, ext=".ogg") mock_url.assert_not_called() @@ -215,8 +215,8 @@ class TestHandleMessageUsesAuthenticatedRead: adapter.handle_message = AsyncMock() with patch( - "plugins.platforms.discord.adapter.cache_image_from_bytes", - return_value="/tmp/img_from_read.png", + "plugins.platforms.discord.adapter.cache_image_from_bytes_async", + new=AsyncMock(return_value="/tmp/img_from_read.png"), ), patch( "plugins.platforms.discord.adapter.cache_image_from_url", new_callable=AsyncMock, diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index a923ea5f5d..d25c5ad574 100644 --- a/tests/gateway/test_feishu.py +++ b/tests/gateway/test_feishu.py @@ -1105,8 +1105,8 @@ class TestAdapterBehavior(unittest.TestCase): side_effect=lambda **_kwargs: _FakeAsyncClient(), ): with patch( - "plugins.platforms.feishu.adapter.cache_document_from_bytes", - return_value="/tmp/cached-doc.bin", + "plugins.platforms.feishu.adapter.cache_document_from_bytes_async", + new=AsyncMock(return_value="/tmp/cached-doc.bin"), ): return await adapter._download_remote_document( "https://example.com/doc.bin", diff --git a/tests/gateway/test_google_chat.py b/tests/gateway/test_google_chat.py index d3a05ea00c..c9f1f1584f 100644 --- a/tests/gateway/test_google_chat.py +++ b/tests/gateway/test_google_chat.py @@ -1336,9 +1336,9 @@ class TestAttachmentSSRFGuard: monkeypatch.setattr(asyncio, "to_thread", _fake_to_thread) from plugins.platforms.google_chat import adapter as gc_mod monkeypatch.setattr( - gc_mod, "cache_document_from_bytes", - lambda data, ext=None, filename=None: str(tmp_path / "out.pdf"), - raising=False, + gc_mod, + "cache_document_from_bytes_async", + AsyncMock(return_value=str(tmp_path / "out.pdf")), ) path, mime = await adapter._download_attachment(attachment) diff --git a/tests/gateway/test_line_plugin.py b/tests/gateway/test_line_plugin.py index e59bd8286e..5a6386a7be 100644 --- a/tests/gateway/test_line_plugin.py +++ b/tests/gateway/test_line_plugin.py @@ -205,10 +205,14 @@ class TestInboundMedia: return adapter.handle_message.await_args.args[0] def test_image_message_uses_photo_type_and_image_mime(self, adapter): - with patch.object(_line, "cache_image_from_bytes", return_value="/cache/image.jpg") as cache: + with patch.object( + _line, + "cache_image_from_bytes_async", + new=AsyncMock(return_value="/cache/image.jpg"), + ) as cache: asyncio.run(adapter._handle_message_event(self._event("image"))) - cache.assert_called_once_with(b"line-bytes", ext=".jpg") + cache.assert_awaited_once_with(b"line-bytes", ext=".jpg") event = self._captured_event(adapter) assert event.message_type is _line.MessageType.PHOTO assert event.media_urls == ["/cache/image.jpg"] diff --git a/tests/gateway/test_signal.py b/tests/gateway/test_signal.py index 078d787d43..fb668dc34d 100644 --- a/tests/gateway/test_signal.py +++ b/tests/gateway/test_signal.py @@ -270,7 +270,10 @@ class TestSignalAttachmentFetch: adapter._rpc, captured = _stub_rpc({"data": b64_data}) - with patch("gateway.platforms.signal.cache_image_from_bytes", return_value="/tmp/test.png"): + with patch( + "gateway.platforms.signal.cache_image_from_bytes_async", + new=AsyncMock(return_value="/tmp/test.png"), + ): await adapter._fetch_attachment("attachment-123") call = captured[0] @@ -1327,7 +1330,10 @@ class TestSignalContentlessEnvelope: b64_data = base64.b64encode(png_data).decode() adapter._rpc, _ = _stub_rpc({"data": b64_data}) - with patch("gateway.platforms.signal.cache_image_from_bytes", return_value="/tmp/img.png"): + with patch( + "gateway.platforms.signal.cache_image_from_bytes_async", + new=AsyncMock(return_value="/tmp/img.png"), + ): await adapter._handle_envelope({ "envelope": { "sourceNumber": "+155****9999", diff --git a/tests/gateway/test_telegram_documents.py b/tests/gateway/test_telegram_documents.py index a82003ce9d..fd3b406a08 100644 --- a/tests/gateway/test_telegram_documents.py +++ b/tests/gateway/test_telegram_documents.py @@ -308,7 +308,10 @@ class TestMediaGroups: msg1 = _make_message(caption="two images", photo=[first_photo]) msg2 = _make_message(photo=[second_photo]) - with patch("plugins.platforms.telegram.adapter.cache_image_from_bytes", side_effect=["/tmp/burst-one.jpg", "/tmp/burst-two.jpg"]): + with patch( + "plugins.platforms.telegram.adapter.cache_image_from_bytes_async", + new=AsyncMock(side_effect=["/tmp/burst-one.jpg", "/tmp/burst-two.jpg"]), + ): await adapter._handle_media_message(_make_update(msg1), MagicMock()) await adapter._handle_media_message(_make_update(msg2), MagicMock()) assert adapter.handle_message.await_count == 0 diff --git a/tests/gateway/test_weixin.py b/tests/gateway/test_weixin.py index ed20aeb8f3..ed9c5e5804 100644 --- a/tests/gateway/test_weixin.py +++ b/tests/gateway/test_weixin.py @@ -685,8 +685,8 @@ class TestWeixinVoiceAlwaysDownloaded: adapter._poll_session = Mock() fake_audio_bytes = b"\\x00\\x01\\x02FAKE_SILK" - monkeypatch.setattr(weixin, "cache_audio_from_bytes", - lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}")) + monkeypatch.setattr(weixin, "cache_audio_from_bytes_async", + AsyncMock(side_effect=lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}"))) async def _fake_download(session, *, cdn_base_url, encrypted_query_param, aes_key_b64, full_url, timeout_seconds): @@ -739,8 +739,8 @@ class TestWeixinVoiceAlwaysDownloaded: adapter._cdn_base_url = "https://example.invalid" adapter._poll_session = Mock() - monkeypatch.setattr(weixin, "cache_audio_from_bytes", - lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}")) + monkeypatch.setattr(weixin, "cache_audio_from_bytes_async", + AsyncMock(side_effect=lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}"))) async def _fake_download(session, *, cdn_base_url, encrypted_query_param, aes_key_b64, full_url, timeout_seconds): @@ -803,8 +803,8 @@ class TestWeixinVoiceGatewayHandoff: adapter._token = None adapter._cdn_base_url = "https://example.invalid" - monkeypatch.setattr(weixin, "cache_audio_from_bytes", - lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}")) + monkeypatch.setattr(weixin, "cache_audio_from_bytes_async", + AsyncMock(side_effect=lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}"))) async def _fake_download(*a, **k): return b"\x00\x01FAKE_SILK" monkeypatch.setattr(weixin, "_download_and_decrypt_media", _fake_download)