diff --git a/plugins/platforms/line/adapter.py b/plugins/platforms/line/adapter.py index b535f9c116..ad9f54a972 100644 --- a/plugins/platforms/line/adapter.py +++ b/plugins/platforms/line/adapter.py @@ -379,6 +379,7 @@ _OUTBOUND_MEDIA = { # Inbound media kinds → cached file extension. _INBOUND_MEDIA_EXT = {"image": ".jpg", "audio": ".m4a", "video": ".mp4", "file": ".bin"} +_LIFECYCLE_EVENTS = frozenset({"follow", "unfollow", "join", "leave"}) class LineAdapter(BasePlatformAdapter): @@ -557,7 +558,7 @@ class LineAdapter(BasePlatformAdapter): await self._handle_message_event(event) elif event_type == "postback": await self._handle_postback_event(event) - elif event_type in {"follow", "unfollow", "join", "leave"}: + elif event_type in _LIFECYCLE_EVENTS: logger.info("LINE: lifecycle event %s from %s", event_type, source) else: logger.debug("LINE: ignoring event type %r", event_type) @@ -580,8 +581,7 @@ class LineAdapter(BasePlatformAdapter): local_path, media_type = await self._download_media( message_id, msg_type, filename=msg.get("fileName") or msg.get("file_name")) if local_path: - media_urls.append(local_path) - media_types.append(media_type) + media_urls, media_types = [local_path], [media_type] text = f"[{msg_type}]" elif msg_type == "sticker": keywords = msg.get("keywords") or [] @@ -593,61 +593,48 @@ class LineAdapter(BasePlatformAdapter): if chat_type == "dm" and self._client: # best-effort typing indicator (DM only) asyncio.create_task(self._client.loading(chat_id)) source_obj = self.build_source( - chat_id=chat_id, chat_type=chat_type, user_id=user_id, user_name=user_id, chat_name=chat_id - ) + chat_id=chat_id, chat_type=chat_type, user_id=user_id, user_name=user_id, chat_name=chat_id) await self.handle_message(MessageEvent( text=text, message_type=_LINE_MESSAGE_TYPES.get(msg_type, MessageType.TEXT), source=source_obj, - raw_message=event, message_id=message_id, media_urls=media_urls, media_types=media_types, - )) + raw_message=event, message_id=message_id, media_urls=media_urls, media_types=media_types)) async def _handle_postback_event(self, event: Dict[str, Any]) -> None: - """User tapped the slow-LLM postback button — deliver cached payload.""" - postback = event.get("postback") or {} - data = postback.get("data", "") or "" + """User tapped the slow-LLM postback button — deliver the cached payload. READY replies (push + fallback) and ERROR replies settle the entry; DELIVERED / PENDING just re-issue their notice.""" reply_token = event.get("replyToken", "") chat_id, _ = _resolve_chat(event.get("source") or {}) try: - parsed = json.loads(data) + parsed = json.loads((event.get("postback") or {}).get("data", "") or "") except (TypeError, json.JSONDecodeError): return - if parsed.get("action") != "show_response": - return - request_id = parsed.get("request_id", "") - if not request_id: - return - entry = self._cache.get(request_id) + request_id = parsed.get("request_id", "") if parsed.get("action") == "show_response" else "" + entry = self._cache.get(request_id) if request_id else None if not self._client or not reply_token or not entry: return - - def _settle() -> None: - self._cache.mark_delivered(request_id) - self._pending_buttons.pop(chat_id, None) - - if entry.state is State.READY: + state = entry.state + if state is State.READY: messages = _text_messages(str(entry.payload or "")) - try: - await self._client.reply(reply_token, messages) - _settle() - except Exception as exc: + elif state is State.ERROR: + messages = [_text_message(str(entry.payload or self.interrupted_text))] + else: + messages = [_text_message(self.delivered_text if state is State.DELIVERED else self.pending_text)] + try: + await self._client.reply(reply_token, messages) + except Exception as exc: + if state is State.READY: logger.warning("LINE: postback reply failed (%s); falling back to push", exc) try: await self._client.push(chat_id, messages) - _settle() except Exception as exc2: logger.error("LINE: postback push fallback failed: %s", exc2) - elif entry.state is State.ERROR: - text = str(entry.payload or self.interrupted_text) - try: - await self._client.reply(reply_token, [_text_message(text)]) - _settle() - except Exception as exc: - logger.warning("LINE: postback ERROR reply failed: %s", exc) - elif entry.state in (State.DELIVERED, State.PENDING): # "already replied" / re-issue wait notice - text = self.delivered_text if entry.state is State.DELIVERED else self.pending_text - try: - await self._client.reply(reply_token, [_text_message(text)]) - except Exception: - pass + return + else: + if state is State.ERROR: + logger.warning("LINE: postback ERROR reply failed: %s", exc) + return + if state in (State.READY, State.ERROR): + self._cache.mark_delivered(request_id) + self._pending_buttons.pop(chat_id, None) async def _download_media( self, message_id: str, msg_type: str, *, filename: Optional[str] = None @@ -692,9 +679,7 @@ class LineAdapter(BasePlatformAdapter): def _consume_reply_token(self, chat_id: str) -> Tuple[str, bool]: """Consume a stashed reply token if present and unexpired → ``(token, used_reply)``.""" token, expires_at = self._reply_tokens.pop(chat_id, None) or ("", 0.0) - if not token or time.time() >= expires_at: - return "", False - return token, True + return (token, True) if token and time.time() < expires_at else ("", False) async def send_typing(self, chat_id: str, metadata=None) -> None: """Trigger LINE's loading-animation indicator (DM only).""" @@ -808,15 +793,14 @@ class LineAdapter(BasePlatformAdapter): path is rechecked against allowed roots (tempdir, ``/tmp``→``/private/tmp`` on macOS, HERMES_HOME).""" from aiohttp import web token = request.match_info["token"] - entry = self._media_tokens.get(token) - if not entry: + file_path, expires_at = self._media_tokens.get(token) or ("", 0.0) + if not file_path: return web.Response(status=404, text="not found") - file_path, expires_at = entry if time.time() > expires_at: self._media_tokens.pop(token, None) return web.Response(status=410, text="gone") path = Path(file_path) - if not path.exists() or not path.is_file(): + if not path.is_file(): return web.Response(status=404, text="not found") try: from hermes_constants import get_hermes_home @@ -828,8 +812,8 @@ class LineAdapter(BasePlatformAdapter): if not any(resolved.is_relative_to(r) for r in allowed_roots): logger.warning("LINE: refusing to serve outside allowed roots: %s", resolved) return web.Response(status=403, text="forbidden") - content_type, _ = mimetypes.guess_type(str(path)) - return web.FileResponse(path, headers={"Content-Type": content_type or "application/octet-stream"}) + content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream" + return web.FileResponse(path, headers={"Content-Type": content_type}) async def send_image_file( self, chat_id: str, image_path: str, caption: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None @@ -841,9 +825,7 @@ class LineAdapter(BasePlatformAdapter): if not url.lower().startswith("https://"): return SendResult(success=False, error=f"LINE image URL must be HTTPS: {url}") msgs: List[Dict[str, Any]] = [{"type": "image", "originalContentUrl": url, "previewImageUrl": url}] - if caption: - msgs.append(_text_message(caption)) - return await self._send_messages(chat_id, msgs) + return await self._send_messages(chat_id, msgs + ([_text_message(caption)] if caption else [])) async def send_voice( self, chat_id: str, audio_path: str, duration_ms: int = 1000, metadata: Optional[Dict[str, Any]] = None @@ -851,8 +833,8 @@ class LineAdapter(BasePlatformAdapter): path, err = self._check_media_file("audio", audio_path) if err: return err - url = self._serve_file(path) - return await self._send_messages(chat_id, [{"type": "audio", "originalContentUrl": url, "duration": int(duration_ms)}]) + msg = {"type": "audio", "originalContentUrl": self._serve_file(path), "duration": int(duration_ms)} + return await self._send_messages(chat_id, [msg]) async def send_video( self, chat_id: str, video_path: str, preview_path: Optional[str] = None, @@ -873,10 +855,8 @@ class LineAdapter(BasePlatformAdapter): except Exception: _unlink_quietly(tmp.name) raise - video_url = self._serve_file(path) - return await self._send_messages( - chat_id, [{"type": "video", "originalContentUrl": video_url, "previewImageUrl": preview_url}] - ) + msg = {"type": "video", "originalContentUrl": self._serve_file(path), "previewImageUrl": preview_url} + return await self._send_messages(chat_id, [msg]) async def _send_messages( self, chat_id: str, messages: List[Dict[str, Any]], *, force_push: bool = False, text: bool = False @@ -887,32 +867,27 @@ class LineAdapter(BasePlatformAdapter): return SendResult(success=False, error="LINE adapter not connected") if not messages: return SendResult(success=True, message_id=None) - first_batch = messages[:LINE_MAX_MESSAGES_PER_CALL] - rest = messages[LINE_MAX_MESSAGES_PER_CALL:] + n = LINE_MAX_MESSAGES_PER_CALL + batches = [messages[i:i + n] for i in range(0, len(messages), n)] token, used_reply = self._consume_reply_token(chat_id) + start = 0 if used_reply and not force_push: try: - await self._client.reply(token, first_batch) - first_batch = None + await self._client.reply(token, batches[0]) if text: return SendResult(success=True, message_id=token) + start = 1 except Exception as exc: logger.info("LINE: reply token rejected (%s); falling back to push", exc) - if first_batch is not None: + for i in range(start, len(batches)): # push the rest (reply token is single-use) try: - await self._client.push(chat_id, first_batch) + await self._client.push(chat_id, batches[i]) except Exception as exc: - if text: + if i > 0: + logger.warning("LINE: push for follow-up batch failed: %s", exc) + elif text: logger.error("LINE: push send failed: %s", exc) return SendResult(success=False, error=str(exc)) - while rest: # subsequent batches: always push (reply token is single-use) - batch = rest[:LINE_MAX_MESSAGES_PER_CALL] - rest = rest[LINE_MAX_MESSAGES_PER_CALL:] - try: - await self._client.push(chat_id, batch) - except Exception as exc: - logger.warning("LINE: push for follow-up batch failed: %s", exc) - return SendResult(success=False, error=str(exc)) return SendResult(success=True, message_id=None) @@ -923,9 +898,13 @@ def _unlink_quietly(path: str) -> None: pass +def _env_credentials_present() -> bool: + return bool(_get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") and _get_scoped_secret("LINE_CHANNEL_SECRET")) + + def check_requirements() -> bool: """Plugin gate: require credentials AND aiohttp at runtime.""" - if not (_get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") and _get_scoped_secret("LINE_CHANNEL_SECRET")): + if not _env_credentials_present(): return False try: import aiohttp # noqa: F401 @@ -935,8 +914,7 @@ def check_requirements() -> bool: def validate_config(config) -> bool: - token, secret = _credentials(config) - return bool(token) and bool(secret) + return all(_credentials(config)) def is_connected(config) -> bool: @@ -946,7 +924,7 @@ def is_connected(config) -> bool: def _env_enablement() -> Optional[Dict[str, Any]]: """Seed PlatformConfig.extra from env-only setups so ``hermes status`` sees them.""" - if not (_get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") and _get_scoped_secret("LINE_CHANNEL_SECRET")): + if not _env_credentials_present(): return None seeded: Dict[str, Any] = {} if os.getenv("LINE_PORT"): @@ -957,17 +935,15 @@ def _env_enablement() -> Optional[Dict[str, Any]]: for env, key in (("LINE_HOST", "host"), ("LINE_PUBLIC_URL", "public_url"), ("LINE_HOME_CHANNEL", "home_channel")): if os.getenv(env): seeded[key] = os.environ[env] - return seeded or {} + return seeded async def _standalone_send( pconfig, chat_id: str, message: str, *, thread_id: Optional[str] = None, media_files: Optional[List[str]] = None, force_document: bool = False, ) -> Dict[str, Any]: - """Out-of-process Push delivery for cron jobs detached from the gateway. - Always Push (no inbound event → no reply token). ``thread_id`` is ignored (LINE - has no threads); ``media_files`` can't be served without the webhook server. - """ + """Out-of-process Push delivery for cron jobs detached from the gateway (no inbound event → no + reply token). ``thread_id`` is ignored (no threads); ``media_files`` need the webhook server.""" extra = getattr(pconfig, "extra", {}) or {} token = _get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") or extra.get("channel_access_token", "") if not token or not chat_id: @@ -1014,8 +990,7 @@ def interactive_setup() -> None: _prompt("LINE_CHANNEL_SECRET", "Channel secret", secret=True) _prompt("LINE_PUBLIC_URL", "Public HTTPS base URL (optional, e.g. https://my-tunnel.example.com)") _prompt("LINE_ALLOWED_USERS", "Allowed user IDs (comma-separated; blank=skip)") - print("Done. Set the webhook URL in the LINE console to " - "/line/webhook and enable 'Use webhook'.") + print("Done. Set the webhook URL in the LINE console to /line/webhook and enable 'Use webhook'.") def register(ctx) -> None: diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py index fa3f251c35..c504abd161 100644 --- a/plugins/platforms/simplex/adapter.py +++ b/plugins/platforms/simplex/adapter.py @@ -1,41 +1,12 @@ -"""SimpleX Chat platform adapter (Hermes plugin). +"""SimpleX Chat adapter: persistent WebSocket to a simplex-chat daemon (``simplex-chat -p 5225``, +or ``docker run -p 5225:5225 simplexchat/simplex-chat-cli -p 5225``); JSON commands out, events in. -Connects to a simplex-chat daemon running in WebSocket mode. Inbound messages -arrive via a persistent WebSocket connection; outbound messages use the same -WebSocket with JSON commands. The plugin loader calls ``register(ctx)`` at -startup and the platform becomes available to ``gateway/run.py`` and -``tools/send_message_tool`` through the registry. - -SimpleX chat daemon setup: - simplex-chat -p 5225 # start daemon on port 5225 - # or via Docker: - # docker run -p 5225:5225 simplexchat/simplex-chat-cli -p 5225 - -Required environment variables: - SIMPLEX_WS_URL WebSocket URL of the daemon - (default: ws://127.0.0.1:5225) - -Optional environment variables: - SIMPLEX_ALLOWED_USERS Comma-separated allowlist. Each entry may be - either a numeric contactId (stable across - renames; visible via `/contacts` in the CLI) - or a contact display name (what the SimpleX - UI shows). Both forms are accepted. - SIMPLEX_ALLOW_ALL_USERS Set 'true' to allow all contacts - SIMPLEX_AUTO_ACCEPT Set 'false' to disable contact-request auto-accept - (default: 'true') - SIMPLEX_GROUP_ALLOWED Comma-separated group IDs to monitor, or '*' - for any group. Omit to disable groups entirely. - SIMPLEX_HOME_CHANNEL Default contact/group ID for cron delivery - SIMPLEX_HOME_CHANNEL_NAME Human label for the home channel - HERMES_SIMPLEX_TEXT_BATCH_DELAY - Quiet-period seconds (default: 0.8) used to - concatenate rapid-fire inbound text messages - into a single MessageEvent. - -``websockets`` is imported lazily so the plugin stays discoverable (and -``hermes setup`` can describe it) when the package is missing; -``check_requirements()`` returns False until it is present. +Env: SIMPLEX_WS_URL (required; default ws://127.0.0.1:5225) · SIMPLEX_ALLOWED_USERS (numeric +contactIds — stable across renames, see ``/contacts`` — or display names) · SIMPLEX_ALLOW_ALL_USERS · +SIMPLEX_AUTO_ACCEPT ('false' disables contact-request auto-accept; default true) · +SIMPLEX_GROUP_ALLOWED (group IDs or '*'; omit to ignore groups) · SIMPLEX_HOME_CHANNEL[_NAME] · +HERMES_SIMPLEX_TEXT_BATCH_DELAY (quiet seconds, default 0.8, merging rapid-fire inbound text). +``websockets`` is imported lazily so the plugin stays discoverable when the package is missing. """ import asyncio @@ -50,9 +21,11 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional +from urllib.parse import unquote + from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret from gateway.config import Platform, PlatformConfig -from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult +from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult, cache_image_from_url logger = logging.getLogger(__name__) @@ -67,6 +40,7 @@ _CORR_PREFIX = "hermes-" _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp"} _AUDIO_EXTS = {".mp3", ".wav", ".ogg", ".m4a", ".aac", ".opus"} +_VOICE_TAG_EXTS = {".ogg", ".mp3", ".wav", ".m4a", ".opus"} # MEDIA: tags sent as voice notes def _parse_comma_list(value: str) -> List[str]: @@ -90,18 +64,16 @@ def _is_audio_ext(ext: str) -> bool: return ext.lower() in _AUDIO_EXTS -def _send_cmd(chat_id: str, items: list) -> str: - """Build a structured ``/_send`` command addressing *chat_id* by ID. +def _display_name(obj: dict, profile_key: str, default: str = "") -> str: + """``localDisplayName`` falling back to the nested profile's ``displayName``.""" + return obj.get("localDisplayName", "") or (obj.get(profile_key, {}) or {}).get("displayName", default) - The structured json form is used (rather than ``@name text`` / ``#[id]``) - because the daemon parses the bare syntax as a display-name lookup and - silently drops messages when the name doesn't resolve; json.dumps also - escapes newlines/special chars correctly. - """ - composed = json.dumps(items) - if chat_id.startswith("group:"): - return f"/_send #{chat_id[6:]} json {composed}" - return f"/_send @{chat_id} json {composed}" + +def _send_cmd(chat_id: str, items: list) -> str: + """Structured ``/_send json …`` addressing *chat_id* by ID. The bare ``@name`` / ``#[id]`` + syntax is a display-name lookup that silently drops unresolved names; json.dumps escapes text.""" + target = f"#{chat_id[6:]}" if chat_id.startswith("group:") else f"@{chat_id}" + return f"/_send {target} json {json.dumps(items)}" async def _cancel_task(task: Optional[asyncio.Task]) -> None: @@ -123,18 +95,14 @@ class SimplexAdapter(BasePlatformAdapter): extra = getattr(config, "extra", {}) or {} self.ws_url = extra.get("ws_url", "ws://127.0.0.1:5225").rstrip("/") - # Auto-accept is on by default; env wins over the ``_env_enablement`` seed. env_auto = _get_scoped_secret("SIMPLEX_AUTO_ACCEPT") if env_auto is not None: self.auto_accept = env_auto.strip().lower() not in {"0", "false", "no", ""} else: self.auto_accept = bool(extra.get("auto_accept", True)) - - # Without SIMPLEX_GROUP_ALLOWED, group messages are ignored entirely - # (safer default). ``*`` accepts any group. - group_allowed_str = (_get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "") - or extra.get("group_allowed", "")) + # Without SIMPLEX_GROUP_ALLOWED group messages are ignored (safer default); ``*`` = any group. + group_allowed_str = _get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "") or extra.get("group_allowed", "") self.group_allow_from = set(_parse_comma_list(group_allowed_str)) self._ws = None # websockets connection @@ -142,31 +110,18 @@ class SimplexAdapter(BasePlatformAdapter): self._health_task: Optional[asyncio.Task] = None self._running = False self._last_ws_activity = 0.0 - - # Cosmetic echo filter: corrIds we minted, bounded. - self._pending_corr_ids: set = set() + self._pending_corr_ids: set = set() # cosmetic echo filter: corrIds we minted, bounded self._max_pending_corr = 200 - - # File transfers awaiting rcvFileComplete (keyed by fileId). - self._pending_file_transfers: Dict[int, dict] = {} - - # Futures for commands whose responses we actually await (``_send_command``). - self._pending_responses: Dict[str, asyncio.Future] = {} + self._pending_file_transfers: Dict[int, dict] = {} # awaiting rcvFileComplete, by fileId + self._pending_responses: Dict[str, asyncio.Future] = {} # awaited command replies self._corr_counter = 0 - # Text batching state consumed by BasePlatformAdapter._enqueue_text_event. self._text_batch_delay = float(os.getenv("HERMES_SIMPLEX_TEXT_BATCH_DELAY", "0.8")) self._pending_text_batches: Dict[str, MessageEvent] = {} self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {} - logger.info( "SimpleX adapter initialized: url=%s auto_accept=%s groups=%s", - self.ws_url, self.auto_accept, "enabled" if self.group_allow_from else "disabled", - ) - - # ------------------------------------------------------------------ - # Lifecycle - # ------------------------------------------------------------------ + self.ws_url, self.auto_accept, "enabled" if self.group_allow_from else "disabled") async def connect(self, *, is_reconnect: bool = False) -> bool: """Connect to the simplex-chat daemon and start the WebSocket listener.""" @@ -179,22 +134,17 @@ class SimplexAdapter(BasePlatformAdapter): if not self.ws_url: logger.error("SimpleX: SIMPLEX_WS_URL is required") return False - - # Quick connectivity check — open and immediately close. - try: + try: # quick connectivity check — open and immediately close async with _wsclient.connect(self.ws_url, open_timeout=10): pass except Exception as e: logger.error("SimpleX: cannot reach daemon at %s: %s", self.ws_url, e) return False - self._running = True self._last_ws_activity = time.time() self._ws_task = asyncio.create_task(self._ws_listener()) self._health_task = asyncio.create_task(self._health_monitor()) - - if hasattr(self, "_mark_connected"): - self._mark_connected() + self._mark_connected() logger.info("SimpleX: connected to %s", self.ws_url) self._wire_plugin_handlers(None) return True @@ -204,33 +154,22 @@ class SimplexAdapter(BasePlatformAdapter): self._running = False await _cancel_task(self._ws_task) await _cancel_task(self._health_task) - if self._ws: try: await self._ws.close() except Exception: pass self._ws = None - - for task in list(self._pending_text_batch_tasks.values()): - if not task.done(): - task.cancel() + for pending in (self._pending_text_batch_tasks.values(), self._pending_responses.values()): + for item in list(pending): + if not item.done(): + item.cancel() self._pending_text_batch_tasks.clear() self._pending_text_batches.clear() - - for fut in self._pending_responses.values(): - if not fut.done(): - fut.cancel() self._pending_responses.clear() - - if hasattr(self, "_mark_disconnected"): - self._mark_disconnected() + self._mark_disconnected() logger.info("SimpleX: disconnected") - # ------------------------------------------------------------------ - # WebSocket listener / health - # ------------------------------------------------------------------ - async def _ws_listener(self) -> None: """Maintain a persistent WebSocket connection to the daemon.""" import websockets as _wsclient @@ -240,14 +179,11 @@ class SimplexAdapter(BasePlatformAdapter): while self._running: try: logger.debug("SimpleX WS: connecting to %s", self.ws_url) - async with _wsclient.connect( - self.ws_url, ping_interval=20, ping_timeout=20, close_timeout=10 - ) as ws: + async with _wsclient.connect(self.ws_url, ping_interval=20, ping_timeout=20, close_timeout=10) as ws: self._ws = ws backoff = WS_RETRY_DELAY_INITIAL self._last_ws_activity = time.time() logger.info("SimpleX WS: connected") - async for raw in ws: if not self._running: break @@ -262,26 +198,19 @@ class SimplexAdapter(BasePlatformAdapter): break except ConnectionClosed as e: if self._running: - logger.warning("SimpleX WS: connection closed: %s (reconnecting in %.0fs)", - e, backoff) + logger.warning("SimpleX WS: connection closed: %s (reconnecting in %.0fs)", e, backoff) except Exception as e: if self._running: - logger.warning("SimpleX WS: unexpected error: %s (reconnecting in %.0fs)", - e, backoff) + logger.warning("SimpleX WS: unexpected error: %s (reconnecting in %.0fs)", e, backoff) finally: self._ws = None - if self._running: await asyncio.sleep(backoff + backoff * 0.2 * random.random()) backoff = min(backoff * 2, WS_RETRY_DELAY_MAX) async def _health_monitor(self) -> None: - """Observe WebSocket idleness without reconnecting healthy quiet links. - - simplex-chat can legitimately stay application-silent for long periods; - the websockets client already sends protocol pings, so idleness is only - logged — reconnecting on it causes needless churn. - """ + """Log (never reconnect on) WebSocket idleness: simplex-chat legitimately stays + application-silent for long periods and the client already sends protocol pings.""" while self._running: await asyncio.sleep(HEALTH_CHECK_INTERVAL) if not self._running: @@ -290,83 +219,74 @@ class SimplexAdapter(BasePlatformAdapter): if elapsed > HEALTH_CHECK_STALE_THRESHOLD: logger.debug("SimpleX: WS application-idle for %.0fs", elapsed) - # ------------------------------------------------------------------ - # Inbound event handling - # ------------------------------------------------------------------ - async def _handle_event(self, event: dict) -> None: """Dispatch a daemon event to the appropriate handler.""" - # Messages are usually {"corrId": ..., "resp": {"type": ...}}, but some - # daemons put the response fields at top level — normalize both. + # Usually {"corrId": ..., "resp": {"type": ...}}, but some daemons put the + # response fields at top level — normalize both. resp = event.get("resp") if isinstance(event.get("resp"), dict) else event corr_id = event.get("corrId") - if corr_id and corr_id in self._pending_responses: fut = self._pending_responses.pop(corr_id) if not fut.done(): fut.set_result(resp) return - - # Cosmetic echo filter: prefixed corrIds are ours but weren't awaited. if corr_id and isinstance(corr_id, str) and corr_id.startswith(_CORR_PREFIX): - self._pending_corr_ids.discard(corr_id) + self._pending_corr_ids.discard(corr_id) # our own un-awaited echo return - resp_type = resp.get("type") or event.get("type", "") - - if resp_type == "contactRequest" and self.auto_accept: - contact_req = resp.get("contactRequest", {}) or {} - contact_req_id = contact_req.get("contactRequestId") - if contact_req_id is not None: - logger.info("SimpleX: auto-accepting contact request %s", - _redact_id(str(contact_req_id))) - await self._send_command(f"/accept {contact_req_id}") - return - - # simplex fires rcvFileDescrReady before newChatItems for some (XFTP) - # files; start the download now, the chat item arrives later. - if resp_type == "rcvFileDescrReady": - rcv_file = resp.get("rcvFileTransfer", {}) or {} - file_id = rcv_file.get("fileId") if isinstance(rcv_file, dict) else None - if file_id is not None: - logger.debug("SimpleX: rcvFileDescrReady for fileId=%s — sending /freceive", - file_id) - await self._send_fire_and_forget(f"/freceive {file_id}") - return - - if resp_type == "newChatItems": - chat_items = resp.get("chatItems", []) or [] - if not isinstance(chat_items, list): - chat_items = [chat_items] - for item in chat_items: - await self._safe_handle_chat_item(item, "SimpleX: error processing chat item") - return - - # Singular variant — some daemon versions emit this - if resp_type == "newChatItem": - await self._safe_handle_chat_item(resp, "SimpleX: error processing chat item") - return - - # File transfer completion — deliver any deferred chat item - if resp_type == "rcvFileComplete": - chat_item_data = (resp.get("chatItem", {}) or {}).get("chatItem", {}) or {} - file_info = chat_item_data.get("file", {}) or {} - file_id = file_info.get("fileId") if isinstance(file_info, dict) else None - if file_id is not None and file_id in self._pending_file_transfers: - pending = self._pending_file_transfers.pop(file_id) - file_source = file_info.get("fileSource", {}) or {} - file_path = file_source.get("filePath") if isinstance(file_source, dict) else None - if file_path: - pending_item_data = pending.get("chatItem", {}) or {} - pending_item_data.setdefault("file", {})["fileSource"] = {"filePath": file_path} - pending["chatItem"] = pending_item_data - await self._safe_handle_chat_item( - pending, "SimpleX: error processing deferred file message") - return - - if resp_type: + handler = self._EVENT_HANDLERS.get(resp_type) + if handler is not None: + await handler(self, resp) + elif resp_type: logger.debug("SimpleX: unhandled event type: %s", resp_type) + async def _on_contact_request(self, resp: dict) -> None: + if not self.auto_accept: + logger.debug("SimpleX: unhandled event type: %s", "contactRequest") + return + contact_req_id = (resp.get("contactRequest", {}) or {}).get("contactRequestId") + if contact_req_id is not None: + logger.info("SimpleX: auto-accepting contact request %s", _redact_id(str(contact_req_id))) + await self._send_command(f"/accept {contact_req_id}") + + async def _on_rcv_file_descr_ready(self, resp: dict) -> None: + """XFTP files fire this before newChatItems; start the download now, the chat item arrives later.""" + rcv_file = resp.get("rcvFileTransfer", {}) or {} + file_id = rcv_file.get("fileId") if isinstance(rcv_file, dict) else None + if file_id is not None: + logger.debug("SimpleX: rcvFileDescrReady for fileId=%s — sending /freceive", file_id) + await self._send_fire_and_forget(f"/freceive {file_id}") + + async def _on_new_chat_items(self, resp: dict) -> None: + chat_items = resp.get("chatItems", []) or [] + for item in chat_items if isinstance(chat_items, list) else [chat_items]: + await self._safe_handle_chat_item(item, "SimpleX: error processing chat item") + + async def _on_new_chat_item(self, resp: dict) -> None: + """Singular variant emitted by some daemon versions.""" + await self._safe_handle_chat_item(resp, "SimpleX: error processing chat item") + + async def _on_rcv_file_complete(self, resp: dict) -> None: + """Deliver a chat item deferred until its file transfer completed.""" + chat_item_data = (resp.get("chatItem", {}) or {}).get("chatItem", {}) or {} + file_info = chat_item_data.get("file", {}) or {} + file_id = file_info.get("fileId") if isinstance(file_info, dict) else None + if file_id is None or file_id not in self._pending_file_transfers: + return + pending = self._pending_file_transfers.pop(file_id) + file_source = file_info.get("fileSource", {}) or {} + file_path = file_source.get("filePath") if isinstance(file_source, dict) else None + if file_path: + pending_item_data = pending.get("chatItem", {}) or {} + pending_item_data.setdefault("file", {})["fileSource"] = {"filePath": file_path} + pending["chatItem"] = pending_item_data + await self._safe_handle_chat_item(pending, "SimpleX: error processing deferred file message") + + _EVENT_HANDLERS = { + "contactRequest": _on_contact_request, "rcvFileDescrReady": _on_rcv_file_descr_ready, + "newChatItems": _on_new_chat_items, "newChatItem": _on_new_chat_item, "rcvFileComplete": _on_rcv_file_complete, + } + async def _safe_handle_chat_item(self, item: dict, err_msg: str) -> None: try: await self._handle_chat_item(item) @@ -381,17 +301,13 @@ class SimplexAdapter(BasePlatformAdapter): meta = chat_item_data.get("meta", {}) or {} content = chat_item_data.get("content", {}) or {} msg_content = content.get("msgContent", {}) or {} - - # Filter out our own messages item_direction = chat_item_data.get("chatDir", {}) or {} direction_type = item_direction.get("type", "") if isinstance(item_direction, dict) else "" - if direction_type in ("directSnd", "groupSnd"): + if direction_type in ("directSnd", "groupSnd"): # our own messages return - content_type = content.get("type", "") if isinstance(content, dict) else "" if content_type != "rcvMsgContent": return - text = "" msg_type_str = msg_content.get("type", "") if isinstance(msg_content, dict) else "" if msg_type_str in ("text", "file", "image", "voice", "link", "video"): @@ -399,36 +315,28 @@ class SimplexAdapter(BasePlatformAdapter): if not text and msg_type_str not in ("image", "file", "voice"): return - sender_id = sender_name = chat_id = "" - is_group = False + is_group = chat_type == "group" if chat_type == "direct": contact = chat_info.get("contact", {}) or {} - sender_id = str(contact.get("contactId", "")) - sender_name = (contact.get("localDisplayName", "") - or contact.get("profile", {}).get("displayName", "")) - chat_id = sender_id - elif chat_type == "group": + sender_id = chat_id = str(contact.get("contactId", "")) + sender_name = chat_name = _display_name(contact, "profile") + elif is_group: group_info = chat_info.get("groupInfo", {}) or {} group_id = str(group_info.get("groupId", "")) chat_id = f"group:{group_id}" - is_group = True - member = item_direction.get("groupMember", {}) or {} sender_id = str(member.get("memberId", "")) - sender_name = (member.get("localDisplayName", "") - or member.get("memberProfile", {}).get("displayName", "")) - - if self.group_allow_from: - if "*" not in self.group_allow_from and group_id not in self.group_allow_from: - logger.debug("SimpleX: group %s not in allowlist", _redact_id(group_id)) - return - else: + sender_name = _display_name(member, "memberProfile") + chat_name = _display_name(group_info, "groupProfile", chat_id) + if not self.group_allow_from: logger.debug("SimpleX: ignoring group message (no SIMPLEX_GROUP_ALLOWED)") return + if "*" not in self.group_allow_from and group_id not in self.group_allow_from: + logger.debug("SimpleX: group %s not in allowlist", _redact_id(group_id)) + return else: logger.debug("SimpleX: unhandled chat type: %s", chat_type) return - if not sender_id: logger.debug("SimpleX: ignoring message with no sender") return @@ -440,22 +348,17 @@ class SimplexAdapter(BasePlatformAdapter): if file_info and isinstance(file_info, dict): file_source = file_info.get("fileSource", {}) or {} file_path = file_source.get("filePath") if isinstance(file_source, dict) else None - file_name = file_info.get("fileName", "") file_id = file_info.get("fileId") - ext = Path(file_path).suffix.lower() if file_path else "" - if not ext and file_name: - ext = Path(file_name).suffix.lower() - - # Voice notes typically arrive before the file finishes downloading; - # defer until rcvFileComplete. /freceive gets no corrId reply, so - # awaiting one would block the event loop. + if not ext and file_info.get("fileName", ""): + ext = Path(file_info["fileName"]).suffix.lower() + # Voice notes typically arrive before the file finishes downloading; defer until + # rcvFileComplete. /freceive gets no corrId reply, so awaiting one would block the loop. if not file_path and _is_audio_ext(ext) and file_id is not None: logger.info("SimpleX: voice file %d not yet received, accepting transfer", file_id) self._pending_file_transfers[file_id] = chat_item await self._send_fire_and_forget(f"/freceive {file_id}") return - if file_path: if _is_image_ext(ext): mime = f"image/{ext.lstrip('.')}" @@ -466,53 +369,34 @@ class SimplexAdapter(BasePlatformAdapter): media_urls.append(file_path) media_types.append(mime) - chat_name = sender_name - if is_group: - chat_name = group_info.get("localDisplayName", "") or group_info.get( - "groupProfile", {} - ).get("displayName", chat_id) source = self.build_source( chat_id=chat_id, chat_name=chat_name, chat_type="group" if is_group else "dm", - user_id=sender_id, user_name=sender_name or sender_id, - ) - + user_id=sender_id, user_name=sender_name or sender_id) msg_type = MessageType.TEXT if media_types: if any(mt.startswith("audio/") for mt in media_types): msg_type = MessageType.VOICE elif any(mt.startswith("image/") for mt in media_types): msg_type = MessageType.PHOTO - else: - # Non-image/non-audio files are documents so run.py's - # document-context injection surfaces the file to the agent. + else: # other files are DOCUMENT so run.py's document-context injection surfaces them msg_type = MessageType.DOCUMENT - ts_str = meta.get("itemTs") or meta.get("createdAt", "") try: - timestamp = (datetime.fromisoformat(ts_str.replace("Z", "+00:00")) if ts_str - else datetime.now(tz=timezone.utc)) + timestamp = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) if ts_str else datetime.now(tz=timezone.utc) except (ValueError, AttributeError): timestamp = datetime.now(tz=timezone.utc) - msg_event = MessageEvent( source=source, text=text or "", message_type=msg_type, media_urls=media_urls, - media_types=media_types, timestamp=timestamp, raw_message=chat_item, - ) - logger.debug("SimpleX: message from %s in %s: %s", - _redact_id(sender_id), chat_id[:20], (text or "")[:50]) - - # Batch rapid-fire text so the agent sees one combined message. - if msg_type == MessageType.TEXT and text: + media_types=media_types, timestamp=timestamp, raw_message=chat_item) + logger.debug("SimpleX: message from %s in %s: %s", _redact_id(sender_id), chat_id[:20], (text or "")[:50]) + if msg_type == MessageType.TEXT and text: # batch rapid-fire text into one combined message self._enqueue_text_event(msg_event) else: await self.handle_message(msg_event) - # ------------------------------------------------------------------ - # Text message batching (enqueue lives on BasePlatformAdapter) - # ------------------------------------------------------------------ + # Text batching: enqueue lives on BasePlatformAdapter. def _text_batch_key(self, event: MessageEvent) -> str: - """Session-scoped key for text message batching.""" return f"{event.source.platform.value}:{event.source.chat_id}" async def _flush_text_batch(self, key: str) -> None: @@ -529,26 +413,14 @@ class SimplexAdapter(BasePlatformAdapter): if self._pending_text_batch_tasks.get(key) is current_task: self._pending_text_batch_tasks.pop(key, None) - # ------------------------------------------------------------------ - # Command interface - # ------------------------------------------------------------------ - def _make_corr_id(self) -> str: - """Mint a correlation ID and remember it for echo-filtering. - - ``_pending_corr_ids`` is bounded: past ``_max_pending_corr`` the - overflow is evicted in a single sweep. - """ + """Mint a correlation ID and remember it for echo-filtering; the set is bounded by + ``_max_pending_corr`` (overflow evicted in one sweep).""" self._corr_counter += 1 corr_id = f"{_CORR_PREFIX}{self._corr_counter}-{int(time.time() * 1000)}" self._pending_corr_ids.add(corr_id) - if len(self._pending_corr_ids) > self._max_pending_corr: - overflow = len(self._pending_corr_ids) - self._max_pending_corr - for _ in range(overflow): - try: - self._pending_corr_ids.pop() - except KeyError: - break + for _ in range(len(self._pending_corr_ids) - self._max_pending_corr): + self._pending_corr_ids.pop() return corr_id async def _send_ws(self, payload: dict) -> None: @@ -570,20 +442,17 @@ class SimplexAdapter(BasePlatformAdapter): return None corr_id = self._make_corr_id() - payload = json.dumps({"corrId": corr_id, "cmd": command}) fut: asyncio.Future = asyncio.get_event_loop().create_future() self._pending_responses[corr_id] = fut try: - await ws.send(payload) + await ws.send(json.dumps({"corrId": corr_id, "cmd": command})) return await asyncio.wait_for(fut, timeout=timeout) except asyncio.TimeoutError: logger.warning("SimpleX: command timed out: %s", command[:50]) - self._pending_responses.pop(corr_id, None) - return None except Exception as e: logger.warning("SimpleX: command failed: %s — %s", command[:50], e) - self._pending_responses.pop(corr_id, None) - return None + self._pending_responses.pop(corr_id, None) + return None async def _send_fire_and_forget(self, command: str) -> None: """Send a command the daemon never replies to with a corrId (e.g. ``/freceive``).""" @@ -592,113 +461,67 @@ class SimplexAdapter(BasePlatformAdapter): async def _send_items(self, chat_id: str, items: list, error: str) -> SendResult: """Send a structured ``/_send`` payload and await the reply.""" result = await self._send_command(_send_cmd(chat_id, items)) - if result is not None: - return SendResult(success=True) - return SendResult(success=False, error=error) - - # ------------------------------------------------------------------ - # Outbound — text - # ------------------------------------------------------------------ + return SendResult(success=True) if result is not None else SendResult(success=False, error=error) async def send( - self, - chat_id: str, - content: str, - reply_to: Optional[str] = None, - metadata: Optional[Dict[str, Any]] = None, + self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: - """Send a text message. - - ``MEDIA:`` tags (embedded by TTS / audio tools) are stripped from - the body and sent as native voice notes or documents. The text send is - fire-and-forget at the WebSocket level: the daemon doesn't always return - a corrId reply for chat commands, and waiting would serialise all - outbound traffic behind a 30-second timeout. - """ - _voice_exts = {".ogg", ".mp3", ".wav", ".m4a", ".opus"} + """Send text; ``MEDIA:`` tags (TTS / audio tools) are stripped and sent as native voice + notes or documents. The text send is fire-and-forget: the daemon doesn't always return a corrId + reply for chat commands, and waiting would serialise all outbound traffic behind a 30s timeout.""" media_paths = re.findall(r"MEDIA:(\S+)", content) if media_paths: content = re.sub(r"MEDIA:\S+", "", content).strip() - if content: - corr_id = self._make_corr_id() cmd_str = _send_cmd(chat_id, [{"msgContent": {"type": "text", "text": content}}]) - await self._send_ws({"corrId": corr_id, "cmd": cmd_str}) - + await self._send_ws({"corrId": self._make_corr_id(), "cmd": cmd_str}) for path in media_paths: - if os.path.splitext(path)[1].lower() in _voice_exts: + if os.path.splitext(path)[1].lower() in _VOICE_TAG_EXTS: media_result = await self.send_voice(chat_id, path) else: media_result = await self.send_document(chat_id, path) if not media_result.success: return media_result - return SendResult(success=True) - # ------------------------------------------------------------------ - # Channel directory enumeration - # ------------------------------------------------------------------ - async def list_channels(self) -> Optional[List[Dict[str, Any]]]: """Enumerate contacts and allowed groups for the channel directory. - Returns ``None`` (not ``[]``) when the WebSocket is down or the daemon - is unresponsive so the directory falls back to session-history - discovery instead of wiping known targets. Entry ``id`` values match - the adapter's send targets: display name for DMs, ``group:`` for groups. + Returns ``None`` (not ``[]``) when the WebSocket is down or the daemon is unresponsive so + the directory falls back to session-history discovery instead of wiping known targets. + Entry ``id`` values match the send targets: display name for DMs, ``group:`` for groups. """ if not self._ws: return None - - channels: List[Dict[str, Any]] = [] - resp = await self._send_command("/contacts", timeout=10.0) if resp is None: return None + channels: List[Dict[str, Any]] = [] for contact in resp.get("contacts") or []: if not isinstance(contact, dict): continue contact_id = contact.get("contactId") - name = (contact.get("localDisplayName", "") - or (contact.get("profile", {}) or {}).get("displayName", "")) + name = _display_name(contact, "profile") if contact_id is None and not name: continue # Display name is what the DM send path addresses; fall back to contactId. - channels.append( - {"id": str(name or contact_id), "name": str(name or contact_id), "type": "dm"}) - + channels.append({"id": str(name or contact_id), "name": str(name or contact_id), "type": "dm"}) resp = await self._send_command("/groups", timeout=10.0) - if resp is not None: - for group in resp.get("groups") or []: - # Each group is either a groupInfo dict or a [groupInfo, groupSummary] pair. - if isinstance(group, list) and group: - group = group[0] - if not isinstance(group, dict): - continue - group_id = group.get("groupId") - if group_id is None: - continue - name = ( - group.get("localDisplayName", "") - or (group.get("groupProfile", {}) or {}).get("displayName", "") - or str(group_id) - ) - channels.append({"id": f"group:{group_id}", "name": str(name), "type": "group"}) - + for group in (resp.get("groups") or []) if resp is not None else []: + if isinstance(group, list) and group: # groupInfo dict or [groupInfo, groupSummary] pair + group = group[0] + if not isinstance(group, dict) or group.get("groupId") is None: + continue + group_id = group["groupId"] + name = _display_name(group, "groupProfile") or str(group_id) + channels.append({"id": f"group:{group_id}", "name": str(name), "type": "group"}) return channels - # ------------------------------------------------------------------ - # Outbound — media - # ------------------------------------------------------------------ - @staticmethod def _prepare_image(file_path: str) -> tuple[str, str]: - """Ensure *file_path* is PNG/JPEG and return ``(png_path, thumb_data_uri)``. - - SimpleX clients can't display WebP etc. inline, so convert to PNG when - needed and build a 128px JPEG thumbnail for the ``image`` field (inline - preview). Uses Pillow when available, else ImageMagick ``convert``. - """ + """Ensure *file_path* is PNG/JPEG and return ``(png_path, thumb_data_uri)``. SimpleX clients + can't show WebP etc. inline, so convert to PNG when needed and build a 128px JPEG thumbnail for + the ``image`` field. Uses Pillow when available, else ImageMagick ``convert``.""" import subprocess import tempfile @@ -706,7 +529,6 @@ class SimplexAdapter(BasePlatformAdapter): png_path = file_path thumb_uri = "" needs_png = p.suffix.lower() not in (".png", ".jpg", ".jpeg") - try: from PIL import Image import io @@ -728,10 +550,8 @@ class SimplexAdapter(BasePlatformAdapter): check=True, capture_output=True, timeout=30, stdin=subprocess.DEVNULL) with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp: tmp_path = tmp.name - subprocess.run( - ["convert", file_path, "-resize", "128x128", "-quality", "70", tmp_path], - check=True, capture_output=True, timeout=30, - ) + subprocess.run(["convert", file_path, "-resize", "128x128", "-quality", "70", tmp_path], + check=True, capture_output=True, timeout=30) with open(tmp_path, "rb") as f: thumb_uri = "data:image/jpg;base64," + base64.b64encode(f.read()).decode() os.remove(tmp_path) @@ -741,49 +561,38 @@ class SimplexAdapter(BasePlatformAdapter): return png_path, thumb_uri async def send_image( - self, chat_id: str, image_url: str, caption: Optional[str] = None, **kwargs - ) -> SendResult: + self, chat_id: str, image_url: str, caption: Optional[str] = None, **kwargs) -> SendResult: """Send an image. Supports ``file://`` URLs and ``http(s)://`` URLs.""" - from urllib.parse import unquote - if image_url.startswith("file://"): file_path = unquote(image_url[7:]) else: try: - from gateway.platforms.base import cache_image_from_url - file_path = await cache_image_from_url(image_url) except Exception as e: logger.warning("SimpleX: failed to download image: %s", e) return SendResult(success=False, error=str(e)) - if not file_path or not Path(file_path).exists(): return SendResult(success=False, error="Image file not found") - png_path, thumb_uri = self._prepare_image(file_path) # /_send addresses by numeric ID; /f only accepts display names. - item = {"filePath": png_path, - "msgContent": {"type": "image", "image": thumb_uri, "text": caption or ""}} + item = {"filePath": png_path, "msgContent": {"type": "image", "image": thumb_uri, "text": caption or ""}} return await self._send_items(chat_id, [item], "Failed to send image") async def send_image_file( self, chat_id: str, image_path: str, caption: Optional[str] = None, - reply_to: Optional[str] = None, **kwargs, - ) -> SendResult: + reply_to: Optional[str] = None, **kwargs) -> SendResult: """Send a local image file via SimpleX.""" return await self.send_image(chat_id, f"file://{image_path}", caption=caption, **kwargs) async def send_video( self, chat_id: str, video_path: str, caption: Optional[str] = None, - reply_to: Optional[str] = None, **kwargs, - ) -> SendResult: + reply_to: Optional[str] = None, **kwargs) -> SendResult: """Send a video file via SimpleX (as a file attachment).""" return await self.send_document(chat_id, video_path, caption=caption) async def send_document( self, chat_id: str, file_path: str, caption: Optional[str] = None, - filename: Optional[str] = None, **kwargs, - ) -> SendResult: + filename: Optional[str] = None, **kwargs) -> SendResult: """Send a document/file attachment.""" if not Path(file_path).exists(): return SendResult(success=False, error="File not found") @@ -792,15 +601,12 @@ class SimplexAdapter(BasePlatformAdapter): async def send_voice( self, chat_id: str, audio_path: str, caption: Optional[str] = None, - reply_to: Optional[str] = None, duration: int = 0, **kwargs, - ) -> SendResult: + reply_to: Optional[str] = None, duration: int = 0, **kwargs) -> SendResult: """Send an audio file as an inline SimpleX voice note (``msgContent.type == "voice"``).""" if not Path(audio_path).exists(): return SendResult(success=False, error="Voice file not found") - item = { - "msgContent": {"type": "voice", "text": caption or "", "duration": duration}, - "fileSource": {"filePath": audio_path}, - } + item = {"msgContent": {"type": "voice", "text": caption or "", "duration": duration}, + "fileSource": {"filePath": audio_path}} return await self._send_items(chat_id, [item], "Failed to send voice message") async def send_typing(self, chat_id: str, metadata=None) -> None: @@ -813,10 +619,6 @@ class SimplexAdapter(BasePlatformAdapter): return {"chat_id": chat_id, "type": "dm", "name": chat_id} -# --------------------------------------------------------------------------- -# Plugin entry-point hooks -# --------------------------------------------------------------------------- - def check_requirements() -> bool: """Plugin gate: require SIMPLEX_WS_URL AND the websockets package.""" if not _get_scoped_secret("SIMPLEX_WS_URL"): @@ -840,70 +642,48 @@ def is_connected(config) -> bool: def _env_enablement() -> Optional[dict]: - """Seed ``PlatformConfig.extra`` from env vars during gateway config load. - - Runs BEFORE adapter construction so ``gateway status`` reflects env-only - configuration. Returns ``None`` when SimpleX isn't minimally configured. - ``home_channel`` is turned into a ``HomeChannel`` by the core hook. - """ + """Seed ``PlatformConfig.extra`` from env BEFORE adapter construction so ``gateway status`` + reflects env-only setups. ``None`` when not minimally configured; ``home_channel`` becomes + a ``HomeChannel`` via the core hook.""" ws_url = _get_scoped_secret("SIMPLEX_WS_URL", "").strip() if not ws_url: return None seed: dict = {"ws_url": ws_url} - auto_accept = _get_scoped_secret("SIMPLEX_AUTO_ACCEPT", "").strip().lower() if auto_accept: seed["auto_accept"] = auto_accept not in {"0", "false", "no"} - group_allowed = _get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "").strip() if group_allowed: seed["group_allowed"] = group_allowed - home = _get_scoped_secret("SIMPLEX_HOME_CHANNEL", "").strip() if home: - seed["home_channel"] = { - "chat_id": home, - "name": _get_scoped_secret("SIMPLEX_HOME_CHANNEL_NAME", "").strip() or home, - } + seed["home_channel"] = {"chat_id": home, "name": _get_scoped_secret("SIMPLEX_HOME_CHANNEL_NAME", "").strip() or home} return seed async def _standalone_send( - pconfig, - chat_id: str, - message: str, - *, - thread_id: Optional[str] = None, - media_files: Optional[List[str]] = None, - force_document: bool = False, + pconfig, chat_id: str, message: str, *, + thread_id: Optional[str] = None, media_files: Optional[List[str]] = None, force_document: bool = False, ) -> Dict[str, Any]: - """Open an ephemeral WebSocket to the daemon, send, and close. - - Used by ``tools/send_message_tool`` when the gateway runner is not in this - process (e.g. ``hermes cron``). ``thread_id``/``force_document`` are - signature parity only; ``media_files`` is accepted but only the text body - is delivered — SimpleX file transfers need the daemon's filesystem-backed - flow, which an ephemeral connection cannot drive safely. - """ + """Ephemeral WebSocket send for ``tools/send_message_tool`` when the gateway runner is not in + this process (``hermes cron``). ``thread_id``/``force_document`` are signature parity only; + ``media_files`` is accepted but only the text body is delivered — SimpleX file transfers need + the daemon's filesystem-backed flow, which an ephemeral connection cannot drive safely.""" try: import websockets as _wsclient except ImportError: return {"error": "websockets not installed. Run: pip install websockets"} - extra = getattr(pconfig, "extra", {}) or {} ws_url = _get_scoped_secret("SIMPLEX_WS_URL") or extra.get("ws_url", "ws://127.0.0.1:5225") if not ws_url: return {"error": "SimpleX standalone send: SIMPLEX_WS_URL is required"} - try: payload = { "corrId": f"{_CORR_PREFIX}snd-{int(time.time() * 1000)}", - "cmd": _send_cmd(chat_id, [{"msgContent": {"type": "text", "text": message}}]), - } + "cmd": _send_cmd(chat_id, [{"msgContent": {"type": "text", "text": message}}])} async with _wsclient.connect(ws_url, open_timeout=10, close_timeout=5) as ws: await ws.send(json.dumps(payload)) - # Give the daemon a moment to process the command before closing. - await asyncio.sleep(0.5) + await asyncio.sleep(0.5) # let the daemon process the command before closing return {"success": True, "platform": "simplex", "chat_id": chat_id} except Exception as e: return {"error": f"SimpleX send failed: {e}"} @@ -911,14 +691,10 @@ async def _standalone_send( def interactive_setup() -> None: """Minimal stdin wizard for ``hermes setup gateway`` → SimpleX; writes ``~/.hermes/.env``.""" - print() - print("SimpleX Chat setup") - print("------------------") - print("Requirements:") - print(" 1. simplex-chat daemon running (e.g. `simplex-chat -p 5225`).") - print(" 2. Python package `websockets` installed (`pip install websockets`).") - print() - + print( + "\nSimpleX Chat setup\n------------------\nRequirements:\n" + " 1. simplex-chat daemon running (e.g. `simplex-chat -p 5225`).\n" + " 2. Python package `websockets` installed (`pip install websockets`).\n") try: from hermes_cli.config import get_env_value, save_env_value except ImportError: @@ -937,12 +713,9 @@ def interactive_setup() -> None: save_env_value(var, value) _prompt("SIMPLEX_WS_URL", "Daemon WebSocket URL (default ws://127.0.0.1:5225)") - _prompt("SIMPLEX_ALLOWED_USERS", - "Allowed contactIds or display names (comma-separated; blank=skip)") - _prompt("SIMPLEX_GROUP_ALLOWED", - "Allowed group IDs (comma-separated, or '*' for any; blank=disable groups)") - _prompt("SIMPLEX_AUTO_ACCEPT", - "Auto-accept incoming contact requests? (true/false, default true)") + _prompt("SIMPLEX_ALLOWED_USERS", "Allowed contactIds or display names (comma-separated; blank=skip)") + _prompt("SIMPLEX_GROUP_ALLOWED", "Allowed group IDs (comma-separated, or '*' for any; blank=disable groups)") + _prompt("SIMPLEX_AUTO_ACCEPT", "Auto-accept incoming contact requests? (true/false, default true)") _prompt("SIMPLEX_HOME_CHANNEL", "Home channel contact/group ID (or empty)") print("Done. Make sure the simplex-chat daemon is running before starting the gateway.") @@ -977,6 +750,4 @@ def register(ctx) -> None: "hard message length limit, but keep responses conversational. " "You can attach native images, voice notes, and arbitrary " "files; the adapter handles MEDIA: tags by sending them " - "as inline voice notes (audio extensions) or documents." - ), - ) + "as inline voice notes (audio extensions) or documents.")) diff --git a/plugins/platforms/teams/adapter.py b/plugins/platforms/teams/adapter.py index 111f550247..df1c5fe68e 100644 --- a/plugins/platforms/teams/adapter.py +++ b/plugins/platforms/teams/adapter.py @@ -1,21 +1,10 @@ -""" -Microsoft Teams platform adapter: microsoft-teams-apps SDK for auth/activity -processing, an aiohttp webhook server for inbound, ``App.send()`` for proactive sends. +"""Microsoft Teams adapter: microsoft-teams-apps SDK for auth/activity processing, an aiohttp +webhook server for inbound, ``App.send()`` for proactive sends. -Requires: - the ``teams`` extra (auto-installed by the gateway on first start, or - manually: ``/bin/pip install microsoft-teams-apps aiohttp``) - TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID env vars - -Configuration in config.yaml: - platforms: - teams: - enabled: true - extra: - client_id: "your-client-id" # or TEAMS_CLIENT_ID env var - client_secret: "your-secret" # or TEAMS_CLIENT_SECRET env var - tenant_id: "your-tenant-id" # or TEAMS_TENANT_ID env var - port: 3978 # or TEAMS_PORT env var +Requires the ``teams`` extra (auto-installed by the gateway on first start, or +``/bin/pip install microsoft-teams-apps aiohttp``) and credentials via env +(TEAMS_CLIENT_ID / TEAMS_CLIENT_SECRET / TEAMS_TENANT_ID, optional TEAMS_PORT) or +``platforms.teams.extra`` in config.yaml (``client_id`` / ``client_secret`` / ``tenant_id`` / ``port``). """ from __future__ import annotations @@ -33,7 +22,6 @@ from urllib.parse import urlparse try: from aiohttp import web - AIOHTTP_AVAILABLE = True except ImportError: AIOHTTP_AVAILABLE = False @@ -41,20 +29,16 @@ except ImportError: def _probe_teams_sdk_available() -> bool: - """True when ``microsoft_teams.apps`` is on sys.path, without importing it. - - Import is deferred: ``microsoft_teams.apps.app`` loads a cwd ``.env`` at import - (#62935); ``check_teams_requirements()`` binds symbols behind a dotenv no-op. - Sibling packages share the ``microsoft_teams`` namespace, so probe the parent - first — ``find_spec("microsoft_teams.apps")`` raises on 3.11+ if it is absent. - """ + """True when ``microsoft_teams.apps`` is on sys.path, without importing it: the SDK loads a cwd + ``.env`` at import, so ``check_teams_requirements()`` binds symbols behind a dotenv no-op. + Sibling packages share the namespace, so probe the parent first — ``find_spec`` of the child + raises on 3.11+ if the parent is absent.""" try: if importlib.util.find_spec("microsoft_teams") is None: return False return importlib.util.find_spec("microsoft_teams.apps") is not None except (ValueError, ModuleNotFoundError, ImportError): - # Test stubs may inject a module without ``__spec__``. - return "microsoft_teams.apps" in sys.modules + return "microsoft_teams.apps" in sys.modules # test stubs may lack ``__spec__`` TEAMS_SDK_AVAILABLE = _probe_teams_sdk_available() @@ -68,13 +52,7 @@ HttpMethod = str # type: ignore[assignment,misc] from gateway.config import Platform, PlatformConfig from gateway.platforms.helpers import MessageDeduplicator from gateway.platforms.base import ( - gateway_trust_env, - BasePlatformAdapter, - MessageEvent, - MessageType, - SendResult, - cache_image_from_url, - cache_media_bytes, + gateway_trust_env, BasePlatformAdapter, MessageEvent, MessageType, SendResult, cache_image_from_url, cache_media_bytes, ) from gateway.platforms._shared import coerce_port, get_scoped_secret as _get_scoped_secret from plugins.platforms.teams.summary_writer import TeamsSummaryWriter # noqa: F401 — re-exported for teams_pipeline @@ -82,28 +60,19 @@ from plugins.platforms.teams.summary_writer import TeamsSummaryWriter # noqa: F logger = logging.getLogger(__name__) _DEFAULT_PORT = 3978 -# Bot Framework activities are JSON well under 1 MiB; bound oversized/chunked bodies. -_MAX_BODY_BYTES = 1_048_576 -# ``None`` host → aiohttp binds IPv4 + IPv6 (a hardcoded "0.0.0.0" was unreachable -# on IPv6-only networks such as Fly.io 6PN). Pin via TEAMS_HOST or extra.host. +_MAX_BODY_BYTES = 1_048_576 # Bot Framework activities are JSON well under 1 MiB +# ``None`` host → aiohttp binds IPv4 + IPv6 ("0.0.0.0" was unreachable on IPv6-only +# networks such as Fly.io 6PN). Pin via TEAMS_HOST or extra.host. _DEFAULT_HOST = None _WEBHOOK_PATH = "/api/messages" - -# Bot Framework default service URL. Regional/government tenants may need a -# different host (e.g. ``https://smba.infra.gov.teams.microsoft.us/``) via -# ``TEAMS_SERVICE_URL`` or ``extra['service_url']``. +# Regional/government tenants override via ``TEAMS_SERVICE_URL`` / ``extra['service_url']``. _DEFAULT_TEAMS_SERVICE_URL = "https://smba.trafficmanager.net/teams/" - -# Bot Framework hosts that may receive a freshly minted bearer token (blocks SSRF / -# token exfiltration via a tampered env var). Exact match only: any Azure customer can -# register ``.trafficmanager.net``. New regions are allowlist additions. +# Hosts that may receive a freshly minted bearer token (blocks SSRF / token exfiltration via a +# tampered env var). Exact match only: any Azure customer can register ``.trafficmanager.net``. _ALLOWED_TEAMS_SERVICE_HOSTS = frozenset({"smba.trafficmanager.net", "smba.infra.gov.teams.microsoft.us"}) - -# Conservative Bot Framework conversation-ID charset (digits, colons, hyphens, -# dots, '@', ``thread.skype`` / ``thread.tacv2`` suffixes) so a hostile value -# cannot path-traverse out of ``/v3/conversations//activities``. +# Conservative conversation-ID charset (``thread.skype`` / ``thread.tacv2`` suffixes included) so a +# hostile value cannot path-traverse out of ``/v3/conversations//activities``. _TEAMS_CONV_ID_RE = re.compile(r"^[A-Za-z0-9:@\-_.]+$") - _BF_TOKEN_SCOPE = "https://api.botframework.com/.default" @@ -111,12 +80,7 @@ def _bf_token_request(tenant_id: str, client_id: str, client_secret: str) -> tup """(token URL, client-credentials form) for a Bot Framework bearer token.""" return ( f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token", - { - "grant_type": "client_credentials", - "client_id": client_id, - "client_secret": client_secret, - "scope": _BF_TOKEN_SCOPE, - }, + {"grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, "scope": _BF_TOKEN_SCOPE}, ) @@ -124,9 +88,7 @@ def _is_allowed_https_host(url: str, *, check_port: bool = False) -> bool: """https + host in ``_ALLOWED_TEAMS_SERVICE_HOSTS`` (+ default port when asked).""" try: parsed = urlparse(url) - if parsed.scheme != "https": - return False - if check_port and parsed.port not in (None, 443): + if parsed.scheme != "https" or (check_port and parsed.port not in (None, 443)): return False return parsed.hostname in _ALLOWED_TEAMS_SERVICE_HOSTS except Exception: @@ -146,19 +108,15 @@ def _validate_teams_service_url(raw: str) -> Optional[str]: class _AiohttpBridgeAdapter: - """HttpServerAdapter bridging SDK route registrations into our aiohttp app. - - Without it ``App()`` unconditionally imports fastapi/uvicorn and allocates a ``FastAPI()``. - """ + """HttpServerAdapter bridging SDK route registrations into our aiohttp app; without it + ``App()`` unconditionally imports fastapi/uvicorn and allocates a ``FastAPI()``.""" def __init__(self, aiohttp_app: "web.Application"): self._aiohttp_app = aiohttp_app def register_route(self, method: "HttpMethod", path: str, handler: "HttpRouteHandler") -> None: async def _aiohttp_handler(request: "web.Request") -> "web.Response": - body = await request.json() - headers = dict(request.headers) - result: "HttpResponse" = await handler(HttpRequest(body=body, headers=headers)) + result: "HttpResponse" = await handler(HttpRequest(body=await request.json(), headers=dict(request.headers))) status = result.get("status", 200) resp_body = result.get("body") if resp_body is not None: @@ -188,8 +146,7 @@ def _credentials(config) -> tuple[str, str, str]: return ( os.getenv("TEAMS_CLIENT_ID") or extra.get("client_id", ""), _get_scoped_secret("TEAMS_CLIENT_SECRET") or extra.get("client_secret", ""), - os.getenv("TEAMS_TENANT_ID") or extra.get("tenant_id", ""), - ) + os.getenv("TEAMS_TENANT_ID") or extra.get("tenant_id", "")) def validate_config(config) -> bool: @@ -203,11 +160,9 @@ def is_connected(config) -> bool: def _env_enablement() -> dict | None: - """Seed ``PlatformConfig.extra`` from env vars before adapter construction. - - Lets ``gateway status`` reflect env-only setups without the SDK. ``None`` when - not minimally configured; ``home_channel`` becomes a ``HomeChannel`` via the core hook. - """ + """Seed ``PlatformConfig.extra`` from env before adapter construction so ``gateway status`` reflects + env-only setups without the SDK. ``None`` when not minimally configured; ``home_channel`` becomes a + ``HomeChannel`` via the core hook.""" client_id = os.getenv("TEAMS_CLIENT_ID", "").strip() client_secret = _get_scoped_secret("TEAMS_CLIENT_SECRET", "").strip() tenant_id = os.getenv("TEAMS_TENANT_ID", "").strip() @@ -230,18 +185,14 @@ async def _standalone_send( pconfig, chat_id: str, message: str, *, thread_id: Optional[str] = None, media_files: Optional[list] = None, force_document: bool = False, ) -> Dict[str, Any]: - """Acquire a Bot Framework bearer token and POST a single message activity. - - Used by ``send_message_tool._send_via_adapter`` when the gateway runner is - not in this process (``hermes cron``). ``TEAMS_SERVICE_URL`` is allowlisted - and ``chat_id`` charset-checked (SSRF/path traversal). ``media_files`` / - ``force_document`` are signature parity only — this path is text-only. - """ + """Acquire a Bot Framework bearer token and POST a single message activity; used by + ``send_message_tool._send_via_adapter`` when the gateway runner is not in this process + (``hermes cron``). ``TEAMS_SERVICE_URL`` is allowlisted and ``chat_id`` charset-checked + (SSRF/path traversal). ``media_files`` / ``force_document`` are signature parity only — text-only.""" extra = getattr(pconfig, "extra", {}) or {} client_id, client_secret, tenant_id = _credentials(pconfig) if not (client_id and client_secret and tenant_id): return {"error": "Teams standalone send: TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID are all required"} - raw_service_url = os.getenv("TEAMS_SERVICE_URL") or extra.get("service_url", "") or _DEFAULT_TEAMS_SERVICE_URL service_url = _validate_teams_service_url(raw_service_url) if service_url is None: @@ -253,13 +204,10 @@ async def _standalone_send( return {"error": "Teams standalone send: chat_id contains characters outside the Bot Framework conversation ID set"} if not _TEAMS_CONV_ID_RE.match(tenant_id): return {"error": "Teams standalone send: TEAMS_TENANT_ID contains characters outside the expected set"} - token_url, token_form = _bf_token_request(tenant_id, client_id, client_secret) activities_url = f"{service_url}v3/conversations/{chat_id}/activities" - if not AIOHTTP_AVAILABLE: return {"error": "Teams standalone send: aiohttp not installed"} - try: import aiohttp as _aiohttp @@ -267,9 +215,7 @@ async def _standalone_send( per_request_timeout = _aiohttp.ClientTimeout(total=15.0) async with _aiohttp.ClientSession(trust_env=gateway_trust_env()) as session: async with session.post( - token_url, - data=token_form, - headers={"Content-Type": "application/x-www-form-urlencoded"}, + token_url, data=token_form, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=per_request_timeout, ) as token_resp: if token_resp.status >= 400: @@ -279,11 +225,8 @@ async def _standalone_send( access_token = token_payload.get("access_token") if not access_token: return {"error": "Teams standalone send: token response missing access_token"} - - activity = {"type": "message", "text": message, "textFormat": "markdown"} async with session.post( - activities_url, - json=activity, + activities_url, json={"type": "message", "text": message, "textFormat": "markdown"}, headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, timeout=per_request_timeout, ) as send_resp: @@ -309,17 +252,13 @@ _SDK_IMPORTS = { "microsoft_teams.api.models.adaptive_card": ("AdaptiveCardActionCardResponse", "AdaptiveCardActionMessageResponse"), "microsoft_teams.api.models.invoke_response": ("InvokeResponse", "AdaptiveCardInvokeResponse"), "microsoft_teams.apps.http.adapter": ("HttpMethod", "HttpRequest", "HttpResponse", "HttpRouteHandler"), - "microsoft_teams.cards": ("AdaptiveCard", "ExecuteAction", "TextBlock"), -} + "microsoft_teams.cards": ("AdaptiveCard", "ExecuteAction", "TextBlock")} @contextmanager def _suppress_third_party_dotenv() -> Iterator[None]: - """No-op ``dotenv.load_dotenv`` while importing the Teams SDK (#62935). - - ``microsoft_teams.apps.app`` loads a cwd-discovered ``.env`` at import, - mutating process-global ``os.environ``. Hermes owns dotenv loading. - """ + """No-op ``dotenv.load_dotenv`` while importing the Teams SDK: ``microsoft_teams.apps.app`` loads a + cwd-discovered ``.env`` at import, mutating process-global ``os.environ``. Hermes owns dotenv loading.""" try: import dotenv as _dotenv except ImportError: @@ -336,12 +275,9 @@ def _suppress_third_party_dotenv() -> Iterator[None]: def check_teams_requirements() -> bool: - """ACTIVE lazy-installer (registry ``ensure_deps_fn``): install the SDK on first use. - - Rebinds all module-level SDK globals on success; True once the SDK and - aiohttp are importable. Gate on ``App is not None`` — ``TEAMS_SDK_AVAILABLE`` - is only a find_spec probe and can be True before any import ran. - """ + """ACTIVE lazy-installer (registry ``ensure_deps_fn``): install the SDK on first use and rebind + the module-level SDK globals. Gate on ``App is not None`` — ``TEAMS_SDK_AVAILABLE`` is only a + find_spec probe and can be True before any import ran.""" if App is not None and AIOHTTP_AVAILABLE: return True @@ -361,7 +297,6 @@ def check_teams_requirements() -> bool: return bindings from tools.lazy_deps import ensure_and_bind - return ensure_and_bind("platform.teams", _import, globals(), prompt=False) @@ -370,14 +305,28 @@ _CHAT_TYPES = {"personal": "dm", "groupChat": "group", "channel": "channel"} # injection gates strictly on MessageType.DOCUMENT (same precedence as Email/Signal). _MEDIA_KIND_PRECEDENCE = ( ("document", MessageType.DOCUMENT), ("image", MessageType.PHOTO), - ("video", MessageType.VIDEO), ("audio", MessageType.AUDIO), -) + ("video", MessageType.VIDEO), ("audio", MessageType.AUDIO)) _APPROVAL_CHOICES = {"approve_once": "once", "approve_session": "session", "approve_always": "always", "deny": "deny"} _APPROVAL_LABELS = { "once": "✅ Allowed (once)", "session": "✅ Allowed (session)", "always": "✅ Always allowed", "deny": "❌ Denied", } +def _truncate(text: str, limit: int) -> str: + return text[:limit] + "..." if len(text) > limit else text + + +def _approval_body(cmd: str, desc: str, *, always: bool = False) -> list: + """Adaptive Card body blocks for an approval prompt; unless ``always``, empty ``cmd``/``desc`` omit their blocks.""" + body = [] + if cmd or always: + body.append(TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder")) + body.append(TextBlock(text=f"```\n{cmd}\n```", wrap=True)) + if desc or always: + body.append(TextBlock(text=f"Reason: {desc}", wrap=True, isSubtle=True)) + return body + + class TeamsAdapter(BasePlatformAdapter): """Microsoft Teams adapter using the microsoft-teams-apps SDK.""" @@ -390,24 +339,21 @@ class TeamsAdapter(BasePlatformAdapter): self._client_id = extra.get("client_id") or os.getenv("TEAMS_CLIENT_ID", "") self._client_secret = extra.get("client_secret") or _get_scoped_secret("TEAMS_CLIENT_SECRET", "") self._tenant_id = extra.get("tenant_id") or os.getenv("TEAMS_TENANT_ID", "") - # (token, expiry monotonic ts) for connector attachment auth; refreshed - # under _bf_token_lock so concurrent attachments can't stampede the STS. + # (token, expiry monotonic ts) for connector attachment auth; refreshed under + # _bf_token_lock so concurrent attachments can't stampede the STS. self._bf_token_cache: Optional[tuple] = None self._bf_token_lock: Optional[asyncio.Lock] = None self._port = coerce_port(extra.get("port") or os.getenv("TEAMS_PORT", str(_DEFAULT_PORT)), _DEFAULT_PORT) - # Falsy host (unset/"") collapses to the dual-stack default (None). - _raw_host = extra.get("host") or os.getenv("TEAMS_HOST", "") or _DEFAULT_HOST + _raw_host = extra.get("host") or os.getenv("TEAMS_HOST", "") or _DEFAULT_HOST # falsy → dual-stack None self._host: Optional[str] = str(_raw_host) if _raw_host else None self._app: Optional["App"] = None self._runner: Optional["web.AppRunner"] = None self._dedup = MessageDeduplicator(max_size=1000) - # chat_id → ConversationReference from incoming messages, so proactive - # cards use the right conversation type (personal/group/channel). + # chat_id → ConversationReference so proactive cards use the right conversation type. self._conv_refs: Dict[str, Any] = {} async def connect(self, *, is_reconnect: bool = False) -> bool: - # create_adapter() already ran the installer, but reconnect paths reach - # here too — re-run to bind SDK globals. Gate on App, not the find_spec flag. + # Reconnect paths reach here without create_adapter()'s installer — re-run to bind SDK globals. check_teams_requirements() pip = f"{sys.executable} -m pip install" for failed, code, message in ( @@ -415,8 +361,7 @@ class TeamsAdapter(BasePlatformAdapter): f"microsoft-teams-apps could not be installed. Run: {pip} microsoft-teams-apps"), (not AIOHTTP_AVAILABLE, "MISSING_SDK", f"aiohttp not installed. Run: {pip} aiohttp"), (not self._client_id or not self._client_secret or not self._tenant_id, "MISSING_CREDENTIALS", - "TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID are all required"), - ): + "TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID are all required")): if failed: self._set_fatal_error(code, message, retryable=False) return False @@ -425,18 +370,12 @@ class TeamsAdapter(BasePlatformAdapter): # aiohttp app first — the bridge adapter wires SDK routes into it. aiohttp_app = web.Application(client_max_size=_MAX_BODY_BYTES) aiohttp_app.router.add_get("/health", lambda _: web.Response(text="ok")) - self._app = App( - client_id=self._client_id, - client_secret=self._client_secret, - tenant_id=self._tenant_id, + client_id=self._client_id, client_secret=self._client_secret, tenant_id=self._tenant_id, http_server_adapter=_AiohttpBridgeAdapter(aiohttp_app), - client=ClientOptions(headers={"User-Agent": "Hermes"}), - ) - - # Handlers (ours, then plugin-registered on_* decorators) must be - # wired before initialize(), which registers POST /api/messages - # on aiohttp_app via the bridge's register_route(). + client=ClientOptions(headers={"User-Agent": "Hermes"})) + # Handlers (ours, then plugin on_* decorators) must be wired before initialize(), + # which registers POST /api/messages on aiohttp_app via the bridge's register_route(). @self._app.on_message async def _handle_message(ctx: ActivityContext[MessageActivity]): await self._on_message(ctx) @@ -449,20 +388,15 @@ class TeamsAdapter(BasePlatformAdapter): self._wire_plugin_handlers(self._app) await self._app.initialize() - self._runner = web.AppRunner(aiohttp_app) await self._runner.setup() site = web.TCPSite(self._runner, self._host, self._port) await site.start() - self._running = True self._mark_connected() logger.info( "[teams] Webhook server listening on %s:%d%s", - self._host or "* (all interfaces, IPv4+IPv6)", - self._port, - _WEBHOOK_PATH, - ) + self._host or "* (all interfaces, IPv4+IPv6)", self._port, _WEBHOOK_PATH) return True except Exception as e: self._set_fatal_error("CONNECT_FAILED", f"Teams connection failed: {e}", retryable=True) @@ -479,11 +413,9 @@ class TeamsAdapter(BasePlatformAdapter): logger.info("[teams] Disconnected") async def _get_botframework_token(self) -> str: - """Bot Framework bearer token (client credentials), cached until ~5 min before expiry. - - Connector attachments are NOT pre-authenticated, unlike SharePoint downloadUrls. - Lock is created lazily: ``asyncio.Lock()`` in __init__ may bind the wrong loop. - """ + """Bot Framework bearer token (client credentials), cached until ~5 min before expiry; connector + attachments are NOT pre-authenticated, unlike SharePoint downloadUrls. The lock is created lazily + because ``asyncio.Lock()`` in __init__ may bind the wrong loop.""" import time import httpx @@ -506,11 +438,8 @@ class TeamsAdapter(BasePlatformAdapter): return token async def _fetch_attachment_bytes(self, url: str, timeout: float = 30.0) -> bytes: - """Download attachment bytes with SSRF protection. - - Bot Framework connector URLs get the bot's bearer token; redirects and body - size go through the shared guards (as the cache_*_from_url helpers). - """ + """Download attachment bytes with SSRF protection. Connector URLs get the bot's bearer token; + redirects and body size go through the shared guards (as the cache_*_from_url helpers).""" from tools.url_safety import create_ssrf_safe_async_client, is_safe_url from gateway.platforms.base import _ssrf_redirect_guard, _read_httpx_body_with_limit @@ -539,16 +468,13 @@ class TeamsAdapter(BasePlatformAdapter): msg_id = getattr(activity, "id", None) if msg_id and self._dedup.is_duplicate(msg_id): return - # Cache the conversation reference for proactive sends (approval cards, etc.) - conv_id = getattr(activity.conversation, "id", None) - if conv_id: + conv = activity.conversation + conv_id = getattr(conv, "id", None) + if conv_id: # cache the conversation reference for proactive sends (approval cards, etc.) self._conv_refs[conv_id] = ctx.conversation_ref - text = activity.text if hasattr(activity, "text") and activity.text else "" if "" in text: # strip the BotName tags Teams prepends for @mentions text = re.sub(r"[^<]*\s*", "", text).strip() - - conv = activity.conversation from_account = activity.from_ user_id = getattr(from_account, "aad_object_id", None) or getattr(from_account, "id", "") source = self.build_source( @@ -557,92 +483,79 @@ class TeamsAdapter(BasePlatformAdapter): chat_type=_CHAT_TYPES.get(getattr(conv, "conversation_type", None) or "", "dm"), user_id=str(user_id), user_name=getattr(from_account, "name", None) or "", - guild_id=getattr(conv, "tenant_id", None) or self._tenant_id, - ) - - media_urls: list = [] - media_types: list = [] - media_kinds: list = [] - - def _add_cached(cached: Any, kind: Optional[str] = None) -> bool: - if not cached: - return False - media_urls.append(cached.path) - media_types.append(cached.media_type) - media_kinds.append(kind or cached.kind) - return True - + guild_id=getattr(conv, "tenant_id", None) or self._tenant_id) + media: list = [] # (path, media_type, kind) for att in getattr(activity, "attachments", None) or []: - content_url = getattr(att, "content_url", None) - content_type = (getattr(att, "content_type", None) or "").lower() - att_name = getattr(att, "name", None) or "" - # Skip non-file payloads: Teams mirrors the message body as a - # text/html attachment, and cards arrive as application/vnd.microsoft.card.* - if content_type in ("text/html", "text/plain") and not content_url: - continue - if content_type.startswith("application/vnd.microsoft.card"): - continue - - if content_type == "application/vnd.microsoft.teams.file.download.info": - # Consent-free download: content carries a pre-authed SharePoint downloadUrl + file type. - content = getattr(att, "content", None) - if not isinstance(content, dict): - content = getattr(content, "__dict__", None) or {} - download_url = content.get("downloadUrl") or content.get("download_url") - file_type = (content.get("fileType") or content.get("file_type") or "").lstrip(".") - if not download_url: - continue - filename = att_name or (f"document.{file_type}" if file_type else "document") - try: - data = await self._fetch_attachment_bytes(download_url) - if not _add_cached(cache_media_bytes(data, filename=filename, mime_type="")): - logger.warning("[teams] Unsupported document type for attachment '%s', skipping", filename) - except Exception as e: - logger.warning("[teams] Failed to cache file attachment '%s': %s", filename, e) - continue - - if content_url and content_type.startswith("image/"): - try: - if _is_botframework_attachment_url(content_url): - # Connector URL needs the bot's bearer token; the generic cache helper sends none. - data = await self._fetch_attachment_bytes(content_url) - ext = content_type.split("/")[-1].split(";")[0] or "png" - cached_m = cache_media_bytes(data, filename=att_name or f"image.{ext}", mime_type=content_type) - if not _add_cached(cached_m, kind="image"): - logger.warning( - "[teams] Bot Framework attachment '%s' returned data that failed image validation, skipping", - att_name or content_url, - ) - else: - cached = await cache_image_from_url(content_url) - if cached: - media_urls.append(cached) - media_types.append(content_type) - media_kinds.append("image") - except Exception as e: - logger.warning("[teams] Failed to cache image attachment: %s", e) - continue - - if content_url: # direct-URL non-image attachment (video/audio/document) - try: - data = await self._fetch_attachment_bytes(content_url) - _add_cached(cache_media_bytes(data, filename=att_name, mime_type=content_type)) - except Exception as e: - logger.warning( - "[teams] Failed to cache attachment '%s' (%s): %s", att_name or content_url, content_type, e - ) - + cached = await self._cache_attachment(att) + if cached: + media.append(cached) + media_kinds = [kind for _, _, kind in media] msg_type = next((t for kind, t in _MEDIA_KIND_PRECEDENCE if kind in media_kinds), MessageType.TEXT) - await self.handle_message(MessageEvent( - text=text, source=source, message_type=msg_type, - media_urls=media_urls, media_types=media_types, message_id=msg_id, - )) + text=text, source=source, message_type=msg_type, message_id=msg_id, + media_urls=[path for path, _, _ in media], media_types=[mt for _, mt, _ in media])) + + async def _cache_attachment(self, att: Any) -> Optional[tuple]: + """Download + cache one inbound attachment → ``(path, media_type, kind)`` or ``None``.""" + content_url = getattr(att, "content_url", None) + content_type = (getattr(att, "content_type", None) or "").lower() + att_name = getattr(att, "name", None) or "" + # Skip non-file payloads: Teams mirrors the message body as a text/html attachment, + # and cards arrive as application/vnd.microsoft.card.* + if content_type in ("text/html", "text/plain") and not content_url: + return None + if content_type.startswith("application/vnd.microsoft.card"): + return None + if content_type == "application/vnd.microsoft.teams.file.download.info": + # Consent-free download: content carries a pre-authed SharePoint downloadUrl + file type. + content = getattr(att, "content", None) + if not isinstance(content, dict): + content = getattr(content, "__dict__", None) or {} + download_url = content.get("downloadUrl") or content.get("download_url") + file_type = (content.get("fileType") or content.get("file_type") or "").lstrip(".") + if not download_url: + return None + 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="") + if not cached: + logger.warning("[teams] Unsupported document type for attachment '%s', skipping", filename) + return None + return cached.path, cached.media_type, cached.kind + except Exception as e: + logger.warning("[teams] Failed to cache file attachment '%s': %s", filename, e) + return None + if content_url and content_type.startswith("image/"): + try: + if _is_botframework_attachment_url(content_url): + # Connector URL needs the bot's bearer token; the generic cache helper sends none. + data = await self._fetch_attachment_bytes(content_url) + ext = content_type.split("/")[-1].split(";")[0] or "png" + cached = cache_media_bytes(data, filename=att_name or f"image.{ext}", mime_type=content_type) + if not cached: + logger.warning( + "[teams] Bot Framework attachment '%s' returned data that failed image validation, skipping", + att_name or content_url) + return None + return cached.path, cached.media_type, "image" + path = await cache_image_from_url(content_url) + return (path, content_type, "image") if path else None + except Exception as e: + logger.warning("[teams] Failed to cache image attachment: %s", e) + return None + if content_url: # direct-URL non-image attachment (video/audio/document) + try: + data = await self._fetch_attachment_bytes(content_url) + cached = cache_media_bytes(data, filename=att_name, mime_type=content_type) + return (cached.path, cached.media_type, cached.kind) if cached else None + except Exception as e: + logger.warning("[teams] Failed to cache attachment '%s' (%s): %s", att_name or content_url, content_type, e) + return None async def _send_card(self, chat_id: str, card: "AdaptiveCard") -> "Any": """Send an AdaptiveCard, using a stored ConversationReference when available.""" from microsoft_teams.api import MessageActivityInput - if not self._app: return None return await self._send_via_conv_ref(chat_id, MessageActivityInput().add_card(card), card) @@ -671,24 +584,20 @@ class TeamsAdapter(BasePlatformAdapter): """Handle an Adaptive Card Action.Execute button click.""" from tools.approval import resolve_gateway_approval, has_blocking_approval - action = ctx.activity.value.action - data = action.data or {} + data = ctx.activity.value.action.data or {} hermes_action = data.get("hermes_action", "") session_key = data.get("session_key", "") if not hermes_action or not session_key: return self._invoke_message("Unknown action.") - # Default-deny: approval clicks require TEAMS_ALLOWED_USERS or an explicit - # TEAMS_ALLOW_ALL_USERS=true opt-in, else anyone who can message the bot - # could approve dangerous commands. + # TEAMS_ALLOW_ALL_USERS=true opt-in, else anyone who can message the bot could approve. allowed_csv = os.getenv("TEAMS_ALLOWED_USERS", "").strip() allow_all = os.getenv("TEAMS_ALLOW_ALL_USERS", "").strip().lower() in {"1", "true", "yes"} if not allow_all: if not allowed_csv: logger.warning( "[teams] card action rejected: TEAMS_ALLOWED_USERS not configured " - "and TEAMS_ALLOW_ALL_USERS not set — default deny" - ) + "and TEAMS_ALLOW_ALL_USERS not set — default deny") return self._invoke_message("⛔ Approval buttons require TEAMS_ALLOWED_USERS to be configured.") from_account = ctx.activity.from_ clicker_id = getattr(from_account, "aad_object_id", None) or getattr(from_account, "id", "") @@ -702,36 +611,20 @@ class TeamsAdapter(BasePlatformAdapter): return self._invoke_message("Unknown action.") if not has_blocking_approval(session_key): return self._invoke_card([TextBlock(text="⚠️ Approval already resolved or expired.", wrap=True)]) - resolve_gateway_approval(session_key, choice) - - cmd = data.get("cmd", "") - desc = data.get("desc", "") - body = [] - if cmd: - body.append(TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder")) - body.append(TextBlock(text=f"```\n{cmd}\n```", wrap=True)) - if desc: - body.append(TextBlock(text=f"Reason: {desc}", wrap=True, isSubtle=True)) + body = _approval_body(data.get("cmd", ""), data.get("desc", "")) body.append(TextBlock(text=_APPROVAL_LABELS[choice], wrap=True, weight="Bolder")) return self._invoke_card(body) async def send_exec_approval( self, chat_id: str, command: str, session_key: str, description: str = "dangerous command", metadata: Optional[Dict[str, Any]] = None, allow_permanent: bool = True, allow_session: bool = True, - smart_denied: bool = False, - ) -> SendResult: + smart_denied: bool = False) -> SendResult: """Send an Adaptive Card approval prompt with Allow/Deny buttons.""" if not self._app: return SendResult(success=False, error="Teams app not initialized") - - cmd_preview = command[:2000] + "..." if len(command) > 2000 else command # Button data carries a truncated cmd — just enough to reconstruct the card body. - btn_data_base = { - "session_key": session_key, - "cmd": command[:200] + "..." if len(command) > 200 else command, - "desc": description, - } + btn_data_base = {"session_key": session_key, "cmd": _truncate(command, 200), "desc": description} def _action(title: str, hermes_action: str, **kw) -> "ExecuteAction": return ExecuteAction( @@ -744,19 +637,13 @@ class TeamsAdapter(BasePlatformAdapter): if allow_permanent: actions.append(_action("Always Allow", "approve_always")) actions.append(_action("Deny", "deny", style="destructive")) - body = [ - TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"), - TextBlock(text=f"```\n{cmd_preview}\n```", wrap=True), - TextBlock(text=f"Reason: {description}", wrap=True, isSubtle=True), - ] + body = _approval_body(_truncate(command, 2000), description, always=True) if smart_denied: body.append(TextBlock(text="Smart DENY: owner override applies to this one operation only.", wrap=True)) card = AdaptiveCard().with_version("1.4").with_body(body).with_actions(actions) - try: result = await self._send_card(chat_id, card) - message_id = getattr(result, "id", None) if result else None - return SendResult(success=True, message_id=message_id) + return SendResult(success=True, message_id=getattr(result, "id", None) if result else None) except Exception as e: logger.error("[teams] send_exec_approval failed: %s", e, exc_info=True) return SendResult(success=False, error=str(e), retryable=True) @@ -766,7 +653,6 @@ class TeamsAdapter(BasePlatformAdapter): ) -> SendResult: if not self._app: return SendResult(success=False, error="Teams app not initialized") - last_message_id = None for chunk in self.truncate_message(self.format_message(content)): try: @@ -774,8 +660,7 @@ class TeamsAdapter(BasePlatformAdapter): try: result = await self._app.reply(chat_id, reply_to, chunk) except Exception as reply_err: - # Group chats 400 on threaded sends; the SDK has no typed - # HTTP errors, so fall back on any exception. + # Group chats 400 on threaded sends; the SDK has no typed HTTP errors → fall back on any. logger.debug("Teams reply() failed, falling back to flat send: %s", reply_err) result = await self._app.send(chat_id, chunk) else: @@ -797,11 +682,8 @@ class TeamsAdapter(BasePlatformAdapter): self, chat_id: str, source: str, default_mime: str, caption: Optional[str] = None, media_label: str = "media" ) -> SendResult: """Send any media file/URL as a Teams attachment (shared by send_image/video/voice/document). - - Remote ``http(s)://`` URLs are attached by reference; local paths - (optional ``file://`` prefix) are base64-encoded into a data URI. - MIME type is guessed from the path, falling back to ``default_mime``. - """ + Remote ``http(s)://`` URLs are attached by reference; local paths (optional ``file://`` prefix) + are base64-encoded into a data URI. MIME is guessed from the path, else ``default_mime``.""" if not self._app: return SendResult(success=False, error="Teams app not initialized") try: @@ -809,7 +691,7 @@ class TeamsAdapter(BasePlatformAdapter): import mimetypes from microsoft_teams.api import Attachment, MessageActivityInput - if source.startswith("http://") or source.startswith("https://"): + if source.startswith(("http://", "https://")): content_url = source mime_type = mimetypes.guess_type(source.split("?")[0])[0] or default_mime else: @@ -829,8 +711,7 @@ class TeamsAdapter(BasePlatformAdapter): 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: + metadata: Optional[Dict[str, Any]] = None) -> SendResult: return await self._send_media_attachment(chat_id, image_url, "image/png", caption=caption, media_label="image") async def send_image_file( @@ -840,14 +721,12 @@ class TeamsAdapter(BasePlatformAdapter): 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: + metadata: Optional[Dict[str, Any]] = None, **kwargs) -> SendResult: return await self._send_media_attachment(chat_id, video_path, "video/mp4", caption=caption, media_label="video") 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: + metadata: Optional[Dict[str, Any]] = None, **kwargs) -> SendResult: return await self._send_media_attachment(chat_id, audio_path, "audio/mpeg", caption=caption, media_label="voice") async def send_document( @@ -855,20 +734,27 @@ class TeamsAdapter(BasePlatformAdapter): reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs, ) -> SendResult: return await self._send_media_attachment( - chat_id, file_path, "application/octet-stream", caption=caption, media_label="document" - ) + chat_id, file_path, "application/octet-stream", caption=caption, media_label="document") async def get_chat_info(self, chat_id: str) -> dict: return {"name": chat_id, "type": "unknown", "chat_id": chat_id} -# ── Interactive setup ───────────────────────────────────────────────────────── - _SETUP_CREDENTIALS = ( ("Client ID", "TEAMS_CLIENT_ID", {}), ("Client secret", "TEAMS_CLIENT_SECRET", {"password": True}), - ("Tenant ID", "TEAMS_TENANT_ID", {}), -) + ("Tenant ID", "TEAMS_TENANT_ID", {})) +_SETUP_INTRO = ( + "You'll need the Teams CLI. If you haven't already:", + " npm install -g @microsoft/teams.cli@preview", + " teams login", + "", + "Then expose port 3978 publicly (devtunnel / ngrok / cloudflared),", + "and create your bot:", + ' teams app create --name "Hermes" --endpoint "https:///api/messages"', + "", + "The CLI will print CLIENT_ID, CLIENT_SECRET, and TENANT_ID. Paste them below.", + "") def interactive_setup() -> None: @@ -882,17 +768,8 @@ def interactive_setup() -> None: if not prompt_yes_no("Reconfigure Teams?", False): return - print_info("You'll need the Teams CLI. If you haven't already:") - print_info(" npm install -g @microsoft/teams.cli@preview") - print_info(" teams login") - print() - print_info("Then expose port 3978 publicly (devtunnel / ngrok / cloudflared),") - print_info("and create your bot:") - print_info(" teams app create --name \"Hermes\" --endpoint \"https:///api/messages\"") - print() - print_info("The CLI will print CLIENT_ID, CLIENT_SECRET, and TENANT_ID. Paste them below.") - print() - + for line in _SETUP_INTRO: + print_info(line) if line else print() for label, env_key, prompt_kwargs in _SETUP_CREDENTIALS: value = prompt(label, default=get_env_value(env_key) or "", **prompt_kwargs) if not value: @@ -919,13 +796,9 @@ def interactive_setup() -> None: print_info("Restart the gateway: hermes gateway restart") -# ── Plugin entry point ──────────────────────────────────────────────────────── - def _install_hint() -> str: - """Install hint derived from the LAZY_DEPS pins (aiohttp is CVE-pinned, so bumps happen). - - ``venv_pip=True`` targets the real Hermes venv, sidestepping PEP 668 on Ubuntu 24.04. - """ + """Install hint derived from the LAZY_DEPS pins (aiohttp is CVE-pinned, so bumps happen); + ``venv_pip=True`` targets the real Hermes venv, sidestepping PEP 668 on Ubuntu 24.04.""" try: from tools.lazy_deps import feature_install_command cmd = feature_install_command("platform.teams", venv_pip=True) @@ -961,6 +834,4 @@ def register(ctx) -> None: "You are chatting via Microsoft Teams. Teams renders a subset of " "markdown — bold (**text**), italic (*text*), and inline code " "(`code`) work, but complex tables or raw HTML do not. Keep " - "responses clear and professional." - ), - ) + "responses clear and professional."))