ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

perf(gateway): offload inbound media cache writes

Browse Source
This commit is contained in:
fangliquanflq
2026-08-24 23:08:35 +08:00
committed by kshitij
parent a3d33fe22f
commit cb0b66c161
25 changed files with 262 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/platforms/base.py
+39 -2
View File
@@ -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"
gateway/platforms/bluebubbles.py
+6 -6
View File
@@ -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(
gateway/platforms/qqbot/adapter.py
+7 -7
View File
@@ -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
gateway/platforms/signal.py
+6 -6
View File
@@ -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
gateway/platforms/weixin.py
+7 -7
View File
@@ -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
gateway/platforms/yuanbao.py
+6 -6
View File
@@ -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",
plugins/platforms/discord/adapter.py
+6 -6
View File
@@ -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:
plugins/platforms/feishu/adapter.py
+9 -9
View File
@@ -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:
plugins/platforms/google_chat/adapter.py
+8 -8
View File
@@ -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
# ------------------------------------------------------------------
plugins/platforms/line/adapter.py
+8 -8
View File
@@ -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:
plugins/platforms/matrix/adapter.py
+6 -6
View File
@@ -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:
plugins/platforms/mattermost/adapter.py
+5 -5
View File
@@ -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:
plugins/platforms/slack/adapter.py
+8 -8
View File
@@ -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)
plugins/platforms/teams/adapter.py
+3 -3
View File
@@ -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:
plugins/platforms/telegram/adapter.py
+16 -16
View File
@@ -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
plugins/platforms/wecom/adapter.py
+6 -6
View File
@@ -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:
tests/gateway/test_async_media_cache.py
+78
View File
@@ -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"
tests/gateway/test_bluebubbles.py
+2 -2
View File
@@ -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,
)
tests/gateway/test_discord_attachment_download.py
+7 -7
View File
@@ -132,8 +132,8 @@ class TestCacheDiscordImage:
att = _make_attachment_with_read(b"<html>forbidden</html>")
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,
tests/gateway/test_feishu.py
+2 -2
View File
@@ -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",
tests/gateway/test_google_chat.py
+3 -3
View File
@@ -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)
tests/gateway/test_line_plugin.py
+6 -2
View File
@@ -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"]
tests/gateway/test_signal.py
+8 -2
View File
@@ -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",
tests/gateway/test_telegram_documents.py
+4 -1
View File
@@ -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
tests/gateway/test_weixin.py
+6 -6
View File
@@ -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)