From 42e9bc9076affba206ed54c401d04a9c827dace7 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:15:26 -0700 Subject: [PATCH 1/8] refactor(matrix): contextlib.suppress for swallow-only excepts, encrypted-upload predicate, flat send() retry, drop unreachable lazy_deps ImportError fallback in setup (3210->3160) --- plugins/platforms/matrix/adapter.py | 272 ++++++++++++---------------- 1 file changed, 111 insertions(+), 161 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index f4ae705a03..147c206e1f 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -20,6 +20,7 @@ from __future__ import annotations import asyncio import array import inspect +from contextlib import suppress import logging import mimetypes import os @@ -52,8 +53,7 @@ except ImportError: PresenceState = type("_PresenceStateStub", (), { # type: ignore[misc,assignment] "ONLINE": "online", "OFFLINE": "offline", "UNAVAILABLE": "unavailable"}) RoomCreatePreset = type("_RoomCreatePresetStub", (), { # type: ignore[misc,assignment] - "PRIVATE": "private_chat", "PUBLIC": "public_chat", "TRUSTED_PRIVATE": "trusted_private_chat", - }) + "PRIVATE": "private_chat", "PUBLIC": "public_chat", "TRUSTED_PRIVATE": "trusted_private_chat"}) TrustState = type("_TrustStateStub", (), {"UNVERIFIED": 0, "VERIFIED": 1}) # type: ignore[misc,assignment] from gateway.config import Platform, PlatformConfig @@ -91,8 +91,7 @@ def _matrix_voice_metadata_for_file(path: Path) -> Dict[str, Any]: if ffmpeg: try: result = _run_media_tool( - [ffmpeg, "-v", "error", "-i", str(path), "-ac", "1", "-ar", "8000", "-f", "s16le", "-"], - timeout=15) + [ffmpeg, "-v", "error", "-i", str(path), "-ac", "1", "-ar", "8000", "-f", "s16le", "-"], timeout=15) if result.returncode == 0 and result.stdout: samples = array.array("h") samples.frombytes(result.stdout) @@ -130,10 +129,8 @@ def _matrix_transcode_voice_to_ogg(path: str) -> Optional[str]: return ogg_path except Exception: logger.debug("Matrix: voice transcode to Ogg/Opus failed for %s", path, exc_info=True) - try: + with suppress(OSError): os.unlink(ogg_path) - except OSError: - pass return None @@ -225,9 +222,8 @@ class _MatrixHtmlSanitizer(HTMLParser): """Allowlist sanitizer for Matrix-compatible formatted HTML.""" _ALLOWED_TAGS = { - "a", "b", "blockquote", "br", "code", "del", "em", "h1", "h2", "h3", "h4", "h5", "h6", "hr", - "i", "li", "ol", "p", "pre", "s", "strike", "strong", "table", "tbody", "td", "th", "thead", - "tr", "ul"} + "a", "b", "blockquote", "br", "code", "del", "em", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "i", "li", "ol", + "p", "pre", "s", "strike", "strong", "table", "tbody", "td", "th", "thead", "tr", "ul"} _VOID_TAGS = {"br", "hr"} def __init__(self) -> None: @@ -359,13 +355,11 @@ def _resolve_max_message_length(config) -> int: if raw is None: raw = os.getenv("MATRIX_MAX_MESSAGE_LENGTH") if raw is None: - try: + with suppress(Exception): from gateway.platform_registry import platform_registry entry = platform_registry.get("matrix") if entry and entry.max_message_length: raw = entry.max_message_length - except Exception: - pass try: value = int(raw) except (TypeError, ValueError): @@ -454,9 +448,7 @@ def _create_matrix_session(proxy_url: str | None): return aiohttp.ClientSession(connector=ProxyConnector.from_url(proxy_url, rdns=True)) except ImportError: logger.warning( - "aiohttp_socks not installed — SOCKS proxy %s ignored. " - "Run: pip install aiohttp-socks", - proxy_url) + "aiohttp_socks not installed — SOCKS proxy %s ignored. Run: pip install aiohttp-socks", proxy_url) return aiohttp.ClientSession(trust_env=gateway_trust_env()) return aiohttp.ClientSession(proxy=proxy_url) @@ -543,10 +535,8 @@ def _write_matrix_recovery_key_output_file(recovery_key: str) -> Optional[Path]: fh.write(recovery_key) fh.write("\n") except Exception: - try: + with suppress(OSError): os.close(fd) - except OSError: - pass raise return path @@ -584,9 +574,8 @@ def _handle_generated_matrix_recovery_key(mxid: str, recovery_key: str) -> None: return if output_path: logger.warning( - "Matrix: bootstrapped cross-signing for %s. A new recovery key was " - "written to %s with mode 0600. Move it to your secret store and set " - "MATRIX_RECOVERY_KEY for future restarts.", + "Matrix: bootstrapped cross-signing for %s. A new recovery key was written to %s with mode 0600. Move it " + "to your secret store and set MATRIX_RECOVERY_KEY for future restarts.", mxid, output_path) else: logger.warning( @@ -687,9 +676,8 @@ def ensure_matrix_deps() -> bool: from mautrix.types import ( ContentURI, EventID, EventType, PresenceState, RoomCreatePreset, RoomID, TrustState, UserID) return { - "ContentURI": ContentURI, "EventID": EventID, "EventType": EventType, - "PresenceState": PresenceState, "RoomCreatePreset": RoomCreatePreset, "RoomID": RoomID, - "TrustState": TrustState, "UserID": UserID} + "ContentURI": ContentURI, "EventID": EventID, "EventType": EventType, "PresenceState": PresenceState, + "RoomCreatePreset": RoomCreatePreset, "RoomID": RoomID, "TrustState": TrustState, "UserID": UserID} if not ensure_and_bind("platform.matrix", _import, globals(), prompt=False): logger.warning( "Matrix: required packages not installed (%s). Run: pip install " @@ -745,10 +733,8 @@ class _CryptoStateStore: content = raw if isinstance(raw, _Enc) else _Enc.deserialize( raw.serialize() if hasattr(raw, "serialize") else raw) if hasattr(self._ss, "set_encryption_info"): - try: + with suppress(Exception): await self._ss.set_encryption_info(_RID(room_id), content) - except Exception: - pass self._enc_info_cache[room_id] = content return content @@ -952,18 +938,17 @@ class MatrixAdapter(BasePlatformAdapter): await crypto_store.delete() return True - async def _migrate_legacy_crypto_pickle(self, crypto_store: Any, crypto_db: Any, acct_id: str, pickle_key: str) -> bool: + async def _migrate_legacy_crypto_pickle( + self, crypto_store: Any, crypto_db: Any, acct_id: str, pickle_key: str) -> bool: """Re-pickle the Olm account under the current pickle key when it changed. The key embeds the device ID; an account created before MATRIX_DEVICE_ID was set lives under ``:default`` and later fails with BAD_ACCOUNT_KEY (silently disabling optional E2EE). False only when an account exists but no key opens it. """ - try: + with suppress(Exception): await crypto_store.get_account() return True - except Exception: - pass from mautrix.crypto.store.asyncpg import PgCryptoStore for legacy_key in (f"{acct_id}:default", acct_id): if legacy_key == pickle_key: @@ -998,8 +983,7 @@ class MatrixAdapter(BasePlatformAdapter): """Re-pickle olm/megolm sessions too — they share the key; account-only breaks key sharing.""" import olm as olm_lib tables = { - "crypto_olm_session": olm_lib.Session, - "crypto_megolm_inbound_session": olm_lib.InboundGroupSession, + "crypto_olm_session": olm_lib.Session, "crypto_megolm_inbound_session": olm_lib.InboundGroupSession, "crypto_megolm_outbound_session": olm_lib.OutboundGroupSession} for table, session_cls in tables.items(): rows = await crypto_db.fetch(f"SELECT session_id, session FROM {table} WHERE account_id=$1", acct_id) @@ -1007,11 +991,9 @@ class MatrixAdapter(BasePlatformAdapter): if row["session"] is None: continue pickled = bytes(row["session"]) - try: + with suppress(Exception): session_cls.from_pickle(pickled, pickle_key) continue # already readable with the current key - except Exception: - pass try: session = session_cls.from_pickle(pickled, legacy_key) except Exception as exc: @@ -1051,13 +1033,11 @@ class MatrixAdapter(BasePlatformAdapter): "stale. Delete %s and restart.", client.device_id, str(self._crypto_db_path)) return False logger.warning("Matrix: server has stale keys for device %s — attempting re-upload", client.device_id) - try: + with suppress(Exception): await client.api.request( client.api.Method.DELETE if hasattr(client.api, "Method") else "DELETE", f"/_matrix/client/v3/devices/{client.device_id}") logger.info("Matrix: deleted stale device %s from server", client.device_id) - except Exception: - pass try: await olm.share_keys() except Exception as exc: @@ -1116,8 +1096,7 @@ class MatrixAdapter(BasePlatformAdapter): f" (device {effective_device_id})" if effective_device_id else "") except Exception as exc: logger.error( - "Matrix: whoami failed — check MATRIX_ACCESS_TOKEN and MATRIX_HOMESERVER: %s", - exc, exc_info=True) + "Matrix: whoami failed — check MATRIX_ACCESS_TOKEN and MATRIX_HOMESERVER: %s", exc, exc_info=True) await api.session.close() return False elif self._password and self._user_id: @@ -1363,10 +1342,8 @@ class MatrixAdapter(BasePlatformAdapter): except Exception as exc: logger.debug("Matrix: could not close crypto DB on disconnect: %s", exc) if self._client: - try: + with suppress(Exception): await self._client.api.session.close() - except Exception: - pass self._client = None logger.info("Matrix: disconnected") @@ -1384,18 +1361,16 @@ class MatrixAdapter(BasePlatformAdapter): last_event_id = await self._send_room_message(chat_id, msg_content) logger.info("Matrix: sent event %s to %s", last_event_id, chat_id) except Exception as exc: - # On E2EE errors, retry after sharing keys. - if self._encryption and getattr(self._client, "crypto", None): - try: - await self._client.crypto.share_keys() - last_event_id = await self._send_room_message(chat_id, msg_content) - logger.info("Matrix: sent event %s to %s (after key share)", last_event_id, chat_id) - continue - except Exception as retry_exc: - logger.error("Matrix: failed to send to %s after retry: %s", chat_id, retry_exc) - return SendResult(success=False, error=str(retry_exc)) - logger.error("Matrix: failed to send to %s: %s", chat_id, exc) - return SendResult(success=False, error=str(exc)) + if not (self._encryption and getattr(self._client, "crypto", None)): + logger.error("Matrix: failed to send to %s: %s", chat_id, exc) + return SendResult(success=False, error=str(exc)) + try: # E2EE error: retry once after sharing keys + await self._client.crypto.share_keys() + last_event_id = await self._send_room_message(chat_id, msg_content) + logger.info("Matrix: sent event %s to %s (after key share)", last_event_id, chat_id) + except Exception as retry_exc: + logger.error("Matrix: failed to send to %s after retry: %s", chat_id, retry_exc) + return SendResult(success=False, error=str(retry_exc)) return SendResult(success=True, message_id=last_event_id) async def _send_room_message(self, chat_id: str, msg_content: Dict[str, Any]) -> str: @@ -1420,25 +1395,19 @@ class MatrixAdapter(BasePlatformAdapter): "auth": { "access_token_present": token_present, "password_present": bool(self._password), "token_preview": "***" if token_present else "", "user_id": user_id, - "device_id_present": bool(device_id), - "device_id_preview": _redact_matrix_value(device_id)}, + "device_id_present": bool(device_id), "device_id_preview": _redact_matrix_value(device_id)}, "sync": { "connected": self._client is not None, "joined_room_count": len(self._joined_rooms), - "last_sync_age_seconds": max(0.0, now - self._last_sync_ts) if self._last_sync_ts else None - }, + "last_sync_age_seconds": max(0.0, now - self._last_sync_ts) if self._last_sync_ts else None}, "e2ee": { - "mode": self._e2ee_mode, "enabled": bool(self._encryption), - "deps_available": _check_e2ee_deps(), + "mode": self._e2ee_mode, "enabled": bool(self._encryption), "deps_available": _check_e2ee_deps(), "crypto_store_path": str(self._crypto_db_path), "recovery_key_configured": bool(_scoped_recovery_key().strip())}, "policy": { - "allowed_user_count": len(self._allowed_user_ids), - "allowed_room_count": len(self._allowed_room_ids), + "allowed_user_count": len(self._allowed_user_ids), "allowed_room_count": len(self._allowed_room_ids), "ignored_user_pattern_count": len(self._ignored_user_patterns), - "require_mention": self._require_mention, - "free_response_room_count": len(self._free_rooms), - "allow_room_mentions": self._allow_room_mentions, - "process_notices": self._process_notices, + "require_mention": self._require_mention, "free_response_room_count": len(self._free_rooms), + "allow_room_mentions": self._allow_room_mentions, "process_notices": self._process_notices, "allow_public_rooms": _env_truthy("MATRIX_ALLOW_PUBLIC_ROOMS")}, "media": {"max_media_bytes": self._max_media_bytes}} @@ -1446,10 +1415,8 @@ class MatrixAdapter(BasePlatformAdapter): async def _set_typing(self, chat_id: str, timeout: int) -> None: if self._client: - try: + with suppress(Exception): await self._client.set_typing(RoomID(chat_id), timeout=timeout) - except Exception: - pass async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None: """Send a typing indicator.""" @@ -1530,8 +1497,7 @@ class MatrixAdapter(BasePlatformAdapter): fetch_url = url for _ in range(20): async with http.get( - fetch_url, timeout=_aiohttp.ClientTimeout(total=30), allow_redirects=False, - **_req_kw) as resp: + fetch_url, timeout=_aiohttp.ClientTimeout(total=30), allow_redirects=False, **_req_kw) as resp: if resp.status in {301, 302, 303, 307, 308}: location = resp.headers.get("Location") if not location: @@ -1612,10 +1578,8 @@ class MatrixAdapter(BasePlatformAdapter): metadata=metadata, is_voice=True) finally: if converted_path: - try: + with suppress(OSError): os.unlink(converted_path) - except OSError: - pass async def send_video( self, chat_id: str, video_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None, @@ -1674,7 +1638,10 @@ class MatrixAdapter(BasePlatformAdapter): f"{self._format_exec_approval(command, description)}\n\n" f"{scope_choices}Reply `!approve` to execute once, or `!deny` to cancel.\n\n" "You can also click the reaction to approve:\n" + "\n".join(reaction_legend_parts)) - reactions = ("✅", "❌") if not allow_session else ("✅", "🌀", "❌") if not allow_permanent else ("✅", "🌀", "♾️", "❌") + if not allow_session: + reactions = ("✅", "❌") + else: + reactions = ("✅", "🌀", "❌") if not allow_permanent else ("✅", "🌀", "♾️", "❌") def _make(message_id, requester, expires_at): old_event = self._approval_prompt_by_session.get(session_key) @@ -1697,7 +1664,8 @@ class MatrixAdapter(BasePlatformAdapter): (str(model_id), str(p.get("slug") or ""), str(p.get("name") or p.get("slug") or "")) for p in providers or [] for model_id in (p.get("models") or [])][:len(_MATRIX_MODEL_PICKER_REACTIONS)] if not flat_choices: - return await self.send(chat_id, "No authenticated models are available for this session.", metadata=metadata) + return await self.send( + chat_id, "No authenticated models are available for this session.", metadata=metadata) try: from hermes_cli.providers import get_label provider_label = get_label(current_provider) @@ -1757,20 +1725,13 @@ class MatrixAdapter(BasePlatformAdapter): return self._media_too_large(len(data)) upload_data = data encrypted_file = None - if self._encryption and getattr(self._client, "crypto", None): - state_store = getattr(self._client, "state_store", None) - if state_store: - try: - room_encrypted = bool(await state_store.is_encrypted(RoomID(room_id))) - except Exception: - room_encrypted = False - if room_encrypted: - try: - from mautrix.crypto.attachments import encrypt_attachment - upload_data, encrypted_file = encrypt_attachment(data) - except Exception as exc: - logger.error("Matrix: attachment encryption failed: %s", exc) - return SendResult(success=False, error=str(exc)) + if await self._room_needs_encrypted_upload(room_id): + try: + from mautrix.crypto.attachments import encrypt_attachment + upload_data, encrypted_file = encrypt_attachment(data) + except Exception as exc: + logger.error("Matrix: attachment encryption failed: %s", exc) + return SendResult(success=False, error=str(exc)) try: mxc_url = await self._client.upload_media( upload_data, mime_type=content_type, filename=filename, size=len(upload_data)) @@ -1796,8 +1757,21 @@ class MatrixAdapter(BasePlatformAdapter): self._apply_relation_metadata(msg_content, reply_to=reply_to, metadata=metadata) return await self._send_content_event(room_id, msg_content) + async def _room_needs_encrypted_upload(self, room_id: str) -> bool: + """E2EE on, Olm machine loaded, and the state store says the room is encrypted.""" + if not (self._encryption and getattr(self._client, "crypto", None)): + return False + state_store = getattr(self._client, "state_store", None) + if not state_store: + return False + try: + return bool(await state_store.is_encrypted(RoomID(room_id))) + except Exception: + return False + def _media_too_large(self, size: int) -> SendResult: - return SendResult(success=False, error=f"Media file exceeds Matrix limit ({size} > {self._max_media_bytes} bytes)") + return SendResult( + success=False, error=f"Media file exceeds Matrix limit ({size} > {self._max_media_bytes} bytes)") async def _send_content_event(self, room_id: str, msg_content: Dict[str, Any]) -> SendResult: """Send a prebuilt m.room.message payload, mapping exceptions to SendResult.""" @@ -1968,7 +1942,8 @@ class MatrixAdapter(BasePlatformAdapter): room_id = str(getattr(event, "room_id", "")) sender = str(getattr(event, "sender", "")) # DEBUG-level proof the callback fires at all (silent-inbound troubleshooting). - logger.debug("Matrix: callback fired — event %s from %s in %s", getattr(event, "event_id", "?"), sender, room_id) + logger.debug( + "Matrix: callback fired — event %s from %s in %s", getattr(event, "event_id", "?"), sender, room_id) if self._is_self_sender(sender): return # Bridge/system identities must never reach the pairing flow (echo loop once paired). @@ -2035,8 +2010,8 @@ class MatrixAdapter(BasePlatformAdapter): if self._require_mention and not is_free_room and not in_bot_thread: if not is_mentioned and not body.startswith("/"): logger.debug( - "Matrix: ignoring message %s in %s — no @mention (set MATRIX_REQUIRE_MENTION=false to disable)", - event_id, room_id) + "Matrix: ignoring message %s in %s — no @mention " + "(set MATRIX_REQUIRE_MENTION=false to disable)", event_id, room_id) return None # thread_require_mention: even inside a bot thread require @mention — prevents # infinite reply loops when several bots share one thread. @@ -2139,14 +2114,13 @@ class MatrixAdapter(BasePlatformAdapter): if url and not str(url).startswith("mxc://"): logger.warning("[Matrix] Rejecting inbound media %s with non-MXC URL", event_id) return - http_url = self._mxc_to_http(url) if url and url.startswith("mxc://") else "" + http_url = self._mxc_to_http(url) if url else "" content_info = source_content.get("info", {}) if not isinstance(content_info, dict): content_info = {} event_mimetype = content_info.get("mimetype", "") - event_size = content_info.get("size") try: - event_size_int = int(event_size) if event_size is not None else 0 + event_size_int = int(content_info.get("size") or 0) except (TypeError, ValueError): event_size_int = 0 if event_size_int and event_size_int > self._max_media_bytes: @@ -2160,8 +2134,7 @@ class MatrixAdapter(BasePlatformAdapter): if url and not str(url).startswith("mxc://"): logger.warning("[Matrix] Rejecting inbound encrypted media %s with non-MXC URL", event_id) return - if url and url.startswith("mxc://"): - http_url = self._mxc_to_http(url) + http_url = self._mxc_to_http(url) if url else "" is_encrypted_media = bool(file_content and isinstance(file_content, dict) and file_content.get("url")) msg_type, media_type, is_voice_message = self._classify_inbound_media(msgtype, event_mimetype, source_content) # Cache locally so downstream tools get a real file path. @@ -2182,7 +2155,8 @@ class MatrixAdapter(BasePlatformAdapter): await self.handle_message(msg_event) @staticmethod - def _classify_inbound_media(msgtype: str, event_mimetype: str, source_content: dict) -> tuple[MessageType, str, bool]: + def _classify_inbound_media( + msgtype: str, event_mimetype: str, source_content: dict) -> tuple[MessageType, str, bool]: """Map a Matrix media msgtype to (MessageType, mime type, is_voice_message).""" if msgtype == "m.image": return MessageType.PHOTO, event_mimetype or "image/png", False @@ -2257,11 +2231,9 @@ class MatrixAdapter(BasePlatformAdapter): # unless we leave the invite; the match is narrow so transient errors keep retrying. msg = str(exc).lower() if ("no servers" in msg) or ("room not found" in msg): - try: + with suppress(Exception): await self._client.leave_room(RoomID(room_id)) logger.info("Matrix: declined dead invite to %s", room_id) - except Exception: - pass return False def _schedule_invite_join(self, room_id: str, *, is_direct: bool = False, inviter: str = "") -> None: @@ -2367,7 +2339,8 @@ class MatrixAdapter(BasePlatformAdapter): content = getattr(event, "content", None) if not content: return - relates_to = content.get("m.relates_to", {}) if isinstance(content, dict) else getattr(content, "relates_to", {}) + relates_to = (content.get("m.relates_to", {}) if isinstance(content, dict) + else getattr(content, "relates_to", {})) reacts_to = key = "" if isinstance(relates_to, dict): reacts_to = relates_to.get("event_id", "") @@ -2457,7 +2430,8 @@ class MatrixAdapter(BasePlatformAdapter): return True value = choice_prompt.choices.get(key) if value is None: - await self._send_invalid_reaction_feedback(room_id, reacts_to, "That reaction is not one of the available choices.") + await self._send_invalid_reaction_feedback( + room_id, reacts_to, "That reaction is not one of the available choices.") return True choice_prompt.resolved = True self._choice_picker_prompts_by_event.pop(reacts_to, None) @@ -2476,7 +2450,8 @@ class MatrixAdapter(BasePlatformAdapter): def _is_authorized_user(self, user_id: str) -> bool: """GATEWAY_ALLOW_ALL_USERS, or membership in MATRIX_ALLOWED_USERS.""" - return _env_truthy("GATEWAY_ALLOW_ALL_USERS") or bool(self._allowed_user_ids and user_id in self._allowed_user_ids) + return _env_truthy("GATEWAY_ALLOW_ALL_USERS") or bool( + self._allowed_user_ids and user_id in self._allowed_user_ids) async def _validate_matrix_prompt_reactor( self, room_id: str, target_event_id: str, sender: str, prompt: Any, prompt_label: str) -> bool: @@ -2540,8 +2515,8 @@ class MatrixAdapter(BasePlatformAdapter): try: pending = self._pending_text_batches.get(key) last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0 - await asyncio.sleep( - self._text_batch_split_delay_seconds if last_len >= self._split_threshold else self._text_batch_delay_seconds) + near_split = last_len >= self._split_threshold + await asyncio.sleep(self._text_batch_split_delay_seconds if near_split else self._text_batch_delay_seconds) event = self._pending_text_batches.pop(key, None) if not event: return @@ -2670,20 +2645,16 @@ class MatrixAdapter(BasePlatformAdapter): """state_store first (cached), then a direct joined_members API query.""" state_store = getattr(self._client, "state_store", None) if self._client else None if state_store: - try: + with suppress(Exception): members = await state_store.get_members(room_id) if members is not None: return len(members) - except Exception: - pass client = getattr(self, "_client", None) # object.__new__-built test doubles may lack it if client is not None and hasattr(client, "joined_members"): - try: + with suppress(Exception): resp = await client.joined_members(room_id) if getattr(resp, "members", None) is not None: return len(resp.members) - except Exception: - pass return None async def _get_room_state_value(self, room_id: str, event_type: str, key: str) -> Optional[str]: @@ -2902,12 +2873,10 @@ class MatrixAdapter(BasePlatformAdapter): """Get a user's display name in a room, falling back to user_id.""" state_store = getattr(self._client, "state_store", None) if self._client else None if state_store: - try: + with suppress(Exception): member = await state_store.get_member(room_id, user_id) if member and getattr(member, "displayname", None): return member.displayname - except Exception: - pass if user_id.startswith("@") and ":" in user_id: return user_id[1:].split(":")[0] return user_id @@ -2921,7 +2890,7 @@ class MatrixAdapter(BasePlatformAdapter): def _markdown_to_html(self, text: str) -> str: """Markdown → org.matrix.custom.html via ``markdown`` when installed, else the regex fallback.""" text = _pre_sanitize_matrix_markdown(text) - try: + with suppress(ImportError): import markdown as _md md = _md.Markdown(extensions=["fenced_code", "tables", "nl2br", "sane_lists"]) if "html_block" in md.preprocessors: @@ -2931,8 +2900,6 @@ class MatrixAdapter(BasePlatformAdapter): if html.count("

") == 1: html = html.replace("

", "").replace("

", "") return _sanitize_matrix_html(html) - except ImportError: - pass return _sanitize_matrix_html(self._markdown_to_html_fallback(text)) @staticmethod @@ -2969,8 +2936,8 @@ class MatrixAdapter(BasePlatformAdapter): lambda m: _protect_html( f'{_html_escape(m.group(1))}'), result) - result = "".join( - p if p.startswith("\x00PROTECTED") else _html_escape(p) for p in re.split(r"(\x00PROTECTED\d+\x00)", result)) + result = "".join(p if p.startswith("\x00PROTECTED") else _html_escape(p) + for p in re.split(r"(\x00PROTECTED\d+\x00)", result)) # Block-level transforms (line-oriented): hr, headers, blockquote, lists. lines = result.split("\n") out_lines: list = [] @@ -3041,14 +3008,11 @@ async def _standalone_send(pconfig, chat_id, message, *, thread_id=None, media_f url = f"{homeserver}/_matrix/client/v3/rooms/{quote(chat_id, safe='')}/send/m.room.message/{txn_id}" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} payload = {"msgtype": "m.text", "body": message} - try: + with suppress(ImportError): import markdown as _md html = _md.markdown(message, extensions=["fenced_code", "tables"]) - html = re.sub(r"(.*?)", r"\1", html) payload["format"] = "org.matrix.custom.html" - payload["formatted_body"] = html - except ImportError: - pass + payload["formatted_body"] = re.sub(r"(.*?)", r"\1", html) # asyncio.wait_for, not aiohttp.ClientTimeout: cron invokes this via # run_coroutine_threadsafe ("Timeout context manager should be used inside a task"). async with aiohttp.ClientSession() as session: @@ -3057,7 +3021,8 @@ async def _standalone_send(pconfig, chat_id, message, *, thread_id=None, media_f if resp.status not in {200, 201}: return {"error": f"Matrix API error ({resp.status}): {await resp.text()}"} data = await resp.json() - return {"success": True, "platform": "matrix", "chat_id": chat_id, "message_id": data.get("event_id")} + return {"success": True, "platform": "matrix", "chat_id": chat_id, + "message_id": data.get("event_id")} try: return await asyncio.wait_for(_do_send(), timeout=30) except asyncio.TimeoutError: @@ -3104,32 +3069,18 @@ def interactive_setup() -> None: save_env_value("MATRIX_ENCRYPTION", "true") print_success("E2EE enabled") matrix_pkg = "mautrix[encryption]" if want_e2ee else "mautrix" - try: - from tools.lazy_deps import ensure as _lazy_ensure, feature_missing - _missing_before = feature_missing("platform.matrix") - if _missing_before: - print_info(f"Installing {matrix_pkg} (+ {len(_missing_before)} runtime deps)...") - try: - _lazy_ensure("platform.matrix", prompt=False) - print_success(f"{matrix_pkg} installed") - except Exception as exc: - print_warning( - "Install failed — run manually: pip install " - "'mautrix[encryption]' asyncpg aiosqlite Markdown aiohttp-socks") - print_info(f" Error: {exc}") - except ImportError: + from tools.lazy_deps import ensure as _lazy_ensure, feature_missing + _missing_before = feature_missing("platform.matrix") + if _missing_before: + print_info(f"Installing {matrix_pkg} (+ {len(_missing_before)} runtime deps)...") try: - __import__("mautrix") - except ImportError: - print_info(f"Installing {matrix_pkg}...") - from hermes_cli.tools_config import _pip_install - result = _pip_install([matrix_pkg]) - if result.returncode == 0: - print_success(f"{matrix_pkg} installed") - else: - print_warning( - f"Install failed — run manually: uv pip install " - f"'{matrix_pkg}' asyncpg aiosqlite Markdown aiohttp-socks") + _lazy_ensure("platform.matrix", prompt=False) + print_success(f"{matrix_pkg} installed") + except Exception as exc: + print_warning( + "Install failed — run manually: pip install " + "'mautrix[encryption]' asyncpg aiosqlite Markdown aiohttp-socks") + print_info(f" Error: {exc}") print_info("🔒 Security: Restrict who can use your bot") print_info(" Matrix user IDs look like @username:server") allowed_users = prompt("Allowed user IDs (comma-separated, leave empty for open access)") @@ -3199,9 +3150,8 @@ def register(ctx) -> None: ctx.register_platform( name="matrix", label="Matrix", adapter_factory=_build_adapter, check_fn=matrix_deps_present, ensure_deps_fn=ensure_matrix_deps, is_connected=_is_connected, - required_env=["MATRIX_HOMESERVER", "MATRIX_ACCESS_TOKEN"], - install_hint="pip install 'mautrix[encryption]'", setup_fn=interactive_setup, - apply_yaml_config_fn=_apply_yaml_config, allowed_users_env="MATRIX_ALLOWED_USERS", + required_env=["MATRIX_HOMESERVER", "MATRIX_ACCESS_TOKEN"], install_hint="pip install 'mautrix[encryption]'", + setup_fn=interactive_setup, apply_yaml_config_fn=_apply_yaml_config, allowed_users_env="MATRIX_ALLOWED_USERS", allow_all_env="MATRIX_ALLOW_ALL_USERS", cron_deliver_env_var="MATRIX_HOME_ROOM", - standalone_sender_fn=_standalone_send, max_message_length=DEFAULT_MAX_MESSAGE_LENGTH, - emoji="🔐", allow_update_command=True) + standalone_sender_fn=_standalone_send, max_message_length=DEFAULT_MAX_MESSAGE_LENGTH, emoji="🔐", + allow_update_command=True) From 8bf747d04d14264d12ec992422520d5702f81b79 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:25:11 -0700 Subject: [PATCH 2/8] refactor(matrix): one _claim_reaction_prompt gate for approval/model/choice pickers, _abort_connect for the 7 close-and-fail sites, clock-skew reset helper, setup _ask helper (3160->3113) --- plugins/platforms/matrix/adapter.py | 213 +++++++++++----------------- 1 file changed, 83 insertions(+), 130 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 147c206e1f..5e5ed96f34 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -783,10 +783,7 @@ class MatrixAdapter(BasePlatformAdapter): self._invite_join_tasks: Dict[str, asyncio.Task] = {} self._closing = False self._startup_ts: float = 0.0 - # Clock-skew detector state (see _note_late_grace_drop). - self._late_grace_drops: int = 0 - self._late_grace_skew: float = 0.0 - self._clock_skew_warned: bool = False + self._reset_clock_skew_detector() self._last_sync_ts: float = 0.0 self._dm_rooms: Dict[str, bool] = {} self._room_identities: Dict[str, MatrixRoomIdentity] = {} @@ -883,8 +880,6 @@ class MatrixAdapter(BasePlatformAdapter): return configured return os.getenv("MATRIX_THREAD_REQUIRE_MENTION", "false").lower() in {"true", "1", "yes", "on"} - # ---- E2EE helpers ---- - @staticmethod def _extract_server_ed25519(device_keys_obj: Any) -> Optional[str]: """Extract the ed25519 identity key from a DeviceKeys object.""" @@ -1048,7 +1043,13 @@ class MatrixAdapter(BasePlatformAdapter): return await self._reverify_keys_after_upload(client, local_ed25519) return True - # ---- connect / disconnect ---- + @staticmethod + async def _abort_connect(api: Any, crypto_db: Any = None) -> bool: + """Close what connect() opened so far; always False so callers can ``return await``.""" + if crypto_db is not None: + await crypto_db.stop() + await api.session.close() + return False async def _connect_authenticate(self, client: Any, api: Any) -> bool: """Authenticate via access token (whoami) or password login; resolve user/device IDs.""" @@ -1097,8 +1098,7 @@ class MatrixAdapter(BasePlatformAdapter): except Exception as exc: logger.error( "Matrix: whoami failed — check MATRIX_ACCESS_TOKEN and MATRIX_HOMESERVER: %s", exc, exc_info=True) - await api.session.close() - return False + return await self._abort_connect(api) elif self._password and self._user_id: try: resp = await client.login( @@ -1109,12 +1109,10 @@ class MatrixAdapter(BasePlatformAdapter): logger.info("Matrix: logged in as %s", self._user_id) except Exception as exc: logger.error("Matrix: login failed — %s", exc) - await api.session.close() - return False + return await self._abort_connect(api) else: logger.error("Matrix: need MATRIX_ACCESS_TOKEN or MATRIX_USER_ID + MATRIX_PASSWORD") - await api.session.close() - return False + return await self._abort_connect(api) return True async def _connect_setup_e2ee(self, client: Any, api: Any, state_store: Any) -> bool: @@ -1129,8 +1127,7 @@ class MatrixAdapter(BasePlatformAdapter): logger.error( "Matrix: E2EE is required but dependencies are missing. %s. Refusing to connect — " "encrypted rooms would silently fail.", _E2EE_INSTALL_HINT) - await api.session.close() - return False + return await self._abort_connect(api) if self._encryption: try: from mautrix.crypto import OlmMachine @@ -1171,9 +1168,7 @@ class MatrixAdapter(BasePlatformAdapter): olm.send_keys_min_trust = TrustState.UNVERIFIED await olm.load() if not await self._verify_device_keys_on_server(client, olm): - await crypto_db.stop() - await api.session.close() - return False + return await self._abort_connect(api, crypto_db) try: await olm.share_keys() except Exception as exc: @@ -1182,9 +1177,7 @@ class MatrixAdapter(BasePlatformAdapter): "Matrix: device %s has stale one-time keys on the server signed with a " "previous identity key. Delete the device from the homeserver and restart, " "or generate a new access token to get a fresh device ID.", client.device_id) - await crypto_db.stop() - await api.session.close() - return False + return await self._abort_connect(api, crypto_db) logger.warning("Matrix: share_keys() warning during startup: %s", exc) await self._verify_or_bootstrap_cross_signing(olm, client) client.crypto = olm @@ -1205,8 +1198,7 @@ class MatrixAdapter(BasePlatformAdapter): self._encryption = False return True logger.error("Matrix: failed to %s E2EE client: %s. %s", what, exc, _E2EE_INSTALL_HINT) - await api.session.close() - return False + return await self._abort_connect(api) async def _verify_or_bootstrap_cross_signing(self, olm: Any, client: Any) -> None: """Verify cross-signing via MATRIX_RECOVERY_KEY, or bootstrap a new key (non-fatal).""" @@ -1302,10 +1294,7 @@ class MatrixAdapter(BasePlatformAdapter): client.add_event_handler(EventType.REACTION, self._on_reaction, wait_sync=True) client.add_event_handler(IntEvt.INVITE, self._on_invite, wait_sync=True) self._startup_ts = time.time() - # Reset the clock-skew detector per connect so a reconnect after an NTP fix starts clean. - self._late_grace_drops = 0 - self._late_grace_skew = 0.0 - self._clock_skew_warned = False + self._reset_clock_skew_detector() # a reconnect after an NTP fix starts clean self._closing = False await self._connect_initial_sync(client) if self._encryption and getattr(client, "crypto", None): @@ -1336,7 +1325,7 @@ class MatrixAdapter(BasePlatformAdapter): await asyncio.gather(*pending, return_exceptions=True) self._invite_join_tasks.clear() self._reaction_redaction_tasks.clear() - if hasattr(self, "_crypto_db") and self._crypto_db: + if getattr(self, "_crypto_db", None): try: await self._crypto_db.stop() except Exception as exc: @@ -1411,8 +1400,6 @@ class MatrixAdapter(BasePlatformAdapter): "allow_public_rooms": _env_truthy("MATRIX_ALLOW_PUBLIC_ROOMS")}, "media": {"max_media_bytes": self._max_media_bytes}} - # ---- optional overrides ---- - async def _set_typing(self, chat_id: str, timeout: int) -> None: if self._client: with suppress(Exception): @@ -1714,8 +1701,6 @@ class MatrixAdapter(BasePlatformAdapter): """Markdown passes through; strip image markdown (media is uploaded separately).""" return re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", r"\2", content) - # ---- file helpers ---- - async def _upload_and_send( self, room_id: str, data: bytes, filename: str, content_type: str, msgtype: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, @@ -1807,8 +1792,6 @@ class MatrixAdapter(BasePlatformAdapter): return await self._upload_and_send( room_id, data, fname, ct, msgtype, caption, reply_to, metadata, is_voice, voice_metadata) - # ---- sync loop ---- - async def _sync_loop(self) -> None: """Continuously sync with the homeserver.""" client = self._client @@ -1849,8 +1832,6 @@ class MatrixAdapter(BasePlatformAdapter): logger.warning("Matrix: sync error: %s — retrying in 5s", exc) await asyncio.sleep(5) - # ---- event callbacks ---- - async def _dispatch_sync_logged(self, sync_data: Dict[str, Any], what: str) -> None: try: await self._dispatch_sync(sync_data) @@ -1879,9 +1860,7 @@ class MatrixAdapter(BasePlatformAdapter): our own events beats an echo loop ("hall of mirrors"). """ own = (self._user_id or "").strip().lower() - if not own: - return True - return sender.strip().lower() == own + return not own or sender.strip().lower() == own @staticmethod def _is_system_or_bridge_sender(sender: str) -> bool: @@ -1903,14 +1882,18 @@ class MatrixAdapter(BasePlatformAdapter): async def _is_allowed_matrix_room_event(self, room_id: str) -> bool: """MATRIX_ALLOWED_ROOMS gate; DMs are exempt so personal chats survive a project allowlist.""" - if self._is_allowed_matrix_room(room_id): - return True try: - return await self._is_dm_room(room_id) + return self._is_allowed_matrix_room(room_id) or await self._is_dm_room(room_id) except Exception as exc: logger.debug("Matrix: could not resolve room identity for allowlist check in %s: %s", room_id, exc) return False + def _reset_clock_skew_detector(self) -> None: + """State for _note_late_grace_drop: consecutive-drop count, their skew, and the once-only warning.""" + self._late_grace_drops: int = 0 + self._late_grace_skew: float = 0.0 + self._clock_skew_warned: bool = False + def _note_late_grace_drop(self, event_ts: float) -> None: """Clock-skew heuristic for grace-check drops well after startup. @@ -2201,8 +2184,7 @@ class MatrixAdapter(BasePlatformAdapter): async def _on_invite(self, event: Any) -> None: """Auto-join rooms when invited, recording DM rooms in m.direct.""" room_id = str(getattr(event, "room_id", "")) - content = getattr(event, "content", None) - is_direct = bool(getattr(content, "is_direct", False)) + is_direct = bool(getattr(getattr(event, "content", None), "is_direct", False)) inviter = str(getattr(event, "sender", "")) # Only authorized inviters — otherwise any federated user could pull the bot into rooms. if not self._is_authorized_user(inviter): @@ -2238,10 +2220,8 @@ class MatrixAdapter(BasePlatformAdapter): def _schedule_invite_join(self, room_id: str, *, is_direct: bool = False, inviter: str = "") -> None: """Schedule an invite join without blocking sync or gateway readiness.""" - if not room_id or room_id in self._joined_rooms: - return existing = self._invite_join_tasks.get(room_id) - if existing and not existing.done(): + if not room_id or room_id in self._joined_rooms or (existing and not existing.done()): return async def _join_invite() -> None: @@ -2257,8 +2237,7 @@ class MatrixAdapter(BasePlatformAdapter): def _schedule_pending_invite_joins(self, sync_data: Dict[str, Any]) -> None: """Join rooms still present in rooms.invite after sync processing.""" - rooms = sync_data.get("rooms", {}) if isinstance(sync_data, dict) else {} - invites = rooms.get("invite", {}) + invites = (sync_data.get("rooms", {}) if isinstance(sync_data, dict) else {}).get("invite", {}) if not isinstance(invites, dict): return for room_id in invites: @@ -2267,8 +2246,6 @@ class MatrixAdapter(BasePlatformAdapter): logger.info("Matrix: reconciling pending invite for %s", room_id) self._schedule_invite_join(str(room_id)) - # ---- reactions (send, receive, processing lifecycle) ---- - async def _send_reaction(self, room_id: str, event_id: str, emoji: str) -> Optional[str]: """Send an emoji reaction; returns the reaction event_id, or None on failure.""" if not self._client: @@ -2305,22 +2282,16 @@ class MatrixAdapter(BasePlatformAdapter): async def on_processing_start(self, event: MessageEvent) -> None: """Add eyes reaction when the agent starts processing a message.""" - if not self._reactions_enabled: - return - msg_id = event.message_id - room_id = event.source.chat_id - if msg_id and room_id: + msg_id, room_id = event.message_id, event.source.chat_id + if self._reactions_enabled and msg_id and room_id: reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440") if reaction_event_id: self._pending_reactions[(room_id, msg_id)] = reaction_event_id async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: """Replace eyes with checkmark (success) or cross (failure).""" - if not self._reactions_enabled: - return - msg_id = event.message_id - room_id = event.source.chat_id - if not msg_id or not room_id or outcome == ProcessingOutcome.CANCELLED: + msg_id, room_id = event.message_id, event.source.chat_id + if not self._reactions_enabled or not msg_id or not room_id or outcome == ProcessingOutcome.CANCELLED: return eyes_event_id = self._pending_reactions.pop((room_id, msg_id), None) if eyes_event_id: @@ -2354,23 +2325,38 @@ class MatrixAdapter(BasePlatformAdapter): if await handler(room_id, reacts_to, key, sender): return + async def _claim_reaction_prompt( + self, registry: dict, room_id: str, reacts_to: str, key: str, sender: str, label: str, invalid_text: str, + on_expired, choices: Optional[dict] = None) -> tuple[bool, Any, Any]: + """Shared gate for reaction prompts: (handled, prompt, selection). + + handled=False => not our prompt; selection=None with handled=True => consumed without + action (wrong room, expired, unauthorized reactor, or a key that is not a choice). + ``choices`` defaults to ``prompt.choices``. + """ + prompt = registry.get(reacts_to) + if not prompt or prompt.resolved: + return False, None, None + if room_id != prompt.chat_id: + return True, prompt, None + if self._matrix_prompt_expired(prompt): + await on_expired(room_id, reacts_to, prompt) + return True, prompt, None + if not await self._validate_matrix_prompt_reactor(room_id, reacts_to, sender, prompt, label): + return True, prompt, None + selection = (prompt.choices if choices is None else choices).get(key) + if selection is None: + await self._send_invalid_reaction_feedback(room_id, reacts_to, invalid_text) + return True, prompt, selection + async def _handle_approval_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Resolve a pending exec-approval prompt from a reaction. True if it was the target.""" - prompt = self._approval_prompts_by_event.get(reacts_to) - if not prompt or prompt.resolved: - return False - if room_id != prompt.chat_id: - return True - if self._matrix_prompt_expired(prompt): - await self._expire_matrix_approval_prompt(room_id, reacts_to, prompt) - return True - if not await self._validate_matrix_prompt_reactor(room_id, reacts_to, sender, prompt, "approval"): - return True - choice = self._approval_reaction_map.get(key) - if not choice: - await self._send_invalid_reaction_feedback( - room_id, reacts_to, "That reaction is not valid for this approval prompt.") - return True + handled, prompt, choice = await self._claim_reaction_prompt( + self._approval_prompts_by_event, room_id, reacts_to, key, sender, "approval", + "That reaction is not valid for this approval prompt.", self._expire_matrix_approval_prompt, + choices=self._approval_reaction_map) + if choice is None: + return handled try: from tools.approval import resolve_gateway_approval count = resolve_gateway_approval(prompt.session_key, choice) @@ -2388,21 +2374,11 @@ class MatrixAdapter(BasePlatformAdapter): async def _handle_model_picker_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Apply a model-picker reaction. True if the reaction targeted a pending picker.""" - model_prompt = self._model_picker_prompts_by_event.get(reacts_to) - if not model_prompt or model_prompt.resolved: - return False - if room_id != model_prompt.chat_id: - return True - if self._matrix_prompt_expired(model_prompt): - await self._expire_matrix_model_picker_prompt(room_id, reacts_to, model_prompt) - return True - if not await self._validate_matrix_prompt_reactor(room_id, reacts_to, sender, model_prompt, "model picker"): - return True - selection = model_prompt.choices.get(key) - if not selection: - await self._send_invalid_reaction_feedback( - room_id, reacts_to, "That reaction is not one of the available model choices.") - return True + handled, model_prompt, selection = await self._claim_reaction_prompt( + self._model_picker_prompts_by_event, room_id, reacts_to, key, sender, "model picker", + "That reaction is not one of the available model choices.", self._expire_matrix_model_picker_prompt) + if selection is None: + return handled model_prompt.resolved = True self._model_picker_prompts_by_event.pop(reacts_to, None) model_id, provider_slug = selection @@ -2418,21 +2394,13 @@ class MatrixAdapter(BasePlatformAdapter): async def _handle_choice_picker_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Apply a choice-picker reaction. True if the reaction targeted a pending picker.""" - choice_prompt = self._choice_picker_prompts_by_event.get(reacts_to) - if not choice_prompt or choice_prompt.resolved: - return False - if room_id != choice_prompt.chat_id: - return True - if self._matrix_prompt_expired(choice_prompt): - self._choice_picker_prompts_by_event.pop(reacts_to, None) - return True - if not await self._validate_matrix_prompt_reactor(room_id, reacts_to, sender, choice_prompt, "choice picker"): - return True - value = choice_prompt.choices.get(key) + async def _expire(_room_id, target_event_id, _prompt): + self._choice_picker_prompts_by_event.pop(target_event_id, None) + handled, choice_prompt, value = await self._claim_reaction_prompt( + self._choice_picker_prompts_by_event, room_id, reacts_to, key, sender, "choice picker", + "That reaction is not one of the available choices.", _expire) if value is None: - await self._send_invalid_reaction_feedback( - room_id, reacts_to, "That reaction is not one of the available choices.") - return True + return handled choice_prompt.resolved = True self._choice_picker_prompts_by_event.pop(reacts_to, None) try: @@ -2507,8 +2475,6 @@ class MatrixAdapter(BasePlatformAdapter): except Exception as exc: logger.debug("Matrix: failed to redact model picker reaction %s: %s", emoji, exc) - # ---- text aggregation (merges Matrix client-side splits) ---- - async def _flush_text_batch(self, key: str) -> None: """Wait for the quiet period then dispatch the aggregated text.""" current_task = asyncio.current_task() @@ -2526,8 +2492,6 @@ class MatrixAdapter(BasePlatformAdapter): if self._pending_text_batch_tasks.get(key) is current_task: self._pending_text_batch_tasks.pop(key, None) - # ---- read receipts / redaction / rooms / presence ---- - def _background_read_receipt(self, room_id: str, event_id: str) -> None: """Fire-and-forget read receipt with error logging.""" @@ -2628,8 +2592,6 @@ class MatrixAdapter(BasePlatformAdapter): logger.debug("Matrix: set_presence failed: %s", exc) return False - # ---- room identity helpers ---- - @staticmethod def _state_event_value(event: Any, key: str) -> Optional[str]: """Extract a simple value from a Matrix state event object or dict (top-level, then .content).""" @@ -2770,8 +2732,6 @@ class MatrixAdapter(BasePlatformAdapter): self._dm_rooms[room_id] = True self._invalidate_room_identities(room_id) - # ---- outbound formatting / mention helpers ---- - def _build_text_message_content(self, text: str, msgtype: str = "m.text") -> Dict[str, Any]: """Build Matrix text content with HTML and outbound mention metadata.""" msg_content: Dict[str, Any] = {"msgtype": msgtype, "body": text} @@ -2987,9 +2947,6 @@ class MatrixAdapter(BasePlatformAdapter): return result -# ---- plugin glue: register(ctx) + the hook implementations for the bundled-plugin touchpoints ---- - - async def _standalone_send(pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False): """standalone_sender_fn: out-of-process delivery via the Client-Server API (cron without gateway).""" extra = getattr(pconfig, "extra", {}) or {} @@ -3044,24 +3001,20 @@ def interactive_setup() -> None: print_info("Works with any Matrix homeserver (Synapse, Conduit, Dendrite, or matrix.org).") print_info(" 1. Create a bot user on your homeserver, or use your own account") print_info(" 2. Get an access token from Element, or provide user ID + password") - homeserver = prompt("Homeserver URL (e.g. https://matrix.example.org)") - if homeserver: - save_env_value("MATRIX_HOMESERVER", homeserver.rstrip("/")) + def _ask(key: str, question: str, **kw) -> str: + value = prompt(question, **kw) + if value: + save_env_value(key, value.rstrip("/") if key == "MATRIX_HOMESERVER" else value) + return value + _ask("MATRIX_HOMESERVER", "Homeserver URL (e.g. https://matrix.example.org)") print_info("Auth: provide an access token (recommended), or user ID + password.") - token = prompt("Access token (leave empty for password login)", password=True) + token = _ask("MATRIX_ACCESS_TOKEN", "Access token (leave empty for password login)", password=True) if token: - save_env_value("MATRIX_ACCESS_TOKEN", token) - user_id = prompt("User ID (@bot:server — optional, will be auto-detected)") - if user_id: - save_env_value("MATRIX_USER_ID", user_id) + _ask("MATRIX_USER_ID", "User ID (@bot:server — optional, will be auto-detected)") print_success("Matrix access token saved") else: - user_id = prompt("User ID (@bot:server)") - if user_id: - save_env_value("MATRIX_USER_ID", user_id) - password = prompt("Password", password=True) - if password: - save_env_value("MATRIX_PASSWORD", password) + _ask("MATRIX_USER_ID", "User ID (@bot:server)") + if _ask("MATRIX_PASSWORD", "Password", password=True): print_success("Matrix credentials saved") if token or get_env_value("MATRIX_PASSWORD"): want_e2ee = prompt_yes_no("Enable end-to-end encryption (E2EE)?", False) From e29f2e1d567728d1e4d285a5203e78e5be1de20f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:36:27 -0700 Subject: [PATCH 3/8] refactor(matrix): one _MatrixPickerPrompt for model/choice pickers, inline single-use predicates (_redact_matrix_value/_room_server_name/_cache_room_identity/_matches_ignored_user_pattern/_is_allowed_matrix_room), sanitizer _emit, fold docstrings (3113->3044) --- plugins/platforms/matrix/adapter.py | 229 ++++++++++------------------ 1 file changed, 80 insertions(+), 149 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 5e5ed96f34..7b388e1bd3 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -111,9 +111,7 @@ def _matrix_voice_metadata_for_file(path: Path) -> Dict[str, Any]: def _matrix_transcode_voice_to_ogg(path: str) -> Optional[str]: """Transcode to a NEW temp .ogg (caller owns cleanup); None if ffmpeg is missing/fails. - - Blocking subprocess work — call via ``asyncio.to_thread`` from async code. - """ + Blocking subprocess work — call via ``asyncio.to_thread`` from async code.""" ffmpeg = shutil.which("ffmpeg") if not ffmpeg: return None @@ -139,11 +137,9 @@ _MATRIX_BANG_COMMAND_RE = re.compile(r"^!([A-Za-z][A-Za-z0-9_-]*)(?=$|\s)(.*)$", def _resolve_matrix_bang_command(name: str) -> str | None: """Resolve a ``!command`` token (Matrix clients reserve ``/``) to a dispatchable token. - - Only known gateway/skill commands resolve, so ordinary exclamations stay chat text. - Returns whichever candidate resolved — raw lowercased first, then the ``_``→``-`` - variant — never a forced canonical form: aliases pass through for the dispatcher. - """ + Only known gateway/skill commands resolve, so ordinary exclamations stay chat text. Returns + whichever candidate resolved — raw lowercased first, then ``_``→``-`` — never a forced + canonical form: aliases pass through for the dispatcher.""" if not name: return None candidates = list(dict.fromkeys((name.lower(), name.lower().replace("_", "-")))) @@ -245,15 +241,12 @@ class _MatrixHtmlSanitizer(HTMLParser): for key, value in attrs: attr = str(key or "").lower() raw_value = "" if value is None else str(value) - if attr.startswith("on"): - continue if tag == "a" and attr == "href": href = self._safe_url(raw_value) if href: safe.append(f' href="{_html_escape(href, quote=True)}"') - elif tag == "code" and attr == "class": - if re.fullmatch(r"language-[A-Za-z0-9_+.-]{1,64}", raw_value): - safe.append(f' class="{_html_escape(raw_value, quote=True)}"') + elif tag == "code" and attr == "class" and re.fullmatch(r"language-[A-Za-z0-9_+.-]{1,64}", raw_value): + safe.append(f' class="{_html_escape(raw_value, quote=True)}"') return "".join(safe) def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: @@ -272,17 +265,18 @@ class _MatrixHtmlSanitizer(HTMLParser): return self._parts.append(f"") - def handle_data(self, data: str) -> None: + def _emit(self, text: str) -> None: if not self._skip_depth: - self._parts.append(_html_escape(data)) + self._parts.append(text) + + def handle_data(self, data: str) -> None: + self._emit(_html_escape(data)) def handle_entityref(self, name: str) -> None: - if not self._skip_depth: - self._parts.append(f"&{name};") + self._emit(f"&{name};") def handle_charref(self, name: str) -> None: - if not self._skip_depth: - self._parts.append(f"&#{name};") + self._emit(f"&#{name};") def get_html(self) -> str: return "".join(self._parts) @@ -317,31 +311,20 @@ class _MatrixApprovalPrompt: @dataclass -class _MatrixModelPickerPrompt: - """Pending reaction-based model picker prompt.""" +class _MatrixPickerPrompt: + """Pending reaction-based picker; ``choices`` maps emoji -> selection, ``on_selected`` is the callback.""" chat_id: str message_id: str session_key: str - choices: dict[str, tuple[str, str]] - on_model_selected: Any + choices: dict + on_selected: Any requester_user_id: str | None = None expires_at: float | None = None resolved: bool = False bot_reaction_events: dict[str, str] = field(default_factory=dict) -@dataclass -class _MatrixChoicePickerPrompt: - """Pending reaction-based choice picker (/reasoning, /fast).""" - chat_id: str - message_id: str - session_key: str - choices: dict[str, str] # emoji -> value - on_choice_selected: Any - requester_user_id: str | None = None - expires_at: float | None = None - resolved: bool = False - bot_reaction_events: dict[str, str] = field(default_factory=dict) +_MatrixModelPickerPrompt = _MatrixChoicePickerPrompt = _MatrixPickerPrompt # Spec allows ~65 KB events; 4000 was too small (split Markdown tables mid-row). @@ -433,12 +416,9 @@ def _matrix_event_timestamp_seconds(event: Any) -> float: def _create_matrix_session(proxy_url: str | None): - """ClientSession whose proxy applies to *all* requests. - - mautrix's ``HTTPAPI._send()`` never forwards per-request ``proxy=``, so it must be - session-level (``proxy=`` for HTTP(S), ``ProxyConnector`` for SOCKS); with no proxy, - ``trust_env`` honours HTTP(S)_PROXY. - """ + """ClientSession whose proxy applies to *all* requests: mautrix's ``HTTPAPI._send()`` never + forwards per-request ``proxy=``, so it must be session-level (``proxy=`` for HTTP(S), + ``ProxyConnector`` for SOCKS); with no proxy, ``trust_env`` honours HTTP(S)_PROXY.""" import aiohttp if not proxy_url: return aiohttp.ClientSession(trust_env=gateway_trust_env()) @@ -455,9 +435,7 @@ def _create_matrix_session(proxy_url: str | None): def _check_e2ee_deps() -> bool: """True if all four E2EE deps import: olm, PgCryptoStore (also drives sqlite), asyncpg, aiosqlite. - - Without all four, encrypted rooms fail at connect with ``No module named 'asyncpg'``. - """ + Without all four, encrypted rooms fail at connect with ``No module named 'asyncpg'``.""" try: from mautrix.crypto import OlmMachine # noqa: F401 from mautrix.crypto.store.asyncpg import PgCryptoStore # noqa: F401 @@ -516,11 +494,6 @@ def _extra_csv_set(config, key: str, env_name: str) -> Set[str]: return _csv_set(raw) -def _redact_matrix_value(value: Any) -> str: - """Return a safe, non-reversible preview for Matrix diagnostics.""" - return "***" if str(value or "").strip() else "" - - def _write_matrix_recovery_key_output_file(recovery_key: str) -> Optional[Path]: """Write a generated recovery key to MATRIX_RECOVERY_KEY_OUTPUT_FILE (0600, never overwritten).""" output_file = os.getenv("MATRIX_RECOVERY_KEY_OUTPUT_FILE", "").strip() @@ -586,11 +559,9 @@ def _handle_generated_matrix_recovery_key(mxid: str, recovery_key: str) -> None: def _scoped_recovery_key() -> str: - """MATRIX_RECOVERY_KEY via the profile-scoped secret store (see _startup_env_secret). - - A bare os.getenv under multiplex resolves the default profile's key and verification - fails with "Key MAC does not match". - """ + """MATRIX_RECOVERY_KEY via the profile-scoped secret store (see _startup_env_secret): a bare + os.getenv under multiplex resolves the default profile's key and verification fails with + "Key MAC does not match".""" return _startup_env_secret("MATRIX_RECOVERY_KEY") @@ -657,11 +628,9 @@ def check_matrix_requirements() -> bool: def ensure_matrix_deps() -> bool: - """ACTIVE deps-only installer (registry ``ensure_deps_fn``); rebinds the type globals. - - Installs the whole ``platform.matrix`` group when ANY declared package is missing — - short-circuiting on ``import mautrix`` left asyncpg/aiosqlite uninstalled forever. - """ + """ACTIVE deps-only installer (registry ``ensure_deps_fn``); rebinds the type globals. Installs the + whole ``platform.matrix`` group when ANY declared package is missing — short-circuiting on + ``import mautrix`` left asyncpg/aiosqlite uninstalled forever.""" try: from tools.lazy_deps import feature_missing, ensure_and_bind missing = feature_missing("platform.matrix") @@ -829,8 +798,8 @@ class MatrixAdapter(BasePlatformAdapter): self._approval_prompt_by_session: Dict[str, str] = {} self._approval_require_sender: bool = _env_truthy("MATRIX_APPROVAL_REQUIRE_SENDER", "true") self._approval_timeout_seconds = _env_number("MATRIX_APPROVAL_TIMEOUT_SECONDS", 300, int) - self._model_picker_prompts_by_event: Dict[str, _MatrixModelPickerPrompt] = {} - self._choice_picker_prompts_by_event: Dict[str, _MatrixChoicePickerPrompt] = {} + self._model_picker_prompts_by_event: Dict[str, _MatrixPickerPrompt] = {} + self._choice_picker_prompts_by_event: Dict[str, _MatrixPickerPrompt] = {} self._allowed_user_ids: Set[str] = _csv_set(os.getenv("MATRIX_ALLOWED_USERS", "")) self._allowed_room_ids: Set[str] = set(self._allowed_rooms) self._ignored_user_patterns: list[re.Pattern[str]] = [] @@ -913,11 +882,9 @@ class MatrixAdapter(BasePlatformAdapter): return True async def _reset_crypto_store_if_device_changed(self, crypto_store: Any, device_id: str) -> bool: - """Reset the Olm account when the token's device changed; True if reset. - - The store is keyed by user ID, so a new device would inherit the old Olm account - whose identity keys can never be published under the new device ID. - """ + """Reset the Olm account when the token's device changed; True if reset. The store is keyed + by user ID, so a new device would inherit the old Olm account whose identity keys can never + be published under the new device ID.""" if not device_id: return False try: @@ -935,12 +902,10 @@ class MatrixAdapter(BasePlatformAdapter): async def _migrate_legacy_crypto_pickle( self, crypto_store: Any, crypto_db: Any, acct_id: str, pickle_key: str) -> bool: - """Re-pickle the Olm account under the current pickle key when it changed. - - The key embeds the device ID; an account created before MATRIX_DEVICE_ID was set - lives under ``:default`` and later fails with BAD_ACCOUNT_KEY (silently - disabling optional E2EE). False only when an account exists but no key opens it. - """ + """Re-pickle the Olm account under the current pickle key when it changed. The key embeds the + device ID; an account created before MATRIX_DEVICE_ID was set lives under ``:default`` + and later fails with BAD_ACCOUNT_KEY (silently disabling optional E2EE). False only when an + account exists but no key opens it.""" with suppress(Exception): await crypto_store.get_account() return True @@ -1139,11 +1104,9 @@ class MatrixAdapter(BasePlatformAdapter): return False if self._encryption: try: - # Remove legacy pickle file from pre-SQLite era. - legacy_pickle = self._store_dir / "crypto_store.pickle" - if legacy_pickle.exists(): + if (self._store_dir / "crypto_store.pickle").exists(): # pre-SQLite era logger.info("Matrix: removing legacy crypto_store.pickle (migrated to SQLite)") - legacy_pickle.unlink() + (self._store_dir / "crypto_store.pickle").unlink() crypto_db = Database.create( f"sqlite:///{self._crypto_db_path}", upgrade_table=PgCryptoStore.upgrade_table) await crypto_db.start() @@ -1384,7 +1347,7 @@ class MatrixAdapter(BasePlatformAdapter): "auth": { "access_token_present": token_present, "password_present": bool(self._password), "token_preview": "***" if token_present else "", "user_id": user_id, - "device_id_present": bool(device_id), "device_id_preview": _redact_matrix_value(device_id)}, + "device_id_present": bool(device_id), "device_id_preview": "***" if str(device_id or "").strip() else ""}, "sync": { "connected": self._client is not None, "joined_room_count": len(self._joined_rooms), "last_sync_age_seconds": max(0.0, now - self._last_sync_ts) if self._last_sync_ts else None}, @@ -1546,11 +1509,8 @@ class MatrixAdapter(BasePlatformAdapter): 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) -> SendResult: - """Upload audio as an MSC3245 voice message. - - Voice bubbles need Ogg/Opus but callers pass any format (e.g. TTS output), so - transcode here — best-effort: without ffmpeg the original file is sent unchanged. - """ + """Upload audio as an MSC3245 voice message. Voice bubbles need Ogg/Opus but callers pass any + format (e.g. TTS output), so transcode here — best-effort: without ffmpeg the original is sent.""" converted_path: Optional[str] = None send_path = audio_path if not str(audio_path).lower().endswith((".ogg", ".oga", ".opus")): @@ -1667,9 +1627,9 @@ class MatrixAdapter(BasePlatformAdapter): lines.append(f"{emoji} `{model_id}` — {provider_name}") return await self._send_reaction_prompt( chat_id, "\n".join(lines), metadata, - lambda message_id, requester, expires_at: _MatrixModelPickerPrompt( + lambda message_id, requester, expires_at: _MatrixPickerPrompt( chat_id=chat_id, message_id=message_id, session_key=session_key, choices=choices, - on_model_selected=on_model_selected, requester_user_id=requester, expires_at=expires_at), + on_selected=on_model_selected, requester_user_id=requester, expires_at=expires_at), self._model_picker_prompts_by_event, choices, "model picker") async def send_choice_picker( @@ -1692,9 +1652,9 @@ class MatrixAdapter(BasePlatformAdapter): lines += ["", "React to choose."] return await self._send_reaction_prompt( chat_id, "\n".join(lines), metadata, - lambda message_id, requester, expires_at: _MatrixChoicePickerPrompt( + lambda message_id, requester, expires_at: _MatrixPickerPrompt( chat_id=chat_id, message_id=message_id, session_key=session_key, choices=emoji_choices, - on_choice_selected=on_choice_selected, requester_user_id=requester, expires_at=expires_at), + on_selected=on_choice_selected, requester_user_id=requester, expires_at=expires_at), self._choice_picker_prompts_by_event, emoji_choices, "choice picker") def format_message(self, content: str) -> str: @@ -1854,36 +1814,26 @@ class MatrixAdapter(BasePlatformAdapter): logger.warning("Matrix: event handler failed during sync dispatch: %s", result) def _is_self_sender(self, sender: str) -> bool: - """True if *sender* is the bot itself (case-insensitive: homeservers vary localpart case). - - With no resolved user_id we can't prove a sender is NOT us, so return True — dropping - our own events beats an echo loop ("hall of mirrors"). - """ + """True if *sender* is the bot itself (case-insensitive: homeservers vary localpart case). With + no resolved user_id we can't prove a sender is NOT us, so return True — dropping our own + events beats an echo loop ("hall of mirrors").""" own = (self._user_id or "").strip().lower() return not own or sender.strip().lower() == own @staticmethod def _is_system_or_bridge_sender(sender: str) -> bool: """True for appservice/bridge/system identities (``@_telegram_123:server``) or malformed IDs. - - Never offer these a pairing code: an approved bridge would relay every outbound - message back as an "authorized user message" (echo loop). - """ + Never offer these a pairing code: an approved bridge would relay every outbound message + back as an "authorized user message" (echo loop).""" localpart = (sender or "").strip().lstrip("@").partition(":")[0] return not localpart or localpart.startswith("_") - def _matches_ignored_user_pattern(self, sender: str) -> bool: - """Return True when sender matches configured Matrix ignore patterns.""" - return any(pattern.search(sender or "") for pattern in self._ignored_user_patterns) - - def _is_allowed_matrix_room(self, room_id: str) -> bool: - """Return True when MATRIX_ALLOWED_ROOMS permits the room.""" - return not self._allowed_room_ids or room_id in self._allowed_room_ids - async def _is_allowed_matrix_room_event(self, room_id: str) -> bool: """MATRIX_ALLOWED_ROOMS gate; DMs are exempt so personal chats survive a project allowlist.""" + if not self._allowed_room_ids or room_id in self._allowed_room_ids: + return True try: - return self._is_allowed_matrix_room(room_id) or await self._is_dm_room(room_id) + return await self._is_dm_room(room_id) except Exception as exc: logger.debug("Matrix: could not resolve room identity for allowlist check in %s: %s", room_id, exc) return False @@ -1895,13 +1845,10 @@ class MatrixAdapter(BasePlatformAdapter): self._clock_skew_warned: bool = False def _note_late_grace_drop(self, event_ts: float) -> None: - """Clock-skew heuristic for grace-check drops well after startup. - - A host clock set ahead of real time makes every live event look "older than startup" and - the bot silently never replies. Warn once when drops keep happening >30s after startup with - a *consistent* skew — unlike backfill from a freshly invited room, whose event ages vary - widely and reset the counter. - """ + """Clock-skew heuristic for grace-check drops well after startup. A host clock set ahead of + real time makes every live event look "older than startup" and the bot silently never + replies. Warn once when drops keep happening >30s after startup with a *consistent* skew — + unlike backfill from a freshly invited room, whose event ages vary widely and reset the counter.""" if self._clock_skew_warned or time.time() - self._startup_ts <= 30: return skew = self._startup_ts - event_ts @@ -1933,7 +1880,7 @@ class MatrixAdapter(BasePlatformAdapter): if self._is_system_or_bridge_sender(sender): logger.debug("Matrix: ignoring system/bridge sender %s in %s", sender, room_id) return - if self._matches_ignored_user_pattern(sender): + if any(pattern.search(sender or "") for pattern in self._ignored_user_patterns): logger.debug("Matrix: ignoring sender %s in %s due to configured ignore pattern", sender, room_id) return if not await self._is_allowed_matrix_room_event(room_id): @@ -2032,11 +1979,9 @@ class MatrixAdapter(BasePlatformAdapter): async def _extract_reply_context( self, room_id: str, body: str, relates_to: dict ) -> tuple[str, Optional[str], Optional[str], Optional[str], Optional[str]]: - """Return (body, reply_to, reply_to_text, reply_to_author_id, reply_to_author_name). - - Captures the inline reply fallback (``> <@user:srv> text\\n\\nreply``) BEFORE stripping it, - so the prompt layer can render "[Replying to: ...]" like Signal/Slack/Telegram. - """ + """Return (body, reply_to, reply_to_text, reply_to_author_id, reply_to_author_name). Captures + the inline reply fallback (``> <@user:srv> text\\n\\nreply``) BEFORE stripping it, so the + prompt layer can render "[Replying to: ...]" like Signal/Slack/Telegram.""" reply_to = (relates_to.get("m.in_reply_to") or {}).get("event_id") reply_to_text = reply_to_author_id = reply_to_author_name = None if reply_to and body.startswith("> "): @@ -2328,12 +2273,9 @@ class MatrixAdapter(BasePlatformAdapter): async def _claim_reaction_prompt( self, registry: dict, room_id: str, reacts_to: str, key: str, sender: str, label: str, invalid_text: str, on_expired, choices: Optional[dict] = None) -> tuple[bool, Any, Any]: - """Shared gate for reaction prompts: (handled, prompt, selection). - - handled=False => not our prompt; selection=None with handled=True => consumed without - action (wrong room, expired, unauthorized reactor, or a key that is not a choice). - ``choices`` defaults to ``prompt.choices``. - """ + """Shared gate for reaction prompts: (handled, prompt, selection). handled=False => not our + prompt; selection=None with handled=True => consumed without action (wrong room, expired, + unauthorized reactor, or a key that is not a choice). ``choices`` defaults to ``prompt.choices``.""" prompt = registry.get(reacts_to) if not prompt or prompt.resolved: return False, None, None @@ -2383,7 +2325,7 @@ class MatrixAdapter(BasePlatformAdapter): self._model_picker_prompts_by_event.pop(reacts_to, None) model_id, provider_slug = selection try: - confirmation = await model_prompt.on_model_selected(room_id, model_id, provider_slug) + confirmation = await model_prompt.on_selected(room_id, model_id, provider_slug) await self._redact_bot_model_picker_reactions(room_id, model_prompt) if confirmation: await self.send(room_id, confirmation, reply_to=reacts_to) @@ -2404,7 +2346,7 @@ class MatrixAdapter(BasePlatformAdapter): choice_prompt.resolved = True self._choice_picker_prompts_by_event.pop(reacts_to, None) try: - confirmation = await choice_prompt.on_choice_selected(room_id, value) + confirmation = await choice_prompt.on_selected(room_id, value) if confirmation: await self.send(room_id, confirmation, reply_to=reacts_to) except Exception as exc: @@ -2630,10 +2572,6 @@ class MatrixAdapter(BasePlatformAdapter): value = (self._state_event_value(event, key) or "").strip() return value or None - @staticmethod - def _room_server_name(room_id: str) -> Optional[str]: - return (room_id.rsplit(":", 1)[-1].strip() or None) if ":" in room_id else None - def _invalidate_room_identities(self, room_id: str | None = None) -> None: """Drop one cached room identity (or all when *room_id* is None).""" if room_id is None: @@ -2643,14 +2581,6 @@ class MatrixAdapter(BasePlatformAdapter): self._room_identities.pop(room_id, None) self._room_identity_cached_at.pop(room_id, None) - def _cache_room_identity(self, room_id: str, identity: MatrixRoomIdentity) -> None: - if len(self._room_identities) >= self._room_identity_cache_max: - oldest = min(self._room_identity_cached_at, key=self._room_identity_cached_at.get, default=None) - if oldest: - self._invalidate_room_identities(oldest) - self._room_identities[room_id] = identity - self._room_identity_cached_at[room_id] = time.monotonic() - async def _resolve_room_identity(self, room_id: str, *, force_refresh: bool = False) -> MatrixRoomIdentity: """Resolve room identity; member count is the primary DM signal (see below).""" cached = self._room_identities.get(room_id) @@ -2669,11 +2599,17 @@ class MatrixAdapter(BasePlatformAdapter): is_likely_dm = (member_count is not None and member_count <= 2) or (is_direct and not has_explicit_name) identity = MatrixRoomIdentity( room_id=room_id, room_name=room_name, room_topic=room_topic, canonical_alias=canonical_alias, - server_name=self._room_server_name(room_id), joined_member_count=member_count, + server_name=(room_id.rsplit(":", 1)[-1].strip() or None) if ":" in room_id else None, + joined_member_count=member_count, is_direct_account_data=is_direct, display_name=room_name or canonical_alias or room_id, has_explicit_name=has_explicit_name, chat_type="dm" if is_likely_dm else "room", conflict=bool(is_direct and has_explicit_name and (member_count is None or member_count > 2))) - self._cache_room_identity(room_id, identity) + if len(self._room_identities) >= self._room_identity_cache_max: + oldest = min(self._room_identity_cached_at, key=self._room_identity_cached_at.get, default=None) + if oldest: + self._invalidate_room_identities(oldest) + self._room_identities[room_id] = identity + self._room_identity_cached_at[room_id] = time.monotonic() return identity async def _is_dm_room(self, room_id: str) -> bool: @@ -2709,11 +2645,9 @@ class MatrixAdapter(BasePlatformAdapter): self._invalidate_room_identities() async def _record_dm_room(self, room_id: str, inviter: str) -> None: - """Persist a room as DM in m.direct account data after an invite. - - ``m.direct`` is absent (404) until the account has had a DM; fetch the current mapping (if - any), append *room_id* under *inviter*, write it back so ``_refresh_dm_cache`` sees the DM. - """ + """Persist a room as DM in m.direct account data after an invite. ``m.direct`` is absent (404) + until the account has had a DM; fetch the current mapping (if any), append *room_id* under + *inviter*, write it back so ``_refresh_dm_cache`` sees the DM.""" if not self._client: return dm_data: Dict[str, list] = await self._fetch_m_direct(require_dict=True) or {} @@ -3063,11 +2997,8 @@ _YAML_LIST_KEYS = ( def _apply_yaml_config(yaml_cfg: dict, matrix_cfg: dict) -> dict | None: - """apply_yaml_config_fn: config.yaml matrix: keys → MATRIX_* env (env wins). Returns None. - - Lowercased flags apply whenever the key is present (None still writes "none"); - list-valued keys skip None. - """ + """apply_yaml_config_fn: config.yaml matrix: keys → MATRIX_* env (env wins). Returns None. Lowercased + flags apply whenever the key is present (None still writes "none"); list-valued keys skip None.""" for key, env_name in _YAML_LOWER_KEYS: if key in matrix_cfg and not os.getenv(env_name): os.environ[env_name] = str(matrix_cfg[key]).lower() From 0c500d4a61d01967e08720bc96f641a3c2c3d6b0 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:42:24 -0700 Subject: [PATCH 4/8] refactor(matrix): _absorb_sync shared by initial sync + sync loop, _client_op for redact/invite/presence, approval legend+reactions built together (3044->3023) --- plugins/platforms/matrix/adapter.py | 143 ++++++++++++---------------- 1 file changed, 61 insertions(+), 82 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 7b388e1bd3..23b4364ca0 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -1046,7 +1046,7 @@ class MatrixAdapter(BasePlatformAdapter): all_devices = (getattr(dev_resp, "device_keys", {}) or {}).get(str(client.mxid)) or {} if len(all_devices) == 1: client.device_id = next(iter(all_devices)) - elif len(all_devices) == 0: + elif not all_devices: logger.warning( "Matrix: no devices found for %s — key verification will be skipped", client.mxid) except Exception as exc: @@ -1205,18 +1205,8 @@ class MatrixAdapter(BasePlatformAdapter): try: sync_data = await client.sync(timeout=10000, full_state=True) if isinstance(sync_data, dict): - self._last_sync_ts = time.time() self._joined_rooms.clear() - self._joined_rooms.update(sync_data.get("rooms", {}).get("join", {}).keys()) - self._invalidate_room_identities() - nb = sync_data.get("next_batch") # incremental syncs resume from here - if nb: - await client.sync_store.put_next_batch(nb) - logger.info("Matrix: initial sync complete, joined %d rooms", len(self._joined_rooms)) - await self._refresh_dm_cache() - # Dispatch so the OlmMachine sees to-device key shares queued while offline. - await self._dispatch_sync_logged(sync_data, "initial sync event dispatch error") - self._schedule_pending_invite_joins(sync_data) + await self._absorb_sync(client, sync_data, initial=True) else: logger.warning("Matrix: initial sync returned unexpected type %s", type(sync_data).__name__) except Exception as exc: @@ -1489,9 +1479,7 @@ class MatrixAdapter(BasePlatformAdapter): for idx, (image_url, alt_text) in enumerate(images, start=1): if human_delay > 0 and idx > 1: await asyncio.sleep(human_delay) - caption = alt_text or None - if total > 1 and caption: - caption = f"{caption} ({idx}/{total})" + caption = f"{alt_text} ({idx}/{total})" if alt_text and total > 1 else (alt_text or None) if image_url.startswith("file://"): result = await self.send_image_file( chat_id=chat_id, image_path=_unquote(image_url[7:]), caption=caption, metadata=metadata) @@ -1512,14 +1500,11 @@ class MatrixAdapter(BasePlatformAdapter): """Upload audio as an MSC3245 voice message. Voice bubbles need Ogg/Opus but callers pass any format (e.g. TTS output), so transcode here — best-effort: without ffmpeg the original is sent.""" converted_path: Optional[str] = None - send_path = audio_path if not str(audio_path).lower().endswith((".ogg", ".oga", ".opus")): converted_path = await asyncio.to_thread(_matrix_transcode_voice_to_ogg, audio_path) - if converted_path: - send_path = converted_path try: return await self._send_local_file( - chat_id, send_path, "m.audio", caption, reply_to, + chat_id, converted_path or audio_path, "m.audio", caption, reply_to, # keep the caller's basename (the temp transcode file has a generated name) file_name=(Path(audio_path).with_suffix(".ogg").name if converted_path else None), metadata=metadata, is_voice=True) @@ -1567,28 +1552,26 @@ class MatrixAdapter(BasePlatformAdapter): """Send a reaction-based exec approval prompt for Matrix.""" if not self._client: return SendResult(success=False, error="Not connected") - scope_choices = "" if smart_denied: scope_choices = "Smart DENY: owner override applies to this one operation only.\n" else: - if allow_session: - scope_choices += "Reply `!approve session` to approve this pattern for the session, " - if allow_permanent: - scope_choices += "`!approve always` to approve permanently, " - reaction_legend_parts = ["✅ = approve once"] + scope_choices = ( + ("Reply `!approve session` to approve this pattern for the session, " if allow_session else "") + + ("`!approve always` to approve permanently, " if allow_permanent else "")) + legend = ["✅ = approve once"] + reactions = ["✅"] if allow_session: - reaction_legend_parts.append("🌀 = approve for this session") + legend.append("🌀 = approve for this session") + reactions.append("🌀") if allow_permanent: - reaction_legend_parts.append("♾️ = approve always") - reaction_legend_parts.append("❎ = deny") + legend.append("♾️ = approve always") + reactions.append("♾️") + legend.append("❎ = deny") + reactions.append("❌") text = ( f"{self._format_exec_approval(command, description)}\n\n" f"{scope_choices}Reply `!approve` to execute once, or `!deny` to cancel.\n\n" - "You can also click the reaction to approve:\n" + "\n".join(reaction_legend_parts)) - if not allow_session: - reactions = ("✅", "❌") - else: - reactions = ("✅", "🌀", "❌") if not allow_permanent else ("✅", "🌀", "♾️", "❌") + "You can also click the reaction to approve:\n" + "\n".join(legend)) def _make(message_id, requester, expires_at): old_event = self._approval_prompt_by_session.get(session_key) @@ -1599,7 +1582,7 @@ class MatrixAdapter(BasePlatformAdapter): session_key=session_key, chat_id=chat_id, message_id=message_id, requester_user_id=requester, expires_at=expires_at) return await self._send_reaction_prompt( - chat_id, text, metadata, _make, self._approval_prompts_by_event, reactions, "approval") + chat_id, text, metadata, _make, self._approval_prompts_by_event, tuple(reactions), "approval") async def send_model_picker( self, chat_id: str, providers: list, current_model: str, current_provider: str, session_key: str, @@ -1686,9 +1669,7 @@ class MatrixAdapter(BasePlatformAdapter): msg_content: Dict[str, Any] = { "msgtype": msgtype, "body": caption or filename, "info": {"mimetype": content_type, "size": len(data)}} if encrypted_file is not None: - file_payload = encrypted_file.serialize() - file_payload["url"] = str(mxc_url) - msg_content["file"] = file_payload + msg_content["file"] = {**encrypted_file.serialize(), "url": str(mxc_url)} else: msg_content["url"] = str(mxc_url) if is_voice: # MSC3245 native voice flag + MSC1767 audio metadata @@ -1744,13 +1725,12 @@ class MatrixAdapter(BasePlatformAdapter): if file_size > self._max_media_bytes: return self._media_too_large(file_size) fname = file_name or p.name - ct = mimetypes.guess_type(fname)[0] or "application/octet-stream" - data = p.read_bytes() # ffprobe/ffmpeg probing is blocking (subprocess timeouts up to 15s) — # run it off the event loop so voice uploads never stall the adapter. voice_metadata = await asyncio.to_thread(_matrix_voice_metadata_for_file, p) if is_voice else None return await self._upload_and_send( - room_id, data, fname, ct, msgtype, caption, reply_to, metadata, is_voice, voice_metadata) + room_id, p.read_bytes(), fname, mimetypes.guess_type(fname)[0] or "application/octet-stream", msgtype, + caption, reply_to, metadata, is_voice, voice_metadata) async def _sync_loop(self) -> None: """Continuously sync with the homeserver.""" @@ -1762,23 +1742,11 @@ class MatrixAdapter(BasePlatformAdapter): sync_data = await asyncio.wait_for(client.sync(since=next_batch, timeout=30000), timeout=45.0) # Auth failures (M_UNKNOWN_TOKEN) arrive as SyncError objects, not exceptions. _sync_msg = getattr(sync_data, "message", None) - if _sync_msg and isinstance(_sync_msg, str): - _lower = _sync_msg.lower() - if "m_unknown_token" in _lower or "unknown_token" in _lower: - logger.error("Matrix: permanent auth error from sync: %s — stopping", _sync_msg) - return + if isinstance(_sync_msg, str) and "unknown_token" in _sync_msg.lower(): + logger.error("Matrix: permanent auth error from sync: %s — stopping", _sync_msg) + return if isinstance(sync_data, dict): - self._last_sync_ts = time.time() - rooms_join = sync_data.get("rooms", {}).get("join", {}) - if rooms_join: - self._joined_rooms.update(rooms_join.keys()) - self._invalidate_room_identities() - nb = sync_data.get("next_batch") - if nb: - next_batch = nb - await client.sync_store.put_next_batch(nb) - await self._dispatch_sync_logged(sync_data, "sync event dispatch error") - self._schedule_pending_invite_joins(sync_data) + next_batch = await self._absorb_sync(client, sync_data) or next_batch await asyncio.sleep(0) # let fresh invite joins start before the next sync except asyncio.CancelledError: return @@ -1792,11 +1760,27 @@ class MatrixAdapter(BasePlatformAdapter): logger.warning("Matrix: sync error: %s — retrying in 5s", exc) await asyncio.sleep(5) - async def _dispatch_sync_logged(self, sync_data: Dict[str, Any], what: str) -> None: + async def _absorb_sync(self, client: Any, sync_data: Dict[str, Any], *, initial: bool = False) -> Optional[str]: + """Apply one sync response: joined rooms, next_batch, event dispatch, pending invites. Returns next_batch. + The initial (full-state) sync also seeds the DM cache and dispatches so the OlmMachine sees + to-device key shares queued while offline.""" + self._last_sync_ts = time.time() + rooms_join = sync_data.get("rooms", {}).get("join", {}) + if rooms_join or initial: + self._joined_rooms.update(rooms_join.keys()) + self._invalidate_room_identities() + nb = sync_data.get("next_batch") # incremental syncs resume from here + if nb: + await client.sync_store.put_next_batch(nb) + if initial: + logger.info("Matrix: initial sync complete, joined %d rooms", len(self._joined_rooms)) + await self._refresh_dm_cache() try: await self._dispatch_sync(sync_data) except Exception as exc: - logger.warning("Matrix: %s: %s", what, exc) + logger.warning("Matrix: %s: %s", "initial sync event dispatch error" if initial else "sync event dispatch error", exc) + self._schedule_pending_invite_joins(sync_data) + return nb async def _dispatch_sync(self, sync_data: Dict[str, Any]) -> None: """Dispatch a sync response through the mautrix event machinery.""" @@ -2466,18 +2450,24 @@ class MatrixAdapter(BasePlatformAdapter): logger.debug("Matrix: read receipt failed: %s", exc) return False - async def redact_message(self, room_id: str, event_id: str, reason: str = "") -> bool: - """Redact (delete) a message or event from a room.""" + async def _client_op(self, coro_factory, ok_msg: tuple, err_msg: str, *, level: str = "warning") -> bool: + """Run one client call when connected: log *ok_msg* and return True, or log the error and return False.""" if not self._client: return False try: - await self._client.redact(RoomID(room_id), EventID(event_id), reason=reason or None) - logger.info("Matrix: redacted %s in %s", event_id, room_id) + await coro_factory() + getattr(logger, "debug" if level == "debug" else "info")(*ok_msg) return True except Exception as exc: - logger.warning("Matrix: redact error: %s", exc) + getattr(logger, level)(err_msg, exc) return False + async def redact_message(self, room_id: str, event_id: str, reason: str = "") -> bool: + """Redact (delete) a message or event from a room.""" + return await self._client_op( + lambda: self._client.redact(RoomID(room_id), EventID(event_id), reason=reason or None), + ("Matrix: redacted %s in %s", event_id, room_id), "Matrix: redact error: %s") + async def create_room( self, name: str = "", topic: str = "", invite: Optional[list] = None, is_direct: bool = False, preset: str = "private_chat") -> Optional[str]: @@ -2504,15 +2494,9 @@ class MatrixAdapter(BasePlatformAdapter): async def invite_user(self, room_id: str, user_id: str) -> bool: """Invite a user to a room.""" - if not self._client: - return False - try: - await self._client.invite_user(RoomID(room_id), UserID(user_id)) - logger.info("Matrix: invited %s to %s", user_id, room_id) - return True - except Exception as exc: - logger.warning("Matrix: invite error: %s", exc) - return False + return await self._client_op( + lambda: self._client.invite_user(RoomID(room_id), UserID(user_id)), + ("Matrix: invited %s to %s", user_id, room_id), "Matrix: invite error: %s") _VALID_PRESENCE_STATES = frozenset(("online", "offline", "unavailable")) @@ -2523,16 +2507,11 @@ class MatrixAdapter(BasePlatformAdapter): if state not in self._VALID_PRESENCE_STATES: logger.warning("Matrix: invalid presence state %r", state) return False - try: - presence_map = { - "online": PresenceState.ONLINE, "offline": PresenceState.OFFLINE, - "unavailable": PresenceState.UNAVAILABLE} - await self._client.set_presence(presence=presence_map[state], status=status_msg or None) - logger.debug("Matrix: presence set to %s", state) - return True - except Exception as exc: - logger.debug("Matrix: set_presence failed: %s", exc) - return False + presence_map = { + "online": PresenceState.ONLINE, "offline": PresenceState.OFFLINE, "unavailable": PresenceState.UNAVAILABLE} + return await self._client_op( + lambda: self._client.set_presence(presence=presence_map[state], status=status_msg or None), + ("Matrix: presence set to %s", state), "Matrix: set_presence failed: %s", level="debug") @staticmethod def _state_event_value(event: Any, key: str) -> Optional[str]: From ebec8fa1bb3adb1bc3005280ae120e7822eedc31 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:52:54 -0700 Subject: [PATCH 5/8] refactor(matrix): single re-upload path in device-key verification, synthetic thread-root policy as one predicate, merged E2EE setup try (phase label), recovery-key path helper (3023->3008) --- plugins/platforms/matrix/adapter.py | 221 +++++++++++++--------------- 1 file changed, 103 insertions(+), 118 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 23b4364ca0..08ce4aadae 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -494,15 +494,18 @@ def _extra_csv_set(config, key: str, env_name: str) -> Set[str]: return _csv_set(raw) +def _recovery_key_output_path() -> Optional[Path]: + output_file = os.getenv("MATRIX_RECOVERY_KEY_OUTPUT_FILE", "").strip() + return Path(output_file).expanduser() if output_file else None + + def _write_matrix_recovery_key_output_file(recovery_key: str) -> Optional[Path]: """Write a generated recovery key to MATRIX_RECOVERY_KEY_OUTPUT_FILE (0600, never overwritten).""" - output_file = os.getenv("MATRIX_RECOVERY_KEY_OUTPUT_FILE", "").strip() - if not output_file: + path = _recovery_key_output_path() + if path is None: return None - path = Path(output_file).expanduser() path.parent.mkdir(parents=True, exist_ok=True) - flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL - fd = os.open(path, flags, 0o600) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(recovery_key) @@ -516,10 +519,9 @@ def _write_matrix_recovery_key_output_file(recovery_key: str) -> Optional[Path]: def _get_matrix_recovery_key_output_target() -> tuple[Optional[Path], str]: """Return a usable one-time recovery-key output path, or a redacted reason.""" - output_file = os.getenv("MATRIX_RECOVERY_KEY_OUTPUT_FILE", "").strip() - if not output_file: + path = _recovery_key_output_path() + if path is None: return None, "not_configured" - path = Path(output_file).expanduser() if path.exists(): return None, "exists" try: @@ -913,11 +915,10 @@ class MatrixAdapter(BasePlatformAdapter): for legacy_key in (f"{acct_id}:default", acct_id): if legacy_key == pickle_key: continue - legacy_store = PgCryptoStore(account_id=acct_id, pickle_key=legacy_key, db=crypto_db) try: - account = await legacy_store.get_account() + account = await PgCryptoStore(account_id=acct_id, pickle_key=legacy_key, db=crypto_db).get_account() except Exception: - continue + account = None if account is None: continue # Sessions first, account last: the account is the commit marker (the fast path @@ -977,36 +978,34 @@ class MatrixAdapter(BasePlatformAdapter): logger.error("Matrix: cannot verify device keys on server: %s — refusing E2EE", exc, exc_info=True) return False local_ed25519 = olm.account.identity_keys.get("ed25519") + + async def _reupload(error_fmt: str, *error_args) -> bool: + try: + await olm.share_keys() + except Exception as exc: + logger.error(error_fmt, *error_args, exc, exc_info=True) + return False + return await self._reverify_keys_after_upload(client, local_ed25519) if not our_keys: logger.warning("Matrix: device keys missing from server — re-uploading") olm.account.shared = False - try: - await olm.share_keys() - except Exception as exc: - logger.error("Matrix: failed to re-upload device keys: %s", exc, exc_info=True) - return False - return await self._reverify_keys_after_upload(client, local_ed25519) - if self._extract_server_ed25519(our_keys) != local_ed25519: - if olm.account.shared: - logger.error( - "Matrix: server has different identity keys for device %s — local crypto state is " - "stale. Delete %s and restart.", client.device_id, str(self._crypto_db_path)) - return False - logger.warning("Matrix: server has stale keys for device %s — attempting re-upload", client.device_id) - with suppress(Exception): - await client.api.request( - client.api.Method.DELETE if hasattr(client.api, "Method") else "DELETE", - f"/_matrix/client/v3/devices/{client.device_id}") - logger.info("Matrix: deleted stale device %s from server", client.device_id) - try: - await olm.share_keys() - except Exception as exc: - logger.error( - "Matrix: cannot upload device keys for %s: %s. Try generating a new access token " - "to get a fresh device.", client.device_id, exc, exc_info=True) - return False - return await self._reverify_keys_after_upload(client, local_ed25519) - return True + return await _reupload("Matrix: failed to re-upload device keys: %s") + if self._extract_server_ed25519(our_keys) == local_ed25519: + return True + if olm.account.shared: + logger.error( + "Matrix: server has different identity keys for device %s — local crypto state is " + "stale. Delete %s and restart.", client.device_id, str(self._crypto_db_path)) + return False + logger.warning("Matrix: server has stale keys for device %s — attempting re-upload", client.device_id) + with suppress(Exception): + await client.api.request( + client.api.Method.DELETE if hasattr(client.api, "Method") else "DELETE", + f"/_matrix/client/v3/devices/{client.device_id}") + logger.info("Matrix: deleted stale device %s from server", client.device_id) + return await _reupload( + "Matrix: cannot upload device keys for %s: %s. Try generating a new access token to get a fresh device.", + client.device_id) @staticmethod async def _abort_connect(api: Any, crypto_db: Any = None) -> bool: @@ -1093,63 +1092,60 @@ class MatrixAdapter(BasePlatformAdapter): "Matrix: E2EE is required but dependencies are missing. %s. Refusing to connect — " "encrypted rooms would silently fail.", _E2EE_INSTALL_HINT) return await self._abort_connect(api) - if self._encryption: + if not self._encryption: + return True + phase = "import" + try: + from mautrix.crypto import OlmMachine + from mautrix.crypto.store.asyncpg import PgCryptoStore + from mautrix.util.async_db import Database + self._store_dir.mkdir(parents=True, exist_ok=True) + phase = "create" + if (self._store_dir / "crypto_store.pickle").exists(): # pre-SQLite era + logger.info("Matrix: removing legacy crypto_store.pickle (migrated to SQLite)") + (self._store_dir / "crypto_store.pickle").unlink() + crypto_db = Database.create( + f"sqlite:///{self._crypto_db_path}", upgrade_table=PgCryptoStore.upgrade_table) + await crypto_db.start() + self._crypto_db = crypto_db + _acct_id = self._user_id or "hermes" + # Key on the RESOLVED client.device_id (token's real device), not the configured + # one, or the Olm account is stored under a key that can never be looked up. + _pickle_key = f"{_acct_id}:{client.device_id or self._device_id or 'default'}" + crypto_store = PgCryptoStore(account_id=_acct_id, pickle_key=_pickle_key, db=crypto_db) + await crypto_store.open() + _store_was_reset = False + if client.device_id: + _store_was_reset = await self._reset_crypto_store_if_device_changed(crypto_store, client.device_id) + await crypto_store.put_device_id(client.device_id) + # A just-deleted store has no account to migrate. + if not _store_was_reset and not await self._migrate_legacy_crypto_pickle( + crypto_store, crypto_db, _acct_id, _pickle_key): + logger.warning("Matrix: crypto pickle migration failed — E2EE may not work correctly") + crypto_state = _CryptoStateStore(state_store, self._joined_rooms, client) + olm = OlmMachine(client, crypto_store, crypto_state) + olm.share_keys_min_trust = TrustState.UNVERIFIED + olm.send_keys_min_trust = TrustState.UNVERIFIED + await olm.load() + if not await self._verify_device_keys_on_server(client, olm): + return await self._abort_connect(api, crypto_db) try: - from mautrix.crypto import OlmMachine - from mautrix.crypto.store.asyncpg import PgCryptoStore - from mautrix.util.async_db import Database - self._store_dir.mkdir(parents=True, exist_ok=True) + await olm.share_keys() except Exception as exc: - if not await self._e2ee_setup_failed("import", exc, api): - return False - if self._encryption: - try: - if (self._store_dir / "crypto_store.pickle").exists(): # pre-SQLite era - logger.info("Matrix: removing legacy crypto_store.pickle (migrated to SQLite)") - (self._store_dir / "crypto_store.pickle").unlink() - crypto_db = Database.create( - f"sqlite:///{self._crypto_db_path}", upgrade_table=PgCryptoStore.upgrade_table) - await crypto_db.start() - self._crypto_db = crypto_db - _acct_id = self._user_id or "hermes" - # Key on the RESOLVED client.device_id (token's real device), not the configured - # one, or the Olm account is stored under a key that can never be looked up. - _pickle_key = f"{_acct_id}:{client.device_id or self._device_id or 'default'}" - crypto_store = PgCryptoStore(account_id=_acct_id, pickle_key=_pickle_key, db=crypto_db) - await crypto_store.open() - _store_was_reset = False - if client.device_id: - _store_was_reset = await self._reset_crypto_store_if_device_changed(crypto_store, client.device_id) - await crypto_store.put_device_id(client.device_id) - # A just-deleted store has no account to migrate. - if not _store_was_reset and not await self._migrate_legacy_crypto_pickle( - crypto_store, crypto_db, _acct_id, _pickle_key): - logger.warning("Matrix: crypto pickle migration failed — E2EE may not work correctly") - crypto_state = _CryptoStateStore(state_store, self._joined_rooms, client) - olm = OlmMachine(client, crypto_store, crypto_state) - olm.share_keys_min_trust = TrustState.UNVERIFIED - olm.send_keys_min_trust = TrustState.UNVERIFIED - await olm.load() - if not await self._verify_device_keys_on_server(client, olm): + if "already exists" in str(exc): + logger.error( + "Matrix: device %s has stale one-time keys on the server signed with a " + "previous identity key. Delete the device from the homeserver and restart, " + "or generate a new access token to get a fresh device ID.", client.device_id) return await self._abort_connect(api, crypto_db) - try: - await olm.share_keys() - except Exception as exc: - if "already exists" in str(exc): - logger.error( - "Matrix: device %s has stale one-time keys on the server signed with a " - "previous identity key. Delete the device from the homeserver and restart, " - "or generate a new access token to get a fresh device ID.", client.device_id) - return await self._abort_connect(api, crypto_db) - logger.warning("Matrix: share_keys() warning during startup: %s", exc) - await self._verify_or_bootstrap_cross_signing(olm, client) - client.crypto = olm - logger.info( - "Matrix: E2EE enabled (store: %s%s)", str(self._crypto_db_path), - f", device_id={client.device_id}" if client.device_id else "") - except Exception as exc: - if not await self._e2ee_setup_failed("create", exc, api): - return False + logger.warning("Matrix: share_keys() warning during startup: %s", exc) + await self._verify_or_bootstrap_cross_signing(olm, client) + client.crypto = olm + logger.info( + "Matrix: E2EE enabled (store: %s%s)", str(self._crypto_db_path), + f", device_id={client.device_id}" if client.device_id else "") + except Exception as exc: + return await self._e2ee_setup_failed(phase, exc, api) return True async def _e2ee_setup_failed(self, what: str, exc: Exception, api: Any) -> bool: @@ -1422,14 +1418,6 @@ class MatrixAdapter(BasePlatformAdapter): raise ValueError("external media is not an image") return b"".join(parts), content_type fname = url.rsplit("/", 1)[-1].split("?")[0] or "image.png" - - def _safe_redirect_target(current_url: str, location: str) -> str: - """Re-validate EVERY redirect hop: a public URL can 302 toward loopback/metadata - endpoints, and checking only the final URL is too late (the hop already connected).""" - next_url = urljoin(current_url, location) - if not is_safe_url(next_url): - raise ValueError("blocked unsafe redirect URL") - return next_url try: import aiohttp as _aiohttp _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(self._proxy_url) @@ -1442,7 +1430,11 @@ class MatrixAdapter(BasePlatformAdapter): location = resp.headers.get("Location") if not location: raise ValueError("redirect missing Location") - fetch_url = _safe_redirect_target(fetch_url, location) + # Re-validate EVERY hop: a public URL can 302 toward loopback/metadata endpoints, + # and checking only the final URL is too late (the hop already connected). + fetch_url = urljoin(fetch_url, location) + if not is_safe_url(fetch_url): + raise ValueError("blocked unsafe redirect URL") continue resp.raise_for_status() data, ct = await _read_capped( @@ -1881,14 +1873,11 @@ class MatrixAdapter(BasePlatformAdapter): content = getattr(event, "content", None) if content is None: return - if hasattr(content, "msgtype"): - msgtype = str(content.msgtype) - else: - msgtype = content.get("msgtype", "") if isinstance(content, dict) else "" if isinstance(content, dict): - source_content = content + source_content, msgtype = content, content.get("msgtype", "") else: source_content = content.serialize() if hasattr(content, "serialize") else {} + msgtype = str(content.msgtype) if hasattr(content, "msgtype") else "" relates_to = source_content.get("m.relates_to", {}) if relates_to.get("rel_type") == "m.replace": # skip edits return @@ -1934,29 +1923,25 @@ class MatrixAdapter(BasePlatformAdapter): "Matrix: ignoring message %s in thread %s — no @mention (thread_require_mention=true)", event_id, thread_id) return None - if is_dm and not thread_id and self._dm_mention_threads and is_mentioned: - thread_id = event_id - self._threads.mark(thread_id) if is_mentioned and self._require_mention: body = self._strip_mention(body) - # Real thread roots are preserved above; synthetic roots follow session-scope policy. + # Real thread roots are preserved above; synthetic roots (this event) follow policy: DM + # @mention threads / DM auto-thread, or room auto-thread unless session_scope pins the room. if not thread_id: if is_dm: - if self._dm_auto_thread: - thread_id = event_id - self._threads.mark(thread_id) - elif self._matrix_session_scope == "room": - thread_id = None - elif self._matrix_session_scope == "thread" or self._auto_thread: + synthetic = (self._dm_mention_threads and is_mentioned) or self._dm_auto_thread + else: + synthetic = self._matrix_session_scope == "thread" or ( + self._matrix_session_scope != "room" and self._auto_thread) + if synthetic: thread_id = event_id - self._threads.mark(thread_id) display_name = await self._get_display_name(room_id, sender) source = self.build_source( chat_id=room_id, chat_name=identity.display_name, chat_type=chat_type, user_id=sender, user_name=display_name, thread_id=thread_id, chat_topic=identity.room_topic, guild_id=identity.server_name, parent_chat_id=room_id if thread_id else None, message_id=event_id) if thread_id: - self._threads.mark(thread_id) + self._threads.mark(thread_id) # covers real roots and synthetic ones alike self._background_read_receipt(room_id, event_id) return body, is_dm, chat_type, thread_id, display_name, source From e657c73f32ba1b9cf60bf7e8d9546b98ee1079cb Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:56:34 -0700 Subject: [PATCH 6/8] refactor(matrix): media http fallback derived once, dm-cache comprehension, encrypted-attachment metadata pairs, setup info loops (3008->2992) --- plugins/platforms/matrix/adapter.py | 70 +++++++++++------------------ 1 file changed, 27 insertions(+), 43 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 08ce4aadae..2a086397fd 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -152,8 +152,7 @@ def _resolve_matrix_bang_command(name: str) -> str | None: logger.debug("Matrix: is_gateway_known_command failed for %r", name, exc_info=True) try: from agent.skill_commands import get_skill_commands - skill_commands = get_skill_commands() or {} - # Skill command keys are slash-prefixed ("/arxiv"). + skill_commands = get_skill_commands() or {} # keys are slash-prefixed ("/arxiv") for candidate in candidates: if f"/{candidate}" in skill_commands: return candidate @@ -682,20 +681,16 @@ class _CryptoStateStore: return (await self.get_encryption_info(room_id)) is not None async def get_encryption_info(self, room_id: str): - info = None - if hasattr(self._ss, "get_encryption_info"): - info = await self._ss.get_encryption_info(room_id) + info = await self._ss.get_encryption_info(room_id) if hasattr(self._ss, "get_encryption_info") else None if info is not None: return info if room_id in self._enc_info_cache: return self._enc_info_cache[room_id] - client = self._client - if client is None: + if self._client is None: return None try: - from mautrix.types import ( - EventType as _ET, RoomEncryptionStateEventContent as _Enc, RoomID as _RID) - raw = await client.get_state_event(_RID(room_id), _ET.ROOM_ENCRYPTION) + from mautrix.types import EventType as _ET, RoomEncryptionStateEventContent as _Enc, RoomID as _RID + raw = await self._client.get_state_event(_RID(room_id), _ET.ROOM_ENCRYPTION) except Exception as exc: logger.debug("Matrix: homeserver encryption-info query failed for %s: %s", room_id, exc) return None @@ -1745,8 +1740,7 @@ class MatrixAdapter(BasePlatformAdapter): except Exception as exc: if self._closing: return - err_str = str(exc).lower() - if any(k in err_str for k in ("401", "403", "unauthorized", "forbidden")): + if any(k in str(exc).lower() for k in ("401", "403", "unauthorized", "forbidden")): logger.error("Matrix: permanent auth error: %s — stopping sync", exc) return logger.warning("Matrix: sync error: %s — retrying in 5s", exc) @@ -2011,7 +2005,6 @@ class MatrixAdapter(BasePlatformAdapter): if url and not str(url).startswith("mxc://"): logger.warning("[Matrix] Rejecting inbound media %s with non-MXC URL", event_id) return - http_url = self._mxc_to_http(url) if url else "" content_info = source_content.get("info", {}) if not isinstance(content_info, dict): content_info = {} @@ -2031,7 +2024,6 @@ class MatrixAdapter(BasePlatformAdapter): if url and not str(url).startswith("mxc://"): logger.warning("[Matrix] Rejecting inbound encrypted media %s with non-MXC URL", event_id) return - http_url = self._mxc_to_http(url) if url else "" is_encrypted_media = bool(file_content and isinstance(file_content, dict) and file_content.get("url")) msg_type, media_type, is_voice_message = self._classify_inbound_media(msgtype, event_mimetype, source_content) # Cache locally so downstream tools get a real file path. @@ -2043,8 +2035,9 @@ class MatrixAdapter(BasePlatformAdapter): is_voice_message, body) except Exception as e: logger.warning("[Matrix] Failed to cache media: %s", e) - allow_http_fallback = bool(http_url) and not is_encrypted_media - media_urls = [cached_path] if cached_path else ([http_url] if allow_http_fallback else None) + # Unencrypted media may fall back to the HTTP download URL when caching failed. + http_url = self._mxc_to_http(url) if url and not is_encrypted_media else "" + media_urls = [cached_path] if cached_path else ([http_url] if http_url else None) msg_event = await self._build_inbound_event( room_id, sender, event_id, body, source_content, relates_to, message_type=msg_type, media_urls=media_urls, media_types=[media_type] if media_urls else None, media_msgtype=msgtype) @@ -2073,11 +2066,9 @@ class MatrixAdapter(BasePlatformAdapter): return None if encrypted_file is not None: from mautrix.crypto.attachments import decrypt_attachment - hashes_value = encrypted_file.get("hashes") + hashes_value, key_value = encrypted_file.get("hashes"), encrypted_file.get("key") hash_value = hashes_value.get("sha256") if isinstance(hashes_value, dict) else None - key_value = encrypted_file.get("key") - if isinstance(key_value, dict): - key_value = key_value.get("k") + key_value = key_value.get("k") if isinstance(key_value, dict) else key_value iv_value = encrypted_file.get("iv") if not (key_value and hash_value and iv_value): logger.warning("[Matrix] Encrypted media event missing decryption metadata for %s", event_id) @@ -2110,10 +2101,8 @@ class MatrixAdapter(BasePlatformAdapter): async def _join_room_by_id(self, room_id: str) -> bool: """Join a room by ID and refresh local caches on success.""" - if not room_id: - return False - if room_id in self._joined_rooms: - return True + if not room_id or room_id in self._joined_rooms: + return bool(room_id) try: await self._client.join_room(RoomID(room_id)) self._joined_rooms.add(room_id) @@ -2372,7 +2361,7 @@ class MatrixAdapter(BasePlatformAdapter): room_id, target_event_id, "This model picker has expired. Run `/model` again to choose a model.") async def _redact_bot_approval_reactions(self, room_id: str, prompt: Any) -> None: - """Redact the bot's seeded approval reactions, leaving only the user's reaction.""" + """Redact the bot's seeded approval reactions (delayed), leaving only the user's reaction.""" for emoji, evt_id in prompt.bot_reaction_events.items(): self._schedule_reaction_redaction(room_id, evt_id, "approval resolved") logger.debug("Matrix: scheduled bot reaction redaction %s (%s)", emoji, evt_id) @@ -2418,8 +2407,7 @@ class MatrixAdapter(BasePlatformAdapter): if not self._client: return False try: - room = RoomID(room_id) - event = EventID(event_id) + room, event = RoomID(room_id), EventID(event_id) if hasattr(self._client, "set_fully_read_marker"): await self._client.set_fully_read_marker(room, event, event) elif hasattr(self._client, "send_receipt"): @@ -2590,9 +2578,7 @@ class MatrixAdapter(BasePlatformAdapter): return None if hasattr(resp, "content") and (not require_dict or isinstance(resp.content, dict)): return resp.content - if isinstance(resp, dict): - return resp - return None + return resp if isinstance(resp, dict) else None async def _refresh_dm_cache(self) -> None: """Refresh the DM room cache from m.direct account data.""" @@ -2601,10 +2587,7 @@ class MatrixAdapter(BasePlatformAdapter): dm_data = await self._fetch_m_direct(log_failure=True) if dm_data is None: return - dm_room_ids: Set[str] = set() - for rooms in dm_data.values(): - if isinstance(rooms, list): - dm_room_ids.update(str(r) for r in rooms if isinstance(r, str)) + dm_room_ids = {str(r) for rooms in dm_data.values() if isinstance(rooms, list) for r in rooms if isinstance(r, str)} self._dm_rooms = {rid: (rid in dm_room_ids) for rid in self._joined_rooms} self._invalidate_room_identities() @@ -2616,8 +2599,7 @@ class MatrixAdapter(BasePlatformAdapter): return dm_data: Dict[str, list] = await self._fetch_m_direct(require_dict=True) or {} rooms_for_user = dm_data.get(inviter, []) - if not isinstance(rooms_for_user, list): - rooms_for_user = [] + rooms_for_user = rooms_for_user if isinstance(rooms_for_user, list) else [] if room_id not in rooms_for_user: rooms_for_user.append(room_id) dm_data[inviter] = rooms_for_user @@ -2896,9 +2878,10 @@ def interactive_setup() -> None: print_info("Matrix: already configured") if not prompt_yes_no("Reconfigure Matrix?", False): return - print_info("Works with any Matrix homeserver (Synapse, Conduit, Dendrite, or matrix.org).") - print_info(" 1. Create a bot user on your homeserver, or use your own account") - print_info(" 2. Get an access token from Element, or provide user ID + password") + for line in ("Works with any Matrix homeserver (Synapse, Conduit, Dendrite, or matrix.org).", + " 1. Create a bot user on your homeserver, or use your own account", + " 2. Get an access token from Element, or provide user ID + password"): + print_info(line) def _ask(key: str, question: str, **kw) -> str: value = prompt(question, **kw) if value: @@ -2940,10 +2923,11 @@ def interactive_setup() -> None: print_success("Matrix allowlist configured") else: print_info("⚠️ No allowlist set - anyone who can message the bot can use it!") - print_info("📬 Home Room: where Hermes delivers cron job results and notifications.") - print_info(" Room IDs look like !abc123:server (shown in Element room settings)") - print_info(" You can also set this later by typing /set-home in a Matrix room.") - print_info("Leave blank to clear a previously saved home room (cron / notifications).") + for line in ("📬 Home Room: where Hermes delivers cron job results and notifications.", + " Room IDs look like !abc123:server (shown in Element room settings)", + " You can also set this later by typing /set-home in a Matrix room.", + "Leave blank to clear a previously saved home room (cron / notifications)."): + print_info(line) home_room = prompt("Home room ID (leave empty to set later with /set-home)").strip() if home_room: save_env_value("MATRIX_HOME_ROOM", home_room) From a8f6cfaadf9f1f6245e5e8455f5415fcddb14412 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:02:59 -0700 Subject: [PATCH 7/8] refactor(matrix): _send_picker + _handle_picker_reaction shared by model/choice pickers; cross-signing skip reasons as a table --- plugins/platforms/matrix/adapter.py | 87 +++++++++++++++-------------- 1 file changed, 45 insertions(+), 42 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 2a086397fd..2364f83542 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -1171,17 +1171,17 @@ class MatrixAdapter(BasePlatformAdapter): logger.warning("Matrix: cross-signing key lookup failed: %s", exc) if own_xsign is None: _, output_error = _get_matrix_recovery_key_output_target() - skipped = "Matrix: cross-signing keys are missing, but automatic bootstrap is skipped because " - if output_error == "not_configured": + if output_error: + reason = { + "not_configured": "is not configured. Configure MATRIX_RECOVERY_KEY from your Matrix client " + "or set MATRIX_RECOVERY_KEY_OUTPUT_FILE to write a new recovery key once " + "with mode 0600.", + "exists": "already exists and will not be overwritten.", + }.get(output_error, "is not usable: %s") logger.warning( - skipped + "MATRIX_RECOVERY_KEY_OUTPUT_FILE is not configured. Configure " - "MATRIX_RECOVERY_KEY from your Matrix client or set MATRIX_RECOVERY_KEY_OUTPUT_FILE " - "to write a new recovery key once with mode 0600.") - elif output_error == "exists": - logger.warning( - skipped + "MATRIX_RECOVERY_KEY_OUTPUT_FILE already exists and will not be overwritten.") - elif output_error: - logger.warning(skipped + "MATRIX_RECOVERY_KEY_OUTPUT_FILE is not usable: %s", output_error) + "Matrix: cross-signing keys are missing, but automatic bootstrap is skipped because " + "MATRIX_RECOVERY_KEY_OUTPUT_FILE " + reason, + *([output_error] if output_error not in ("not_configured", "exists") else [])) else: try: new_recovery_key = await olm.generate_recovery_key() @@ -1595,12 +1595,20 @@ class MatrixAdapter(BasePlatformAdapter): for emoji, (model_id, provider_slug, provider_name) in zip(_MATRIX_MODEL_PICKER_REACTIONS, flat_choices): choices[emoji] = (model_id, provider_slug) lines.append(f"{emoji} `{model_id}` — {provider_name}") + return await self._send_picker( + chat_id, lines, choices, session_key, on_model_selected, metadata, self._model_picker_prompts_by_event, + "model picker") + + async def _send_picker( + self, chat_id: str, lines: list, choices: dict, session_key: str, on_selected, metadata, registry: dict, + label: str) -> SendResult: + """Send picker *lines*, register a _MatrixPickerPrompt under the event, seed its reactions.""" return await self._send_reaction_prompt( chat_id, "\n".join(lines), metadata, lambda message_id, requester, expires_at: _MatrixPickerPrompt( chat_id=chat_id, message_id=message_id, session_key=session_key, choices=choices, - on_selected=on_model_selected, requester_user_id=requester, expires_at=expires_at), - self._model_picker_prompts_by_event, choices, "model picker") + on_selected=on_selected, requester_user_id=requester, expires_at=expires_at), + registry, choices, label) async def send_choice_picker( self, chat_id: str, title: str, choices: list, session_key: str, on_choice_selected, @@ -1620,12 +1628,9 @@ class MatrixAdapter(BasePlatformAdapter): if not emoji_choices: return SendResult(success=False, error="No choices") lines += ["", "React to choose."] - return await self._send_reaction_prompt( - chat_id, "\n".join(lines), metadata, - lambda message_id, requester, expires_at: _MatrixPickerPrompt( - chat_id=chat_id, message_id=message_id, session_key=session_key, choices=emoji_choices, - on_selected=on_choice_selected, requester_user_id=requester, expires_at=expires_at), - self._choice_picker_prompts_by_event, emoji_choices, "choice picker") + return await self._send_picker( + chat_id, lines, emoji_choices, session_key, on_choice_selected, metadata, + self._choice_picker_prompts_by_event, "choice picker") def format_message(self, content: str) -> str: """Markdown passes through; strip image markdown (media is uploaded separately).""" @@ -2274,42 +2279,40 @@ class MatrixAdapter(BasePlatformAdapter): async def _handle_model_picker_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Apply a model-picker reaction. True if the reaction targeted a pending picker.""" - handled, model_prompt, selection = await self._claim_reaction_prompt( + return await self._handle_picker_reaction( self._model_picker_prompts_by_event, room_id, reacts_to, key, sender, "model picker", - "That reaction is not one of the available model choices.", self._expire_matrix_model_picker_prompt) - if selection is None: - return handled - model_prompt.resolved = True - self._model_picker_prompts_by_event.pop(reacts_to, None) - model_id, provider_slug = selection - try: - confirmation = await model_prompt.on_selected(room_id, model_id, provider_slug) - await self._redact_bot_model_picker_reactions(room_id, model_prompt) - if confirmation: - await self.send(room_id, confirmation, reply_to=reacts_to) - except Exception as exc: - logger.error("Failed to switch model from Matrix reaction: %s", exc) - await self.send(room_id, f"Failed to switch model: {exc}", reply_to=reacts_to) - return True + "That reaction is not one of the available model choices.", self._expire_matrix_model_picker_prompt, + ("switch model", "switch model"), redact_bot_reactions=True) async def _handle_choice_picker_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Apply a choice-picker reaction. True if the reaction targeted a pending picker.""" async def _expire(_room_id, target_event_id, _prompt): self._choice_picker_prompts_by_event.pop(target_event_id, None) - handled, choice_prompt, value = await self._claim_reaction_prompt( + return await self._handle_picker_reaction( self._choice_picker_prompts_by_event, room_id, reacts_to, key, sender, "choice picker", - "That reaction is not one of the available choices.", _expire) - if value is None: + "That reaction is not one of the available choices.", _expire, ("apply choice", "apply selection")) + + async def _handle_picker_reaction( + self, registry: dict, room_id: str, reacts_to: str, key: str, sender: str, label: str, invalid_text: str, + on_expired, verbs: tuple[str, str], *, redact_bot_reactions: bool = False) -> bool: + """Claim the picker, fire ``on_selected(room_id, *selection)`` and post its confirmation (or the error). + ``verbs`` = (log verb, user-facing verb).""" + handled, prompt, selection = await self._claim_reaction_prompt( + registry, room_id, reacts_to, key, sender, label, invalid_text, on_expired) + if selection is None: return handled - choice_prompt.resolved = True - self._choice_picker_prompts_by_event.pop(reacts_to, None) + prompt.resolved = True + registry.pop(reacts_to, None) + args = selection if isinstance(selection, tuple) else (selection,) try: - confirmation = await choice_prompt.on_selected(room_id, value) + confirmation = await prompt.on_selected(room_id, *args) + if redact_bot_reactions: + await self._redact_bot_model_picker_reactions(room_id, prompt) if confirmation: await self.send(room_id, confirmation, reply_to=reacts_to) except Exception as exc: - logger.error("Failed to apply choice from Matrix reaction: %s", exc) - await self.send(room_id, f"Failed to apply selection: {exc}", reply_to=reacts_to) + logger.error("Failed to %s from Matrix reaction: %s", verbs[0], exc) + await self.send(room_id, f"Failed to {verbs[1]}: {exc}", reply_to=reacts_to) return True def _matrix_prompt_expired(self, prompt: Any) -> bool: From 9ae8614c65bd7ab9b275b8a0e7468af761997a7f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:05:24 -0700 Subject: [PATCH 8/8] refactor(matrix): drop docstrings that restate the method name (base ABC documents the public surface) (2995->2954) --- plugins/platforms/matrix/adapter.py | 41 ----------------------------- 1 file changed, 41 deletions(-) diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 2364f83542..64e30a131e 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -446,7 +446,6 @@ def _check_e2ee_deps() -> bool: def _normalize_e2ee_mode(value: Any) -> str: - """Normalize Matrix E2EE mode to off/optional/required.""" raw = str(value or "").strip().lower() if raw in ("required", "require", "true", "1", "yes", "on"): return "required" @@ -848,7 +847,6 @@ class MatrixAdapter(BasePlatformAdapter): @staticmethod def _extract_server_ed25519(device_keys_obj: Any) -> Optional[str]: - """Extract the ed25519 identity key from a DeviceKeys object.""" for kid, kval in (getattr(device_keys_obj, "keys", {}) or {}).items(): if str(kid).startswith("ed25519:"): return str(kval) @@ -1204,7 +1202,6 @@ class MatrixAdapter(BasePlatformAdapter): logger.warning("Matrix: initial sync error: %s", exc) async def connect(self, *, is_reconnect: bool = False) -> bool: - """Connect to the Matrix homeserver and start syncing.""" self._device_id_unverified = False if self._client is not None: try: @@ -1252,7 +1249,6 @@ class MatrixAdapter(BasePlatformAdapter): return True async def disconnect(self) -> None: - """Disconnect from Matrix.""" self._closing = True if self._sync_task and not self._sync_task.done(): self._sync_task.cancel() @@ -1283,7 +1279,6 @@ class MatrixAdapter(BasePlatformAdapter): async def send( self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> SendResult: - """Send a message to a Matrix room.""" if not content: return SendResult(success=True) last_event_id = None @@ -1313,12 +1308,10 @@ class MatrixAdapter(BasePlatformAdapter): return str(event_id) async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: - """Return room name and type (dm/group).""" identity = await self._resolve_room_identity(chat_id) return {"name": identity.display_name, "type": "dm" if identity.chat_type == "dm" else "group"} def get_diagnostics(self) -> Dict[str, Any]: - """Return redacted Matrix readiness/status diagnostics.""" now = time.time() token_present = bool(self._access_token) user_id = self._user_id or getattr(self._client, "mxid", "") or "" @@ -1350,15 +1343,12 @@ class MatrixAdapter(BasePlatformAdapter): await self._client.set_typing(RoomID(chat_id), timeout=timeout) async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None: - """Send a typing indicator.""" await self._set_typing(chat_id, 30000) async def stop_typing(self, chat_id: str) -> None: - """Clear the typing indicator.""" await self._set_typing(chat_id, 0) async def edit_message(self, chat_id: str, message_id: str, content: str, *, finalize: bool = False) -> SendResult: - """Edit an existing message (via m.replace).""" formatted = self.format_message(content) new_content = self._build_text_message_content(formatted) msg_content: Dict[str, Any] = {"msgtype": "m.text", "body": f"* {formatted}", "m.new_content": new_content} @@ -1373,7 +1363,6 @@ class MatrixAdapter(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: - """Download an image URL and upload it to Matrix.""" from tools.url_safety import is_safe_url if not is_safe_url(image_url): logger.warning("Matrix: blocked unsafe image URL (SSRF protection)") @@ -1452,13 +1441,11 @@ class MatrixAdapter(BasePlatformAdapter): async def send_image_file( self, chat_id: str, image_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> SendResult: - """Upload a local image file to Matrix.""" return await self._send_local_file(chat_id, image_path, "m.image", caption, reply_to, metadata=metadata) async def send_multiple_images( self, chat_id: str, images: list[tuple[str, str]], metadata: Optional[Dict[str, Any]] = None, human_delay: float = 0.0) -> None: - """Send multiple Matrix images as one ordered logical batch.""" if not images: return from urllib.parse import unquote as _unquote @@ -1478,7 +1465,6 @@ class MatrixAdapter(BasePlatformAdapter): async def send_document( self, chat_id: str, file_path: str, caption: Optional[str] = None, file_name: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> SendResult: - """Upload a local file as a document.""" return await self._send_local_file(chat_id, file_path, "m.file", caption, reply_to, file_name, metadata) async def send_voice( @@ -1503,7 +1489,6 @@ class MatrixAdapter(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) -> SendResult: - """Upload a video file.""" return await self._send_local_file(chat_id, video_path, "m.video", caption, reply_to, metadata=metadata) # Template attrs for the shared _format_exec_approval core (header + fence + reason only; @@ -1536,7 +1521,6 @@ class MatrixAdapter(BasePlatformAdapter): self, chat_id: str, command: str, session_key: str, description: str = "dangerous command", metadata: Optional[dict] = None, allow_permanent: bool = True, allow_session: bool = True, smart_denied: bool = False) -> SendResult: - """Send a reaction-based exec approval prompt for Matrix.""" if not self._client: return SendResult(success=False, error="Not connected") if smart_denied: @@ -1574,7 +1558,6 @@ class MatrixAdapter(BasePlatformAdapter): async def send_model_picker( self, chat_id: str, providers: list, current_model: str, current_provider: str, session_key: str, on_model_selected, metadata: Optional[Dict[str, Any]] = None) -> SendResult: - """Send a Matrix reaction-based model picker.""" if not self._client: return SendResult(success=False, error="Not connected") flat_choices = [ @@ -1640,7 +1623,6 @@ class MatrixAdapter(BasePlatformAdapter): self, room_id: str, data: bytes, filename: str, content_type: str, msgtype: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, is_voice: bool = False, voice_metadata: Optional[Dict[str, Any]] = None) -> SendResult: - """Upload bytes to Matrix and send as a media message.""" if len(data) > self._max_media_bytes: return self._media_too_large(len(data)) upload_data = data @@ -1703,7 +1685,6 @@ class MatrixAdapter(BasePlatformAdapter): self, room_id: str, file_path: str, msgtype: str, caption: Optional[str] = None, reply_to: Optional[str] = None, file_name: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, is_voice: bool = False) -> SendResult: - """Read a local file and upload it.""" p = Path(file_path).expanduser() if not p.exists(): # file_path is host-local; never echo it into chat. @@ -1725,7 +1706,6 @@ class MatrixAdapter(BasePlatformAdapter): caption, reply_to, metadata, is_voice, voice_metadata) async def _sync_loop(self) -> None: - """Continuously sync with the homeserver.""" client = self._client next_batch = await client.sync_store.get_next_batch() # resume from the initial sync while not self._closing: @@ -1843,7 +1823,6 @@ class MatrixAdapter(BasePlatformAdapter): self._clock_skew_warned = True async def _on_room_message(self, event: Any) -> None: - """Handle incoming room message events (text, media).""" room_id = str(getattr(event, "room_id", "")) sender = str(getattr(event, "sender", "")) # DEBUG-level proof the callback fires at all (silent-inbound troubleshooting). @@ -1988,7 +1967,6 @@ class MatrixAdapter(BasePlatformAdapter): async def _handle_text_message( self, room_id: str, sender: str, event_id: str, event_ts: float, source_content: dict, relates_to: dict) -> None: - """Process a text message event.""" body = source_content.get("body", "") or "" if not body: return @@ -2004,7 +1982,6 @@ class MatrixAdapter(BasePlatformAdapter): async def _handle_media_message( self, room_id: str, sender: str, event_id: str, event_ts: float, source_content: dict, relates_to: dict, msgtype: str) -> None: - """Process a media message event (image, audio, video, file).""" body = source_content.get("body", "") or "" url = source_content.get("url", "") if url and not str(url).startswith("mxc://"): @@ -2105,7 +2082,6 @@ class MatrixAdapter(BasePlatformAdapter): self._schedule_invite_join(room_id, is_direct=is_direct and bool(inviter), inviter=inviter) async def _join_room_by_id(self, room_id: str) -> bool: - """Join a room by ID and refresh local caches on success.""" if not room_id or room_id in self._joined_rooms: return bool(room_id) try: @@ -2168,7 +2144,6 @@ class MatrixAdapter(BasePlatformAdapter): return None async def _redact_reaction(self, room_id: str, reaction_event_id: str, reason: str = "") -> bool: - """Remove a reaction by redacting its event.""" return await self.redact_message(room_id, reaction_event_id, reason) def _schedule_reaction_redaction(self, room_id: str, reaction_event_id: str, reason: str = "") -> None: @@ -2189,7 +2164,6 @@ class MatrixAdapter(BasePlatformAdapter): task.add_done_callback(self._reaction_redaction_tasks.discard) async def on_processing_start(self, event: MessageEvent) -> None: - """Add eyes reaction when the agent starts processing a message.""" msg_id, room_id = event.message_id, event.source.chat_id if self._reactions_enabled and msg_id and room_id: reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440") @@ -2197,7 +2171,6 @@ class MatrixAdapter(BasePlatformAdapter): self._pending_reactions[(room_id, msg_id)] = reaction_event_id async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: - """Replace eyes with checkmark (success) or cross (failure).""" msg_id, room_id = event.message_id, event.source.chat_id if not self._reactions_enabled or not msg_id or not room_id or outcome == ProcessingOutcome.CANCELLED: return @@ -2207,7 +2180,6 @@ class MatrixAdapter(BasePlatformAdapter): await self._send_reaction(room_id, msg_id, "\u2705" if outcome == ProcessingOutcome.SUCCESS else "\u274c") async def _on_reaction(self, event: Any) -> None: - """Handle incoming reaction events.""" sender = str(getattr(event, "sender", "")) if self._is_self_sender(sender): return @@ -2370,7 +2342,6 @@ class MatrixAdapter(BasePlatformAdapter): logger.debug("Matrix: scheduled bot reaction redaction %s (%s)", emoji, evt_id) async def _redact_bot_model_picker_reactions(self, room_id: str, prompt: Any) -> None: - """Redact the bot's seeded model picker reactions.""" for emoji, evt_id in prompt.bot_reaction_events.items(): try: await self.redact_message(room_id, evt_id, "model picker resolved") @@ -2396,7 +2367,6 @@ class MatrixAdapter(BasePlatformAdapter): self._pending_text_batch_tasks.pop(key, None) def _background_read_receipt(self, room_id: str, event_id: str) -> None: - """Fire-and-forget read receipt with error logging.""" async def _send() -> None: try: @@ -2406,7 +2376,6 @@ class MatrixAdapter(BasePlatformAdapter): asyncio.ensure_future(_send()) async def send_read_receipt(self, room_id: str, event_id: str) -> bool: - """Send a read receipt (m.read) for an event.""" if not self._client: return False try: @@ -2439,7 +2408,6 @@ class MatrixAdapter(BasePlatformAdapter): return False async def redact_message(self, room_id: str, event_id: str, reason: str = "") -> bool: - """Redact (delete) a message or event from a room.""" return await self._client_op( lambda: self._client.redact(RoomID(room_id), EventID(event_id), reason=reason or None), ("Matrix: redacted %s in %s", event_id, room_id), "Matrix: redact error: %s") @@ -2447,7 +2415,6 @@ class MatrixAdapter(BasePlatformAdapter): async def create_room( self, name: str = "", topic: str = "", invite: Optional[list] = None, is_direct: bool = False, preset: str = "private_chat") -> Optional[str]: - """Create a new Matrix room.""" if not self._client: return None if preset == "public_chat" and not _env_truthy("MATRIX_ALLOW_PUBLIC_ROOMS"): @@ -2469,7 +2436,6 @@ class MatrixAdapter(BasePlatformAdapter): return None async def invite_user(self, room_id: str, user_id: str) -> bool: - """Invite a user to a room.""" return await self._client_op( lambda: self._client.invite_user(RoomID(room_id), UserID(user_id)), ("Matrix: invited %s to %s", user_id, room_id), "Matrix: invite error: %s") @@ -2477,7 +2443,6 @@ class MatrixAdapter(BasePlatformAdapter): _VALID_PRESENCE_STATES = frozenset(("online", "offline", "unavailable")) async def set_presence(self, state: str = "online", status_msg: str = "") -> bool: - """Set the bot's presence status.""" if not self._client: return False if state not in self._VALID_PRESENCE_STATES: @@ -2568,7 +2533,6 @@ class MatrixAdapter(BasePlatformAdapter): return identity async def _is_dm_room(self, room_id: str) -> bool: - """Check if a room is a DM.""" return (await self._resolve_room_identity(room_id)).chat_type == "dm" async def _fetch_m_direct(self, *, log_failure: bool = False, require_dict: bool = False): @@ -2584,7 +2548,6 @@ class MatrixAdapter(BasePlatformAdapter): return resp if isinstance(resp, dict) else None async def _refresh_dm_cache(self) -> None: - """Refresh the DM room cache from m.direct account data.""" if not self._client: return dm_data = await self._fetch_m_direct(log_failure=True) @@ -2646,7 +2609,6 @@ class MatrixAdapter(BasePlatformAdapter): msg_content["m.relates_to"] = relates_to def _extract_outbound_mentions(self, text: str) -> list[str]: - """Return unique Matrix user IDs mentioned in outbound text.""" protected, _ = self._protect_outbound_mention_regions(text) return list(dict.fromkeys(m.group(1) for m in _OUTBOUND_MENTION_RE.finditer(protected))) @@ -2725,7 +2687,6 @@ class MatrixAdapter(BasePlatformAdapter): return user_id def _mxc_to_http(self, mxc_url: str) -> str: - """Convert mxc://server/media_id to an HTTP download URL.""" if not mxc_url.startswith("mxc://"): return mxc_url return f"{self._homeserver}/_matrix/client/v1/media/download/{mxc_url[6:]}" @@ -2747,7 +2708,6 @@ class MatrixAdapter(BasePlatformAdapter): @staticmethod def _sanitize_link_url(url: str) -> str: - """Sanitize a URL for use in an href attribute.""" stripped = url.strip() if ":" in stripped and stripped.split(":", 1)[0].lower().strip() in {"javascript", "data", "vbscript"}: return "" @@ -2981,7 +2941,6 @@ def _build_adapter(config): def register(ctx) -> None: - """Plugin entry point — called by the Hermes plugin system.""" ctx.register_platform( name="matrix", label="Matrix", adapter_factory=_build_adapter, check_fn=matrix_deps_present, ensure_deps_fn=ensure_matrix_deps, is_connected=_is_connected,