From 689b51bef68f9ec95b638121bb9c7fefa3703fb2 Mon Sep 17 00:00:00 2001 From: Ben Barclay Date: Sat, 25 Jul 2026 23:15:52 +1000 Subject: [PATCH] =?UTF-8?q?feat(relay):=20Phase=202=20media=20parity=20?= =?UTF-8?q?=E2=80=94=20send=5Fmedia=20egress=20+=20inbound=20media=20local?= =?UTF-8?q?ization=20(#71363)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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). --- docs/relay-connector-contract.md | 35 ++- gateway/relay/adapter.py | 279 +++++++++++++++++++++++ gateway/relay/media.py | 205 +++++++++++++++++ tests/gateway/relay/stub_connector.py | 4 + tests/gateway/relay/test_relay_media.py | 283 ++++++++++++++++++++++++ 5 files changed, 804 insertions(+), 2 deletions(-) create mode 100644 gateway/relay/media.py create mode 100644 tests/gateway/relay/test_relay_media.py diff --git a/docs/relay-connector-contract.md b/docs/relay-connector-contract.md index 79e4d480b145..5e5344288bfd 100644 --- a/docs/relay-connector-contract.md +++ b/docs/relay-connector-contract.md @@ -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 diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index 6f957ee98fe4..97f98da13727 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -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, + ) diff --git a/gateway/relay/media.py b/gateway/relay/media.py new file mode 100644 index 000000000000..32da5490ef0d --- /dev/null +++ b/gateway/relay/media.py @@ -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"] diff --git a/tests/gateway/relay/stub_connector.py b/tests/gateway/relay/stub_connector.py index c549a982c4e2..25867ca33eff 100644 --- a/tests/gateway/relay/stub_connector.py +++ b/tests/gateway/relay/stub_connector.py @@ -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]: diff --git a/tests/gateway/relay/test_relay_media.py b/tests/gateway/relay/test_relay_media.py new file mode 100644 index 000000000000..c49afc0a1921 --- /dev/null +++ b/tests/gateway/relay/test_relay_media.py @@ -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