mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-25 17:18:11 +00:00
feat(relay): Phase 2 media parity — send_media egress + inbound media localization (#71363)
- gateway/relay/media.py: RelayMediaClient for the connector's /relay/media plane (upload local files → re-host reference for send_media; download re-hosted inbound attachments → local temp paths). Same connector base URL the WS dials, same per-gateway signed bearer as the upgrade (auth.py) — no new configuration. stdlib urllib in a thread executor (no new deps); 25MB cap mirroring the connector's MEDIA_MAX_BYTES. - RelayAdapter: send_image / send_image_file / send_voice / send_video / send_document overrides route through ONE send_media op (media by reference: local paths upload first, public URLs pass through). Gated on supported_ops advertising send_media — legacy connectors keep today's text fallbacks; connector declines/failed uploads degrade the same way. Scope/user egress discriminators ride metadata exactly like send. - Inbound: _localize_inbound_media downloads each media_urls entry to a local temp path (native-adapter parity — vision/file tools consume paths); dead re-host refs are dropped, public URLs survive a missing client. Best-effort, never blocks handle_message. - docs/relay-connector-contract.md §4: send_media op row + media ingress/egress semantics (replaces the 'deferred to a later revision' note). Additive within contract_version 1. - tests: tests/gateway/relay/test_relay_media.py (15) — kind mapping, upload-first path handling, op gating (explicit + legacy-empty), decline/upload-failure fallbacks, scope metadata, inbound localization matrix, client URL derivation/credential gating. Stub connector grew a canned send_media result. Cross-repo pair: gateway-gateway 'Phase 2 media parity' PR (re-host plane + four ingress lanes + four platform send_media senders).
This commit is contained in:
Ben Barclay
GitHub
parent
6ad632bf9b
commit
689b51bef6
5 changed files with 804 additions and 2 deletions
|
|
@ -394,10 +394,41 @@ The gateway calls the transport with action dicts. Source of truth:
|
|||
| `edit` | `chat_id`, `message_id`, `content`, `metadata?` | `{success: bool, error?}` |
|
||||
| `typing` | `chat_id`, `content?`, `metadata?` | `{success: bool}` |
|
||||
| `follow_up` | `session_key`, `kind`, `content`, `metadata?` | `{success: bool, message_id?, error?}` |
|
||||
| `send_media` | `chat_id`, `media_kind`, `source_url`, `content?` (caption), `filename?`, `reply_to?`, `metadata?` | `{success: bool, message_id?, error?}` |
|
||||
|
||||
`get_chat_info(chat_id)` is a separate proxied call returning at least
|
||||
`{name, type}`. Media actions follow the same envelope shape (deferred to a
|
||||
later contract revision; additive).
|
||||
`{name, type}`.
|
||||
|
||||
**`send_media` (Phase 2 media egress).** Media crosses the wire BY REFERENCE:
|
||||
`source_url` is either (a) a **connector re-host** the gateway previously
|
||||
uploaded via `POST {connector}/relay/media` (raw bytes body, `Content-Type` +
|
||||
optional `X-Media-Filename` headers, per-gateway HMAC bearer — the same token
|
||||
scheme as the WS upgrade; response `{id, size}` → reference
|
||||
`{connector}/relay/media/{id}`), or (b) a **public http(s) URL** (e.g. a
|
||||
fal.media generation) the connector downloads directly. `media_kind` is one of
|
||||
`image` / `voice` / `audio` / `video` / `document` and selects the
|
||||
platform-native upload lane (Telegram `sendPhoto`/`sendVoice`/…, Discord
|
||||
multipart attachment, Slack external upload, WhatsApp media upload + media
|
||||
message). The caption rides `content` and renders through the platform's
|
||||
normal markdown lane; platforms without native captions get a follow-up text
|
||||
send (connector-side). Both routes and the op are gated on `supported_ops`
|
||||
advertising `send_media` — a legacy connector never sees the op (the gateway's
|
||||
media sends degrade to their pre-media text fallbacks). Size cap 25 MB
|
||||
(connector `mediaStore.ts` MEDIA_MAX_BYTES; uploads over it are rejected 413).
|
||||
|
||||
**Inbound media (Phase 2 media ingress).** An inbound event's `media_urls`
|
||||
carry fetchable references: platform-public URLs pass through (Discord CDN);
|
||||
auth-gated/expiring platform URLs (Telegram file API, Slack `url_private`,
|
||||
WhatsApp Graph media) are downloaded connector-side with the PLATFORM
|
||||
credential and re-hosted as `{connector}/relay/media/{id}` — the platform
|
||||
credential never crosses the wire. Re-host references are readable by any
|
||||
authenticated gateway (capability-URL semantics: the id is 128-bit random and
|
||||
was already delivered to every admitted recipient); the gateway downloads each
|
||||
reference with its per-gateway bearer and presents LOCAL file paths to the
|
||||
agent, mirroring native adapters. Re-hosts expire (TTL ~1h) — download on
|
||||
receipt, not lazily. A parallel `media` array (same order) adds `kind`, `mime`,
|
||||
`size`, `filename`, `caption` metadata; `message_type` reflects the first
|
||||
attachment's kind (`image`/`audio`/`document`).
|
||||
|
||||
**`typing` `content?` (Slack status clear).** A `typing` frame normally omits
|
||||
`content` — the connector renders its platform's active indicator ("is
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ from typing import Any, Callable, Dict, Optional
|
|||
from gateway.config import Platform, PlatformConfig
|
||||
from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult
|
||||
from gateway.relay.descriptor import CapabilityDescriptor
|
||||
from gateway.relay.media import RelayMediaClient
|
||||
from gateway.relay.transport import RelayTransport
|
||||
from gateway.session import SessionSource
|
||||
|
||||
|
|
@ -87,6 +88,13 @@ class RelayAdapter(BasePlatformAdapter):
|
|||
# non-retryable "relay disabled" fatal so the dashboard stops showing a
|
||||
# red "retrying" spin against a dead credential.
|
||||
self._revocation_monitor: Optional[asyncio.Task[None]] = None
|
||||
# Phase 2 media: the authenticated client for the connector's
|
||||
# /relay/media routes (upload for send_media source_url; download for
|
||||
# inbound re-hosted attachments → local paths the vision/file tools
|
||||
# consume). Built lazily from the relay dial URL + per-gateway creds;
|
||||
# None when either is absent (media lanes then degrade to the
|
||||
# pre-media text fallbacks).
|
||||
self._media_client: Optional["RelayMediaClient"] = None
|
||||
|
||||
# ── capability surface (from descriptor) ─────────────────────────────
|
||||
@property
|
||||
|
|
@ -227,8 +235,49 @@ class RelayAdapter(BasePlatformAdapter):
|
|||
async def _on_inbound(self, event) -> None:
|
||||
"""Bridge a connector-delivered MessageEvent into the normal adapter path."""
|
||||
self._capture_scope(event)
|
||||
await self._localize_inbound_media(event)
|
||||
await self.handle_message(event)
|
||||
|
||||
async def _localize_inbound_media(self, event) -> None:
|
||||
"""Download connector re-hosted attachments to local temp paths.
|
||||
|
||||
The wire's ``media_urls`` name connector re-hosts
|
||||
(``{connector}/relay/media/{id}``, per-gateway-bearer-authenticated) or
|
||||
public platform CDN URLs (Discord pass-through). Every NATIVE adapter
|
||||
presents inbound media to the agent as LOCAL FILE PATHS (the vision /
|
||||
file tools consume paths, and an authenticated URL would be useless in
|
||||
the agent's context anyway) — so mirror that here: fetch each URL and
|
||||
swap the list entries for temp paths. Best-effort per entry: a failed
|
||||
download drops that entry (never the message); no client ⇒ only
|
||||
re-host URLs are dropped (they'd 401 for every consumer downstream),
|
||||
public URLs stay.
|
||||
"""
|
||||
try:
|
||||
urls = list(getattr(event, "media_urls", None) or [])
|
||||
if not urls:
|
||||
return
|
||||
client = self._get_media_client()
|
||||
localized: list[str] = []
|
||||
for url in urls:
|
||||
if not isinstance(url, str) or not url:
|
||||
continue
|
||||
if client is None:
|
||||
# No authenticated client: keep public URLs, drop re-hosts.
|
||||
if "/relay/media/" not in url:
|
||||
localized.append(url)
|
||||
continue
|
||||
path = await client.download(url)
|
||||
if path:
|
||||
localized.append(path)
|
||||
elif "/relay/media/" not in url:
|
||||
# A public URL that failed to download still has value as
|
||||
# a URL (native adapters pass URLs to vision in some
|
||||
# lanes); a dead re-host reference does not.
|
||||
localized.append(url)
|
||||
event.media_urls = localized
|
||||
except Exception: # noqa: BLE001 - media localization must never break inbound
|
||||
logger.debug("relay inbound media localization failed", exc_info=True)
|
||||
|
||||
def _capture_scope(self, event) -> None:
|
||||
"""Remember a chat_id's egress discriminator from an inbound event so our
|
||||
outbound (the agent's reply) can re-assert it for the connector's egress
|
||||
|
|
@ -772,3 +821,233 @@ class RelayAdapter(BasePlatformAdapter):
|
|||
message_id=result.get("message_id"),
|
||||
error=result.get("error"),
|
||||
)
|
||||
|
||||
# ── Phase 2 media ─────────────────────────────────────────────────────
|
||||
|
||||
def _get_media_client(self) -> Optional[RelayMediaClient]:
|
||||
"""Lazily build the authenticated /relay/media client.
|
||||
|
||||
Uses the SAME connector base URL the WS dials and the SAME per-gateway
|
||||
(id, secret) the upgrade authenticates with — no new configuration.
|
||||
None when either is unavailable (unenrolled/dev), and every media lane
|
||||
then degrades to its pre-media fallback.
|
||||
"""
|
||||
if self._media_client is not None:
|
||||
return self._media_client
|
||||
try:
|
||||
from gateway.relay import relay_connection_auth, relay_url
|
||||
from gateway.relay.media import media_base_url
|
||||
|
||||
url = relay_url()
|
||||
gateway_id, secret = relay_connection_auth()
|
||||
if not url:
|
||||
return None
|
||||
client = RelayMediaClient(media_base_url(url), gateway_id, secret)
|
||||
if not client.enabled:
|
||||
return None
|
||||
self._media_client = client
|
||||
return client
|
||||
except Exception: # noqa: BLE001 - media plumbing must never break the adapter
|
||||
logger.debug("relay media client init failed", exc_info=True)
|
||||
return None
|
||||
|
||||
async def _send_media(
|
||||
self,
|
||||
chat_id: str,
|
||||
*,
|
||||
media_kind: str,
|
||||
source: str,
|
||||
source_is_path: bool,
|
||||
caption: Optional[str] = None,
|
||||
filename: Optional[str] = None,
|
||||
reply_to: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
) -> Optional[SendResult]:
|
||||
"""Egress one media object via the connector's ``send_media`` op.
|
||||
|
||||
``source`` is either a LOCAL file path (uploaded to the connector's
|
||||
/relay/media first — the connector cannot reach our filesystem) or an
|
||||
already-public URL (fal.media output etc. — passed through; the
|
||||
connector downloads it directly).
|
||||
|
||||
Returns None when the lane is unavailable (op not advertised, no
|
||||
transport, upload failed) so each caller can fall back to its
|
||||
pre-media behaviour — media delivery is progressive enhancement,
|
||||
never a regression when the connector predates the op.
|
||||
"""
|
||||
if self._transport is None or not self.descriptor.supports_op("send_media"):
|
||||
return None
|
||||
source_url = source
|
||||
if source_is_path:
|
||||
client = self._get_media_client()
|
||||
if client is None:
|
||||
return None
|
||||
uploaded = await client.upload(source, filename=filename)
|
||||
if not uploaded:
|
||||
return None
|
||||
source_url = uploaded
|
||||
action: Dict[str, Any] = {
|
||||
"op": "send_media",
|
||||
"chat_id": chat_id,
|
||||
"media_kind": media_kind,
|
||||
"source_url": source_url,
|
||||
"content": caption or "",
|
||||
"reply_to": reply_to,
|
||||
"metadata": self._with_scope(chat_id, metadata),
|
||||
}
|
||||
if filename:
|
||||
action["filename"] = filename
|
||||
try:
|
||||
result = await self._transport.send_outbound(
|
||||
action,
|
||||
platform=self._platform_by_chat.get(str(chat_id)),
|
||||
)
|
||||
except Exception: # noqa: BLE001 - transport failure degrades to the caller's fallback
|
||||
logger.debug("relay send_media transport failure", exc_info=True)
|
||||
return None
|
||||
if not result.get("success"):
|
||||
# A structured connector decline (size cap, platform rejection).
|
||||
# Surface it as a failed lane so the caller's fallback still
|
||||
# delivers SOMETHING (the caption/notice), mirroring native
|
||||
# adapters' upload-failure paths.
|
||||
logger.warning(
|
||||
"relay send_media declined for %s: %s",
|
||||
chat_id,
|
||||
result.get("error"),
|
||||
)
|
||||
return None
|
||||
return SendResult(
|
||||
success=True,
|
||||
message_id=result.get("message_id"),
|
||||
raw_response=result,
|
||||
)
|
||||
|
||||
async def send_image(
|
||||
self,
|
||||
chat_id: str,
|
||||
image_url: str,
|
||||
caption: Optional[str] = None,
|
||||
reply_to: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
) -> SendResult:
|
||||
"""Send an image (public URL) as a native attachment via the connector."""
|
||||
result = await self._send_media(
|
||||
chat_id,
|
||||
media_kind="image",
|
||||
source=image_url,
|
||||
source_is_path=False,
|
||||
caption=caption,
|
||||
reply_to=reply_to,
|
||||
metadata=metadata,
|
||||
)
|
||||
if result is not None:
|
||||
return result
|
||||
return await super().send_image(
|
||||
chat_id, image_url, caption=caption, reply_to=reply_to, metadata=metadata
|
||||
)
|
||||
|
||||
async def send_image_file(
|
||||
self,
|
||||
chat_id: str,
|
||||
image_path: str,
|
||||
caption: Optional[str] = None,
|
||||
reply_to: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
**kwargs,
|
||||
) -> SendResult:
|
||||
"""Send a local image file natively (upload → send_media)."""
|
||||
result = await self._send_media(
|
||||
chat_id,
|
||||
media_kind="image",
|
||||
source=image_path,
|
||||
source_is_path=True,
|
||||
caption=caption,
|
||||
reply_to=reply_to,
|
||||
metadata=metadata,
|
||||
)
|
||||
if result is not None:
|
||||
return result
|
||||
return await super().send_image_file(
|
||||
chat_id, image_path, caption=caption, reply_to=reply_to,
|
||||
metadata=metadata, **kwargs,
|
||||
)
|
||||
|
||||
async def send_voice(
|
||||
self,
|
||||
chat_id: str,
|
||||
audio_path: str,
|
||||
caption: Optional[str] = None,
|
||||
reply_to: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
**kwargs,
|
||||
) -> SendResult:
|
||||
"""Send a local audio file as a native voice message (upload → send_media)."""
|
||||
result = await self._send_media(
|
||||
chat_id,
|
||||
media_kind="voice",
|
||||
source=audio_path,
|
||||
source_is_path=True,
|
||||
caption=caption,
|
||||
reply_to=reply_to,
|
||||
metadata=metadata,
|
||||
)
|
||||
if result is not None:
|
||||
return result
|
||||
return await super().send_voice(
|
||||
chat_id, audio_path, caption=caption, reply_to=reply_to,
|
||||
metadata=metadata, **kwargs,
|
||||
)
|
||||
|
||||
async def send_video(
|
||||
self,
|
||||
chat_id: str,
|
||||
video_path: str,
|
||||
caption: Optional[str] = None,
|
||||
reply_to: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
**kwargs,
|
||||
) -> SendResult:
|
||||
"""Send a local video file natively (upload → send_media)."""
|
||||
result = await self._send_media(
|
||||
chat_id,
|
||||
media_kind="video",
|
||||
source=video_path,
|
||||
source_is_path=True,
|
||||
caption=caption,
|
||||
reply_to=reply_to,
|
||||
metadata=metadata,
|
||||
)
|
||||
if result is not None:
|
||||
return result
|
||||
return await super().send_video(
|
||||
chat_id, video_path, caption=caption, reply_to=reply_to,
|
||||
metadata=metadata, **kwargs,
|
||||
)
|
||||
|
||||
async def send_document(
|
||||
self,
|
||||
chat_id: str,
|
||||
file_path: str,
|
||||
caption: Optional[str] = None,
|
||||
file_name: Optional[str] = None,
|
||||
reply_to: Optional[str] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
**kwargs,
|
||||
) -> SendResult:
|
||||
"""Send a local file as a downloadable attachment (upload → send_media)."""
|
||||
result = await self._send_media(
|
||||
chat_id,
|
||||
media_kind="document",
|
||||
source=file_path,
|
||||
source_is_path=True,
|
||||
caption=caption,
|
||||
filename=file_name,
|
||||
reply_to=reply_to,
|
||||
metadata=metadata,
|
||||
)
|
||||
if result is not None:
|
||||
return result
|
||||
return await super().send_document(
|
||||
chat_id, file_path, caption=caption, file_name=file_name,
|
||||
reply_to=reply_to, metadata=metadata, **kwargs,
|
||||
)
|
||||
|
|
|
|||
205
gateway/relay/media.py
Normal file
205
gateway/relay/media.py
Normal file
|
|
@ -0,0 +1,205 @@
|
|||
"""Relay media client — gateway↔connector media plane (Phase 2). EXPERIMENTAL.
|
||||
|
||||
The relay wire contract carries media BY REFERENCE, never by value: an inbound
|
||||
event's ``media_urls`` name connector re-hosted attachments
|
||||
(``{connector}/relay/media/{id}``), and an outbound ``send_media`` op names a
|
||||
``source_url`` the connector resolves back to bytes. This module is the
|
||||
gateway-side HTTP client for that plane:
|
||||
|
||||
- ``download(url)`` → GET a re-hosted attachment to a local temp file (the
|
||||
agent's vision/file tools consume LOCAL paths, matching every native
|
||||
adapter's inbound media behaviour).
|
||||
- ``upload(path)`` → POST local file bytes to ``/relay/media``; returns the
|
||||
``/relay/media/{id}`` reference for a subsequent ``send_media`` op. This is
|
||||
how a locally-generated artifact (image_generate output, TTS voice note,
|
||||
a document) crosses to the connector WITHOUT the gateway needing a public
|
||||
URL.
|
||||
|
||||
Both requests present the SAME per-gateway signed bearer the WS upgrade uses
|
||||
(``make_upgrade_token``, gateway/relay/auth.py — the channel authenticator; the
|
||||
connector authenticates it with ``authenticateGatewayBearer`` on the mirrored
|
||||
routes). Uploads are
|
||||
per-gateway-owned on the connector (only this gateway can reference the id
|
||||
back); downloads accept both this gateway's uploads and connector ingest
|
||||
re-hosts.
|
||||
|
||||
Transport is stdlib ``urllib`` run in a thread executor (the same dependency
|
||||
posture as ``_post_provision`` — the relay lane adds no HTTP client deps).
|
||||
EXPERIMENTAL: may change without a deprecation cycle (docs/relay-connector-contract.md).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
import tempfile
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from gateway.relay.auth import make_upgrade_token
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mirror the connector's MEDIA_MAX_BYTES (mediaStore.ts) so an oversized local
|
||||
# artifact fails fast here instead of round-tripping to a connector 413.
|
||||
MEDIA_MAX_BYTES = 25 * 1024 * 1024
|
||||
|
||||
_REQUEST_TIMEOUT_S = 30.0
|
||||
|
||||
|
||||
def media_base_url(relay_dial_url: str) -> str:
|
||||
"""Map the ``ws(s)://…/relay`` dial URL to the ``http(s)://…`` base.
|
||||
|
||||
Same host derivation as ``_provision_url`` (gateway/relay/__init__.py):
|
||||
scheme ws→http / wss→https, trailing ``/relay`` stripped.
|
||||
"""
|
||||
raw = (relay_dial_url or "").strip().rstrip("/")
|
||||
if raw.startswith("ws://"):
|
||||
raw = "http://" + raw[len("ws://") :]
|
||||
elif raw.startswith("wss://"):
|
||||
raw = "https://" + raw[len("wss://") :]
|
||||
if raw.endswith("/relay"):
|
||||
raw = raw[: -len("/relay")]
|
||||
return raw
|
||||
|
||||
|
||||
class RelayMediaClient:
|
||||
"""Authenticated client for the connector's ``/relay/media`` routes."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
gateway_id: Optional[str],
|
||||
secret: Optional[str],
|
||||
) -> None:
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._gateway_id = gateway_id or ""
|
||||
self._secret = secret or ""
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
"""True when the client can authenticate (per-gateway creds present)."""
|
||||
return bool(self._base_url and self._gateway_id and self._secret)
|
||||
|
||||
def _bearer(self) -> str:
|
||||
return make_upgrade_token(self._gateway_id, self._secret)
|
||||
|
||||
def is_relay_media_url(self, url: str) -> bool:
|
||||
"""Is ``url`` a connector re-host reference (needs our bearer to GET)?"""
|
||||
return "/relay/media/" in (url or "")
|
||||
|
||||
async def upload(
|
||||
self,
|
||||
file_path: str,
|
||||
*,
|
||||
mime: Optional[str] = None,
|
||||
filename: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
"""POST local file bytes to ``/relay/media``; return the reference URL.
|
||||
|
||||
Returns the ``{base}/relay/media/{id}`` reference for a ``send_media``
|
||||
op's ``source_url``, or None on any failure (callers fall back to their
|
||||
pre-media behaviour — media delivery is best-effort by design).
|
||||
"""
|
||||
if not self.enabled:
|
||||
return None
|
||||
path = Path(file_path)
|
||||
try:
|
||||
data = path.read_bytes()
|
||||
except OSError:
|
||||
logger.warning("relay media upload: cannot read %s", file_path)
|
||||
return None
|
||||
if not data or len(data) > MEDIA_MAX_BYTES:
|
||||
logger.warning(
|
||||
"relay media upload: %s size %d outside (0, %d]",
|
||||
file_path,
|
||||
len(data),
|
||||
MEDIA_MAX_BYTES,
|
||||
)
|
||||
return None
|
||||
content_type = (
|
||||
mime
|
||||
or mimetypes.guess_type(filename or path.name)[0]
|
||||
or "application/octet-stream"
|
||||
)
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self._bearer()}",
|
||||
"Content-Type": content_type,
|
||||
"X-Media-Filename": (filename or path.name)[:255],
|
||||
}
|
||||
url = f"{self._base_url}/relay/media"
|
||||
|
||||
def _post() -> Optional[str]:
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT_S) as resp:
|
||||
import json
|
||||
|
||||
body = json.loads(resp.read().decode("utf-8"))
|
||||
media_id = body.get("id")
|
||||
if not media_id:
|
||||
return None
|
||||
return f"{self._base_url}/relay/media/{media_id}"
|
||||
except (urllib.error.URLError, ValueError, OSError) as exc:
|
||||
logger.warning("relay media upload failed: %s", exc)
|
||||
return None
|
||||
|
||||
return await asyncio.get_running_loop().run_in_executor(None, _post)
|
||||
|
||||
async def download(self, url: str, *, suggested_name: Optional[str] = None) -> Optional[str]:
|
||||
"""GET a re-hosted attachment to a local temp file; return its path.
|
||||
|
||||
Presents the per-gateway bearer for connector re-host URLs; plain
|
||||
public URLs (e.g. a Discord CDN pass-through) are fetched without it.
|
||||
Returns None on any failure (the event then keeps the remote URL, and
|
||||
downstream consumers that need a local file skip it — best-effort).
|
||||
"""
|
||||
if not url:
|
||||
return None
|
||||
needs_auth = self.is_relay_media_url(url)
|
||||
if needs_auth and not self.enabled:
|
||||
return None
|
||||
headers = {}
|
||||
if needs_auth:
|
||||
headers["Authorization"] = f"Bearer {self._bearer()}"
|
||||
|
||||
def _get() -> Optional[str]:
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT_S) as resp:
|
||||
length = int(resp.headers.get("Content-Length") or 0)
|
||||
if length > MEDIA_MAX_BYTES:
|
||||
logger.warning("relay media download too large: %s", url)
|
||||
return None
|
||||
data = resp.read(MEDIA_MAX_BYTES + 1)
|
||||
if not data or len(data) > MEDIA_MAX_BYTES:
|
||||
return None
|
||||
# Extension: prefer the response's content-disposition /
|
||||
# suggested name, fall back to the mime type, then .bin —
|
||||
# vision/file tools sniff by extension.
|
||||
name = suggested_name or ""
|
||||
if not name:
|
||||
cd = resp.headers.get("Content-Disposition") or ""
|
||||
if "filename=" in cd:
|
||||
name = cd.split("filename=", 1)[1].strip().strip('"')
|
||||
ext = Path(name).suffix if name else ""
|
||||
if not ext:
|
||||
mime = (resp.headers.get("Content-Type") or "").split(";")[0]
|
||||
ext = mimetypes.guess_extension(mime) or ".bin"
|
||||
fd, tmp_path = tempfile.mkstemp(prefix="relay_media_", suffix=ext)
|
||||
with os.fdopen(fd, "wb") as fh:
|
||||
fh.write(data)
|
||||
return tmp_path
|
||||
except (urllib.error.URLError, ValueError, OSError) as exc:
|
||||
logger.warning("relay media download failed for %s: %s", url, exc)
|
||||
return None
|
||||
|
||||
return await asyncio.get_running_loop().run_in_executor(None, _get)
|
||||
|
||||
|
||||
__all__ = ["RelayMediaClient", "media_base_url", "MEDIA_MAX_BYTES"]
|
||||
|
|
@ -42,6 +42,8 @@ class StubConnector:
|
|||
self.chat_info: Dict[str, Dict[str, Any]] = {}
|
||||
# Canned result for the next send_outbound (override per-test).
|
||||
self.next_send_result: Dict[str, Any] = {"success": True, "message_id": "m1"}
|
||||
# Canned result for the next send_media op (Phase 2; override per-test).
|
||||
self.next_media_result: Dict[str, Any] = {"success": True, "message_id": "md1"}
|
||||
# Canned result for the next send_follow_up (override per-test). Default
|
||||
# mimics a resolved capability egress; set success=False to simulate an
|
||||
# absent/expired capability or a tenant mismatch on the connector side.
|
||||
|
|
@ -80,6 +82,8 @@ class StubConnector:
|
|||
self.sent_platforms.append(platform)
|
||||
if action.get("op") == "send":
|
||||
return dict(self.next_send_result)
|
||||
if action.get("op") == "send_media":
|
||||
return dict(self.next_media_result)
|
||||
return {"success": True}
|
||||
|
||||
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
|
||||
|
|
|
|||
283
tests/gateway/relay/test_relay_media.py
Normal file
283
tests/gateway/relay/test_relay_media.py
Normal file
|
|
@ -0,0 +1,283 @@
|
|||
"""Relay Phase 2 media tests — send_media egress lanes + inbound media localization.
|
||||
|
||||
Covers:
|
||||
- the five ``send_*`` overrides route through ONE ``send_media`` op with the
|
||||
right ``media_kind`` and honor op-level capability gating (a connector not
|
||||
advertising ``send_media`` falls back to the base-class behaviour);
|
||||
- local-path sources upload through the RelayMediaClient first (the
|
||||
connector cannot reach our filesystem) and public URLs pass through;
|
||||
- a connector decline / failed upload degrades to the pre-media fallback;
|
||||
- inbound ``media_urls`` are localized to temp paths (re-hosts downloaded
|
||||
with the per-gateway bearer; dead re-host refs dropped; public URLs kept
|
||||
when no client is available);
|
||||
- the RelayMediaClient URL derivation + auth header shape.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.config import PlatformConfig
|
||||
from gateway.relay.adapter import RelayAdapter
|
||||
from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
|
||||
from gateway.relay.media import RelayMediaClient, media_base_url
|
||||
|
||||
from tests.gateway.relay.stub_connector import StubConnector
|
||||
|
||||
|
||||
def make_desc(**kw) -> CapabilityDescriptor:
|
||||
base = dict(
|
||||
contract_version=CONTRACT_VERSION,
|
||||
platform="telegram",
|
||||
label="Telegram",
|
||||
max_message_length=4096,
|
||||
supports_draft_streaming=False,
|
||||
supports_edit=True,
|
||||
supports_threads=True,
|
||||
markdown_dialect="markdown_v2",
|
||||
len_unit="utf16",
|
||||
supported_ops=(
|
||||
"send",
|
||||
"edit",
|
||||
"typing",
|
||||
"get_chat_info",
|
||||
"send_media",
|
||||
),
|
||||
)
|
||||
base.update(kw)
|
||||
return CapabilityDescriptor(**base)
|
||||
|
||||
|
||||
class FakeMediaClient:
|
||||
"""In-memory stand-in for RelayMediaClient (no HTTP)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.enabled = True
|
||||
self.uploads: list[tuple[str, Optional[str]]] = []
|
||||
self.downloads: list[str] = []
|
||||
self.upload_result: Optional[str] = "https://conn.example/relay/media/aa11"
|
||||
self.download_result: Optional[str] = "/tmp/relay_media_fake.png"
|
||||
|
||||
async def upload(self, file_path, *, mime=None, filename=None):
|
||||
self.uploads.append((str(file_path), filename))
|
||||
return self.upload_result
|
||||
|
||||
async def download(self, url, *, suggested_name=None):
|
||||
self.downloads.append(url)
|
||||
return self.download_result
|
||||
|
||||
def is_relay_media_url(self, url: str) -> bool:
|
||||
return "/relay/media/" in (url or "")
|
||||
|
||||
|
||||
def _adapter(**desc_kw) -> tuple[RelayAdapter, StubConnector, FakeMediaClient]:
|
||||
stub = StubConnector(make_desc(**desc_kw))
|
||||
adapter = RelayAdapter(PlatformConfig(), make_desc(**desc_kw), transport=stub)
|
||||
fake = FakeMediaClient()
|
||||
adapter._media_client = fake # bypass env-derived construction
|
||||
return adapter, stub, fake
|
||||
|
||||
|
||||
# ── egress: the five overrides ───────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_image_url_passes_through_without_upload():
|
||||
adapter, stub, fake = _adapter()
|
||||
result = await adapter.send_image(
|
||||
"chat1", "https://fal.media/x.png", caption="a pic", reply_to="m9"
|
||||
)
|
||||
assert result.success is True
|
||||
assert result.message_id == "md1"
|
||||
assert fake.uploads == [] # public URL → no upload leg
|
||||
action = stub.sent[-1]
|
||||
assert action["op"] == "send_media"
|
||||
assert action["media_kind"] == "image"
|
||||
assert action["source_url"] == "https://fal.media/x.png"
|
||||
assert action["content"] == "a pic"
|
||||
assert action["reply_to"] == "m9"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_local_path_lanes_upload_first(tmp_path: Path):
|
||||
adapter, stub, fake = _adapter()
|
||||
f = tmp_path / "clip.ogg"
|
||||
f.write_bytes(b"oggbytes")
|
||||
result = await adapter.send_voice("chat1", str(f), caption="listen")
|
||||
assert result.success is True
|
||||
assert fake.uploads == [(str(f), None)]
|
||||
action = stub.sent[-1]
|
||||
assert action["op"] == "send_media"
|
||||
assert action["media_kind"] == "voice"
|
||||
# The wire carries the RE-HOST reference, never the local path.
|
||||
assert action["source_url"] == fake.upload_result
|
||||
assert str(f) not in str(action)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_each_override_maps_to_its_media_kind(tmp_path: Path):
|
||||
adapter, stub, fake = _adapter()
|
||||
f = tmp_path / "x.bin"
|
||||
f.write_bytes(b"data")
|
||||
await adapter.send_image_file("c", str(f))
|
||||
await adapter.send_voice("c", str(f))
|
||||
await adapter.send_video("c", str(f))
|
||||
await adapter.send_document("c", str(f), file_name="report.pdf")
|
||||
kinds = [a["media_kind"] for a in stub.sent if a["op"] == "send_media"]
|
||||
assert kinds == ["image", "voice", "video", "document"]
|
||||
doc_action = stub.sent[-1]
|
||||
assert doc_action["filename"] == "report.pdf"
|
||||
# The document upload passed the user-facing filename through.
|
||||
assert fake.uploads[-1] == (str(f), "report.pdf")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_op_gating_falls_back_when_not_advertised(tmp_path: Path):
|
||||
# Connector advertises only the legacy ops — send_media must never hit the wire.
|
||||
adapter, stub, fake = _adapter(
|
||||
supported_ops=("send", "edit", "typing", "get_chat_info")
|
||||
)
|
||||
result = await adapter.send_image("chat1", "https://x.io/a.png", caption="hi")
|
||||
# Base-class fallback: caption + URL as a text send.
|
||||
assert result.success is True
|
||||
ops = [a["op"] for a in stub.sent]
|
||||
assert "send_media" not in ops
|
||||
assert ops[-1] == "send"
|
||||
assert "https://x.io/a.png" in stub.sent[-1]["content"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_legacy_empty_ops_also_gates_send_media():
|
||||
# An empty supported_ops = legacy connector; send_media is NOT in LEGACY_OPS.
|
||||
adapter, stub, _fake = _adapter(supported_ops=())
|
||||
await adapter.send_image("chat1", "https://x.io/a.png")
|
||||
assert all(a["op"] != "send_media" for a in stub.sent)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connector_decline_degrades_to_fallback(tmp_path: Path):
|
||||
adapter, stub, fake = _adapter()
|
||||
stub.next_media_result = {"success": False, "error": "media too large"}
|
||||
f = tmp_path / "big.mp4"
|
||||
f.write_bytes(b"x")
|
||||
result = await adapter.send_video("chat1", str(f), caption="cap")
|
||||
# The video lane failed → base fallback notice still delivers (a send op).
|
||||
assert result.success is True
|
||||
assert stub.sent[-1]["op"] == "send"
|
||||
assert "cap" in stub.sent[-1]["content"]
|
||||
# And the local path never leaked into the fallback text.
|
||||
assert str(f) not in stub.sent[-1]["content"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_upload_degrades_to_fallback(tmp_path: Path):
|
||||
adapter, stub, fake = _adapter()
|
||||
fake.upload_result = None # upload leg fails
|
||||
f = tmp_path / "pic.png"
|
||||
f.write_bytes(b"png")
|
||||
result = await adapter.send_image_file("chat1", str(f))
|
||||
assert result.success is True
|
||||
assert all(a["op"] != "send_media" for a in stub.sent)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scope_metadata_rides_send_media(tmp_path: Path):
|
||||
"""The egress guard resolves tenants from metadata — send_media must carry
|
||||
the same scope/user discriminators a plain send does."""
|
||||
adapter, stub, fake = _adapter()
|
||||
adapter._scope_by_chat["chat1"] = "guild9"
|
||||
await adapter.send_image("chat1", "https://x.io/p.png")
|
||||
action = stub.sent[-1]
|
||||
assert action["op"] == "send_media"
|
||||
assert (action.get("metadata") or {}).get("scope_id") == "guild9"
|
||||
|
||||
|
||||
# ── inbound localization ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_event(media_urls):
|
||||
from gateway.platforms.base import MessageEvent, MessageType
|
||||
from gateway.session import SessionSource
|
||||
|
||||
return MessageEvent(
|
||||
text="look",
|
||||
message_type=MessageType.TEXT,
|
||||
source=SessionSource(
|
||||
platform="telegram", chat_id="c1", chat_type="dm", user_id="u1"
|
||||
),
|
||||
media_urls=list(media_urls),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inbound_rehost_urls_are_localized():
|
||||
adapter, _stub, fake = _adapter()
|
||||
event = _make_event(["https://conn.example/relay/media/deadbeef"])
|
||||
await adapter._localize_inbound_media(event)
|
||||
assert fake.downloads == ["https://conn.example/relay/media/deadbeef"]
|
||||
assert event.media_urls == ["/tmp/relay_media_fake.png"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inbound_dead_rehost_ref_is_dropped_public_url_kept():
|
||||
adapter, _stub, fake = _adapter()
|
||||
fake.download_result = None # every download fails
|
||||
event = _make_event(
|
||||
[
|
||||
"https://conn.example/relay/media/deadbeef", # dead re-host → dropped
|
||||
"https://cdn.discordapp.com/attachments/a/b.png", # public → kept as URL
|
||||
]
|
||||
)
|
||||
await adapter._localize_inbound_media(event)
|
||||
assert event.media_urls == ["https://cdn.discordapp.com/attachments/a/b.png"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inbound_without_client_keeps_public_drops_rehost():
|
||||
adapter, _stub, _fake = _adapter()
|
||||
adapter._media_client = None
|
||||
adapter._get_media_client = lambda: None # type: ignore[method-assign]
|
||||
event = _make_event(
|
||||
[
|
||||
"https://conn.example/relay/media/deadbeef",
|
||||
"https://cdn.discordapp.com/attachments/a/b.png",
|
||||
]
|
||||
)
|
||||
await adapter._localize_inbound_media(event)
|
||||
assert event.media_urls == ["https://cdn.discordapp.com/attachments/a/b.png"]
|
||||
|
||||
|
||||
# ── RelayMediaClient unit surface ────────────────────────────────────────
|
||||
|
||||
|
||||
def test_media_base_url_derivation():
|
||||
assert media_base_url("wss://conn.example/relay") == "https://conn.example"
|
||||
assert media_base_url("ws://localhost:8080/relay") == "http://localhost:8080"
|
||||
assert media_base_url("https://conn.example") == "https://conn.example"
|
||||
|
||||
|
||||
def test_client_enabled_requires_full_credentials():
|
||||
assert RelayMediaClient("https://c.example", "gw1", "sec").enabled is True
|
||||
assert RelayMediaClient("https://c.example", None, "sec").enabled is False
|
||||
assert RelayMediaClient("https://c.example", "gw1", None).enabled is False
|
||||
assert RelayMediaClient("", "gw1", "sec").enabled is False
|
||||
|
||||
|
||||
def test_client_recognizes_rehost_urls():
|
||||
c = RelayMediaClient("https://c.example", "gw1", "sec")
|
||||
assert c.is_relay_media_url("https://c.example/relay/media/abc") is True
|
||||
assert c.is_relay_media_url("https://cdn.discordapp.com/a/b.png") is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_upload_rejects_oversize_and_missing(tmp_path: Path):
|
||||
c = RelayMediaClient("https://c.example", "gw1", "sec")
|
||||
# Missing file → None (no network attempted).
|
||||
assert await c.upload(str(tmp_path / "nope.bin")) is None
|
||||
# Empty file → None.
|
||||
empty = tmp_path / "empty.bin"
|
||||
empty.write_bytes(b"")
|
||||
assert await c.upload(str(empty)) is None
|
||||
Loading…
Add table
Add a link
Reference in a new issue