| 1 | """Matrix (Element) channel — inbound sync + outbound message/media delivery.""" |
| 2 | |
| 3 | import asyncio |
| 4 | import json |
| 5 | import logging |
| 6 | import mimetypes |
| 7 | import time |
| 8 | from dataclasses import dataclass |
| 9 | from pathlib import Path |
| 10 | from typing import Any, Literal, TypeAlias |
| 11 | |
| 12 | from loguru import logger |
| 13 | from pydantic import Field |
| 14 | |
| 15 | try: |
| 16 | import nh3 |
| 17 | from mistune import HTMLRenderer, create_markdown |
| 18 | from nio import ( |
| 19 | AsyncClient, |
| 20 | AsyncClientConfig, |
| 21 | DownloadError, |
| 22 | InviteEvent, |
| 23 | JoinError, |
| 24 | LoginResponse, |
| 25 | MatrixRoom, |
| 26 | MemoryDownloadResponse, |
| 27 | RoomEncryptedMedia, |
| 28 | RoomMessage, |
| 29 | RoomMessageMedia, |
| 30 | RoomMessageText, |
| 31 | RoomSendError, |
| 32 | RoomSendResponse, |
| 33 | RoomTypingError, |
| 34 | SyncError, |
| 35 | UploadError, |
| 36 | ) |
| 37 | from nio.crypto.attachments import decrypt_attachment |
| 38 | from nio.exceptions import EncryptionError |
| 39 | except ImportError as e: |
| 40 | raise ImportError( |
| 41 | "Matrix dependencies not installed. Run: pip install echo-director-agent[matrix]" |
| 42 | ) from e |
| 43 | |
| 44 | from nanobot.bus.events import OutboundMessage |
| 45 | from nanobot.bus.queue import MessageBus |
| 46 | from nanobot.channels.base import BaseChannel |
| 47 | from nanobot.config.paths import get_data_dir, get_media_dir |
| 48 | from nanobot.config.schema import Base |
| 49 | from nanobot.utils.helpers import safe_filename |
| 50 | |
| 51 | TYPING_NOTICE_TIMEOUT_MS = 30_000 |
| 52 | # Must stay below TYPING_NOTICE_TIMEOUT_MS so the indicator doesn't expire mid-processing. |
| 53 | TYPING_KEEPALIVE_INTERVAL_MS = 20_000 |
| 54 | MATRIX_HTML_FORMAT = "org.matrix.custom.html" |
| 55 | _ATTACH_MARKER = "[attachment: {}]" |
| 56 | _ATTACH_TOO_LARGE = "[attachment: {} - too large]" |
| 57 | _ATTACH_FAILED = "[attachment: {} - download failed]" |
| 58 | _ATTACH_UPLOAD_FAILED = "[attachment: {} - upload failed]" |
| 59 | _DEFAULT_ATTACH_NAME = "attachment" |
| 60 | _MSGTYPE_MAP = {"m.image": "image", "m.audio": "audio", "m.video": "video", "m.file": "file"} |
| 61 | |
| 62 | MATRIX_MEDIA_EVENT_FILTER = (RoomMessageMedia, RoomEncryptedMedia) |
| 63 | MatrixMediaEvent: TypeAlias = RoomMessageMedia | RoomEncryptedMedia |
| 64 | |
| 65 | MATRIX_MARKDOWN = create_markdown( |
| 66 | renderer=HTMLRenderer(escape=True, allow_harmful_protocols={"mxc:"}), |
| 67 | escape=True, |
| 68 | plugins=["table", "strikethrough", "url", "superscript", "subscript"], |
| 69 | ) |
| 70 | |
| 71 | MATRIX_ALLOWED_HTML_TAGS = { |
| 72 | "p", "a", "strong", "em", "del", "code", "pre", "blockquote", |
| 73 | "ul", "ol", "li", "h1", "h2", "h3", "h4", "h5", "h6", |
| 74 | "hr", "br", "table", "thead", "tbody", "tr", "th", "td", |
| 75 | "caption", "sup", "sub", "img", |
| 76 | } |
| 77 | MATRIX_ALLOWED_HTML_ATTRIBUTES: dict[str, set[str]] = { |
| 78 | "a": {"href"}, "code": {"class"}, "ol": {"start"}, |
| 79 | "img": {"src", "alt", "title", "width", "height"}, |
| 80 | } |
| 81 | MATRIX_ALLOWED_URL_SCHEMES = {"https", "http", "matrix", "mailto", "mxc"} |
| 82 | |
| 83 | |
| 84 | def _filter_matrix_html_attribute(tag: str, attr: str, value: str) -> str | None: |
| 85 | """Filter attribute values to a safe Matrix-compatible subset.""" |
| 86 | if tag == "a" and attr == "href": |
| 87 | return value if value.lower().startswith(("https://", "http://", "matrix:", "mailto:")) else None |
| 88 | if tag == "img" and attr == "src": |
| 89 | return value if value.lower().startswith("mxc://") else None |
| 90 | if tag == "code" and attr == "class": |
| 91 | classes = [c for c in value.split() if c.startswith("language-") and not c.startswith("language-_")] |
| 92 | return " ".join(classes) if classes else None |
| 93 | return value |
| 94 | |
| 95 | |
| 96 | MATRIX_HTML_CLEANER = nh3.Cleaner( |
| 97 | tags=MATRIX_ALLOWED_HTML_TAGS, |
| 98 | attributes=MATRIX_ALLOWED_HTML_ATTRIBUTES, |
| 99 | attribute_filter=_filter_matrix_html_attribute, |
| 100 | url_schemes=MATRIX_ALLOWED_URL_SCHEMES, |
| 101 | strip_comments=True, |
| 102 | link_rel="noopener noreferrer", |
| 103 | ) |
| 104 | |
| 105 | @dataclass |
| 106 | class _StreamBuf: |
| 107 | """ |
| 108 | Represents a buffer for managing LLM response stream data. |
| 109 | |
| 110 | :ivar text: Stores the text content of the buffer. |
| 111 | :type text: str |
| 112 | :ivar event_id: Identifier for the associated event. None indicates no |
| 113 | specific event association. |
| 114 | :type event_id: str | None |
| 115 | :ivar last_edit: Timestamp of the most recent edit to the buffer. |
| 116 | :type last_edit: float |
| 117 | """ |
| 118 | text: str = "" |
| 119 | event_id: str | None = None |
| 120 | last_edit: float = 0.0 |
| 121 | |
| 122 | def _render_markdown_html(text: str) -> str | None: |
| 123 | """Render markdown to sanitized HTML; returns None for plain text.""" |
| 124 | try: |
| 125 | formatted = MATRIX_HTML_CLEANER.clean(MATRIX_MARKDOWN(text)).strip() |
| 126 | except Exception: |
| 127 | return None |
| 128 | if not formatted: |
| 129 | return None |
| 130 | # Skip formatted_body for plain <p>text</p> to keep payload minimal. |
| 131 | if formatted.startswith("<p>") and formatted.endswith("</p>"): |
| 132 | inner = formatted[3:-4] |
| 133 | if "<" not in inner and ">" not in inner: |
| 134 | return None |
| 135 | return formatted |
| 136 | |
| 137 | |
| 138 | def _build_matrix_text_content( |
| 139 | text: str, |
| 140 | event_id: str | None = None, |
| 141 | thread_relates_to: dict[str, object] | None = None, |
| 142 | ) -> dict[str, object]: |
| 143 | """ |
| 144 | Constructs and returns a dictionary representing the matrix text content with optional |
| 145 | HTML formatting and reference to an existing event for replacement. This function is |
| 146 | primarily used to create content payloads compatible with the Matrix messaging protocol. |
| 147 | |
| 148 | :param text: The plain text content to include in the message. |
| 149 | :type text: str |
| 150 | :param event_id: Optional ID of the event to replace. If provided, the function will |
| 151 | include information indicating that the message is a replacement of the specified |
| 152 | event. |
| 153 | :type event_id: str | None |
| 154 | :param thread_relates_to: Optional Matrix thread relation metadata. For edits this is |
| 155 | stored in ``m.new_content`` so the replacement remains in the same thread. |
| 156 | :type thread_relates_to: dict[str, object] | None |
| 157 | :return: A dictionary containing the matrix text content, potentially enriched with |
| 158 | HTML formatting and replacement metadata if applicable. |
| 159 | :rtype: dict[str, object] |
| 160 | """ |
| 161 | content: dict[str, object] = {"msgtype": "m.text", "body": text, "m.mentions": {}} |
| 162 | if html := _render_markdown_html(text): |
| 163 | content["format"] = MATRIX_HTML_FORMAT |
| 164 | content["formatted_body"] = html |
| 165 | if event_id: |
| 166 | content["m.new_content"] = { |
| 167 | "body": text, |
| 168 | "msgtype": "m.text", |
| 169 | } |
| 170 | content["m.relates_to"] = { |
| 171 | "rel_type": "m.replace", |
| 172 | "event_id": event_id, |
| 173 | } |
| 174 | if thread_relates_to: |
| 175 | content["m.new_content"]["m.relates_to"] = thread_relates_to |
| 176 | elif thread_relates_to: |
| 177 | content["m.relates_to"] = thread_relates_to |
| 178 | |
| 179 | return content |
| 180 | |
| 181 | |
| 182 | class _NioLoguruHandler(logging.Handler): |
| 183 | """Route matrix-nio stdlib logs into Loguru.""" |
| 184 | |
| 185 | def emit(self, record: logging.LogRecord) -> None: |
| 186 | try: |
| 187 | level = logger.level(record.levelname).name |
| 188 | except ValueError: |
| 189 | level = record.levelno |
| 190 | frame, depth = logging.currentframe(), 2 |
| 191 | while frame and frame.f_code.co_filename == logging.__file__: |
| 192 | frame, depth = frame.f_back, depth + 1 |
| 193 | logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage()) |
| 194 | |
| 195 | |
| 196 | def _configure_nio_logging_bridge() -> None: |
| 197 | """Bridge matrix-nio logs to Loguru (idempotent).""" |
| 198 | nio_logger = logging.getLogger("nio") |
| 199 | if not any(isinstance(h, _NioLoguruHandler) for h in nio_logger.handlers): |
| 200 | nio_logger.handlers = [_NioLoguruHandler()] |
| 201 | nio_logger.propagate = False |
| 202 | |
| 203 | |
| 204 | class MatrixConfig(Base): |
| 205 | """Matrix (Element) channel configuration.""" |
| 206 | |
| 207 | enabled: bool = False |
| 208 | homeserver: str = "https://matrix.org" |
| 209 | user_id: str = "" |
| 210 | password: str = "" |
| 211 | access_token: str = "" |
| 212 | device_id: str = "" |
| 213 | e2ee_enabled: bool = Field(default=True, alias="e2eeEnabled") |
| 214 | sync_stop_grace_seconds: int = 2 |
| 215 | max_media_bytes: int = 20 * 1024 * 1024 |
| 216 | allow_from: list[str] = Field(default_factory=list) |
| 217 | group_policy: Literal["open", "mention", "allowlist"] = "open" |
| 218 | group_allow_from: list[str] = Field(default_factory=list) |
| 219 | allow_room_mentions: bool = False, |
| 220 | streaming: bool = False |
| 221 | |
| 222 | |
| 223 | class MatrixChannel(BaseChannel): |
| 224 | """Matrix (Element) channel using long-polling sync.""" |
| 225 | |
| 226 | name = "matrix" |
| 227 | display_name = "Matrix" |
| 228 | _STREAM_EDIT_INTERVAL = 2 # min seconds between edit_message_text calls |
| 229 | monotonic_time = time.monotonic |
| 230 | |
| 231 | @classmethod |
| 232 | def default_config(cls) -> dict[str, Any]: |
| 233 | return MatrixConfig().model_dump(by_alias=True) |
| 234 | |
| 235 | def __init__( |
| 236 | self, |
| 237 | config: Any, |
| 238 | bus: MessageBus, |
| 239 | *, |
| 240 | restrict_to_workspace: bool = False, |
| 241 | workspace: str | Path | None = None, |
| 242 | ): |
| 243 | if isinstance(config, dict): |
| 244 | config = MatrixConfig.model_validate(config) |
| 245 | super().__init__(config, bus) |
| 246 | self.client: AsyncClient | None = None |
| 247 | self._sync_task: asyncio.Task | None = None |
| 248 | self._typing_tasks: dict[str, asyncio.Task] = {} |
| 249 | self._restrict_to_workspace = bool(restrict_to_workspace) |
| 250 | self._workspace = ( |
| 251 | Path(workspace).expanduser().resolve(strict=False) if workspace is not None else None |
| 252 | ) |
| 253 | self._server_upload_limit_bytes: int | None = None |
| 254 | self._server_upload_limit_checked = False |
| 255 | self._stream_bufs: dict[str, _StreamBuf] = {} |
| 256 | |
| 257 | |
| 258 | async def start(self) -> None: |
| 259 | """Start Matrix client and begin sync loop.""" |
| 260 | self._running = True |
| 261 | _configure_nio_logging_bridge() |
| 262 | |
| 263 | self.store_path = get_data_dir() / "matrix-store" |
| 264 | self.store_path.mkdir(parents=True, exist_ok=True) |
| 265 | self.session_path = self.store_path / "session.json" |
| 266 | |
| 267 | self.client = AsyncClient( |
| 268 | homeserver=self.config.homeserver, user=self.config.user_id, |
| 269 | store_path=self.store_path, |
| 270 | config=AsyncClientConfig(store_sync_tokens=True, encryption_enabled=self.config.e2ee_enabled), |
| 271 | ) |
| 272 | |
| 273 | self._register_event_callbacks() |
| 274 | self._register_response_callbacks() |
| 275 | |
| 276 | if not self.config.e2ee_enabled: |
| 277 | logger.warning("Matrix E2EE disabled; encrypted rooms may be undecryptable.") |
| 278 | |
| 279 | if self.config.password: |
| 280 | if self.config.access_token or self.config.device_id: |
| 281 | logger.warning("Password-based Matrix login active; access_token and device_id fields will be ignored.") |
| 282 | |
| 283 | create_new_session = True |
| 284 | if self.session_path.exists(): |
| 285 | logger.info("Found session.json at {}; attempting to use existing session...", self.session_path) |
| 286 | try: |
| 287 | with open(self.session_path, "r", encoding="utf-8") as f: |
| 288 | session = json.load(f) |
| 289 | self.client.user_id = self.config.user_id |
| 290 | self.client.access_token = session["access_token"] |
| 291 | self.client.device_id = session["device_id"] |
| 292 | self.client.load_store() |
| 293 | logger.info("Successfully loaded from existing session") |
| 294 | create_new_session = False |
| 295 | except Exception as e: |
| 296 | logger.warning("Failed to load from existing session: {}", e) |
| 297 | logger.info("Falling back to password login...") |
| 298 | |
| 299 | if create_new_session: |
| 300 | logger.info("Using password login...") |
| 301 | resp = await self.client.login(self.config.password) |
| 302 | if isinstance(resp, LoginResponse): |
| 303 | logger.info("Logged in using a password; saving details to disk") |
| 304 | self._write_session_to_disk(resp) |
| 305 | else: |
| 306 | logger.error("Failed to log in: {}", resp) |
| 307 | return |
| 308 | |
| 309 | elif self.config.access_token and self.config.device_id: |
| 310 | try: |
| 311 | self.client.user_id = self.config.user_id |
| 312 | self.client.access_token = self.config.access_token |
| 313 | self.client.device_id = self.config.device_id |
| 314 | self.client.load_store() |
| 315 | logger.info("Successfully loaded from existing session") |
| 316 | except Exception as e: |
| 317 | logger.warning("Failed to load from existing session: {}", e) |
| 318 | |
| 319 | else: |
| 320 | logger.warning("Unable to load a Matrix session due to missing password, access_token, or device_id; encryption may not work") |
| 321 | return |
| 322 | |
| 323 | self._sync_task = asyncio.create_task(self._sync_loop()) |
| 324 | |
| 325 | async def stop(self) -> None: |
| 326 | """Stop the Matrix channel with graceful sync shutdown.""" |
| 327 | self._running = False |
| 328 | for room_id in list(self._typing_tasks): |
| 329 | await self._stop_typing_keepalive(room_id, clear_typing=False) |
| 330 | if self.client: |
| 331 | self.client.stop_sync_forever() |
| 332 | if self._sync_task: |
| 333 | try: |
| 334 | await asyncio.wait_for(asyncio.shield(self._sync_task), |
| 335 | timeout=self.config.sync_stop_grace_seconds) |
| 336 | except (asyncio.TimeoutError, asyncio.CancelledError): |
| 337 | self._sync_task.cancel() |
| 338 | try: |
| 339 | await self._sync_task |
| 340 | except asyncio.CancelledError: |
| 341 | pass |
| 342 | if self.client: |
| 343 | await self.client.close() |
| 344 | |
| 345 | def _write_session_to_disk(self, resp: LoginResponse) -> None: |
| 346 | """Save login session to disk for persistence across restarts.""" |
| 347 | session = { |
| 348 | "access_token": resp.access_token, |
| 349 | "device_id": resp.device_id, |
| 350 | } |
| 351 | try: |
| 352 | with open(self.session_path, "w", encoding="utf-8") as f: |
| 353 | json.dump(session, f, indent=2) |
| 354 | logger.info("Session saved to {}", self.session_path) |
| 355 | except Exception as e: |
| 356 | logger.warning("Failed to save session: {}", e) |
| 357 | |
| 358 | def _is_workspace_path_allowed(self, path: Path) -> bool: |
| 359 | """Check path is inside workspace (when restriction enabled).""" |
| 360 | if not self._restrict_to_workspace or not self._workspace: |
| 361 | return True |
| 362 | try: |
| 363 | path.resolve(strict=False).relative_to(self._workspace) |
| 364 | return True |
| 365 | except ValueError: |
| 366 | return False |
| 367 | |
| 368 | def _collect_outbound_media_candidates(self, media: list[str]) -> list[Path]: |
| 369 | """Deduplicate and resolve outbound attachment paths.""" |
| 370 | seen: set[str] = set() |
| 371 | candidates: list[Path] = [] |
| 372 | for raw in media: |
| 373 | if not isinstance(raw, str) or not raw.strip(): |
| 374 | continue |
| 375 | path = Path(raw.strip()).expanduser() |
| 376 | try: |
| 377 | key = str(path.resolve(strict=False)) |
| 378 | except OSError: |
| 379 | key = str(path) |
| 380 | if key not in seen: |
| 381 | seen.add(key) |
| 382 | candidates.append(path) |
| 383 | return candidates |
| 384 | |
| 385 | @staticmethod |
| 386 | def _build_outbound_attachment_content( |
| 387 | *, filename: str, mime: str, size_bytes: int, |
| 388 | mxc_url: str, encryption_info: dict[str, Any] | None = None, |
| 389 | ) -> dict[str, Any]: |
| 390 | """Build Matrix content payload for an uploaded file/image/audio/video.""" |
| 391 | prefix = mime.split("/")[0] |
| 392 | msgtype = {"image": "m.image", "audio": "m.audio", "video": "m.video"}.get(prefix, "m.file") |
| 393 | content: dict[str, Any] = { |
| 394 | "msgtype": msgtype, "body": filename, "filename": filename, |
| 395 | "info": {"mimetype": mime, "size": size_bytes}, "m.mentions": {}, |
| 396 | } |
| 397 | if encryption_info: |
| 398 | content["file"] = {**encryption_info, "url": mxc_url} |
| 399 | else: |
| 400 | content["url"] = mxc_url |
| 401 | return content |
| 402 | |
| 403 | def _is_encrypted_room(self, room_id: str) -> bool: |
| 404 | if not self.client: |
| 405 | return False |
| 406 | room = getattr(self.client, "rooms", {}).get(room_id) |
| 407 | return bool(getattr(room, "encrypted", False)) |
| 408 | |
| 409 | async def _send_room_content(self, room_id: str, |
| 410 | content: dict[str, Any]) -> None | RoomSendResponse | RoomSendError: |
| 411 | """Send m.room.message with E2EE options.""" |
| 412 | if not self.client: |
| 413 | return None |
| 414 | kwargs: dict[str, Any] = {"room_id": room_id, "message_type": "m.room.message", "content": content} |
| 415 | |
| 416 | if self.config.e2ee_enabled: |
| 417 | kwargs["ignore_unverified_devices"] = True |
| 418 | response = await self.client.room_send(**kwargs) |
| 419 | return response |
| 420 | |
| 421 | async def _resolve_server_upload_limit_bytes(self) -> int | None: |
| 422 | """Query homeserver upload limit once per channel lifecycle.""" |
| 423 | if self._server_upload_limit_checked: |
| 424 | return self._server_upload_limit_bytes |
| 425 | self._server_upload_limit_checked = True |
| 426 | if not self.client: |
| 427 | return None |
| 428 | try: |
| 429 | response = await self.client.content_repository_config() |
| 430 | except Exception: |
| 431 | return None |
| 432 | upload_size = getattr(response, "upload_size", None) |
| 433 | if isinstance(upload_size, int) and upload_size > 0: |
| 434 | self._server_upload_limit_bytes = upload_size |
| 435 | return upload_size |
| 436 | return None |
| 437 | |
| 438 | async def _effective_media_limit_bytes(self) -> int: |
| 439 | """min(local config, server advertised) — 0 blocks all uploads.""" |
| 440 | local_limit = max(int(self.config.max_media_bytes), 0) |
| 441 | server_limit = await self._resolve_server_upload_limit_bytes() |
| 442 | if server_limit is None: |
| 443 | return local_limit |
| 444 | return min(local_limit, server_limit) if local_limit else 0 |
| 445 | |
| 446 | async def _upload_and_send_attachment( |
| 447 | self, room_id: str, path: Path, limit_bytes: int, |
| 448 | relates_to: dict[str, Any] | None = None, |
| 449 | ) -> str | None: |
| 450 | """Upload one local file to Matrix and send it as a media message. Returns failure marker or None.""" |
| 451 | if not self.client: |
| 452 | return _ATTACH_UPLOAD_FAILED.format(path.name or _DEFAULT_ATTACH_NAME) |
| 453 | |
| 454 | resolved = path.expanduser().resolve(strict=False) |
| 455 | filename = safe_filename(resolved.name) or _DEFAULT_ATTACH_NAME |
| 456 | fail = _ATTACH_UPLOAD_FAILED.format(filename) |
| 457 | |
| 458 | if not resolved.is_file() or not self._is_workspace_path_allowed(resolved): |
| 459 | return fail |
| 460 | try: |
| 461 | size_bytes = resolved.stat().st_size |
| 462 | except OSError: |
| 463 | return fail |
| 464 | if limit_bytes <= 0 or size_bytes > limit_bytes: |
| 465 | return _ATTACH_TOO_LARGE.format(filename) |
| 466 | |
| 467 | mime = mimetypes.guess_type(filename, strict=False)[0] or "application/octet-stream" |
| 468 | try: |
| 469 | with resolved.open("rb") as f: |
| 470 | upload_result = await self.client.upload( |
| 471 | f, content_type=mime, filename=filename, |
| 472 | encrypt=self.config.e2ee_enabled and self._is_encrypted_room(room_id), |
| 473 | filesize=size_bytes, |
| 474 | ) |
| 475 | except Exception: |
| 476 | return fail |
| 477 | |
| 478 | upload_response = upload_result[0] if isinstance(upload_result, tuple) else upload_result |
| 479 | encryption_info = upload_result[1] if isinstance(upload_result, tuple) and isinstance(upload_result[1], dict) else None |
| 480 | if isinstance(upload_response, UploadError): |
| 481 | return fail |
| 482 | mxc_url = getattr(upload_response, "content_uri", None) |
| 483 | if not isinstance(mxc_url, str) or not mxc_url.startswith("mxc://"): |
| 484 | return fail |
| 485 | |
| 486 | content = self._build_outbound_attachment_content( |
| 487 | filename=filename, mime=mime, size_bytes=size_bytes, |
| 488 | mxc_url=mxc_url, encryption_info=encryption_info, |
| 489 | ) |
| 490 | if relates_to: |
| 491 | content["m.relates_to"] = relates_to |
| 492 | try: |
| 493 | await self._send_room_content(room_id, content) |
| 494 | except Exception: |
| 495 | return fail |
| 496 | return None |
| 497 | |
| 498 | async def send(self, msg: OutboundMessage) -> None: |
| 499 | """Send outbound content; clear typing for non-progress messages.""" |
| 500 | if not self.client: |
| 501 | return |
| 502 | text = msg.content or "" |
| 503 | candidates = self._collect_outbound_media_candidates(msg.media) |
| 504 | relates_to = self._build_thread_relates_to(msg.metadata) |
| 505 | is_progress = bool((msg.metadata or {}).get("_progress")) |
| 506 | try: |
| 507 | failures: list[str] = [] |
| 508 | if candidates: |
| 509 | limit_bytes = await self._effective_media_limit_bytes() |
| 510 | for path in candidates: |
| 511 | if fail := await self._upload_and_send_attachment( |
| 512 | room_id=msg.chat_id, |
| 513 | path=path, |
| 514 | limit_bytes=limit_bytes, |
| 515 | relates_to=relates_to, |
| 516 | ): |
| 517 | failures.append(fail) |
| 518 | if failures: |
| 519 | text = f"{text.rstrip()}\n{chr(10).join(failures)}" if text.strip() else "\n".join(failures) |
| 520 | if text or not candidates: |
| 521 | content = _build_matrix_text_content(text) |
| 522 | if relates_to: |
| 523 | content["m.relates_to"] = relates_to |
| 524 | await self._send_room_content(msg.chat_id, content) |
| 525 | finally: |
| 526 | if not is_progress: |
| 527 | await self._stop_typing_keepalive(msg.chat_id, clear_typing=True) |
| 528 | |
| 529 | async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: |
| 530 | meta = metadata or {} |
| 531 | relates_to = self._build_thread_relates_to(metadata) |
| 532 | |
| 533 | if meta.get("_stream_end"): |
| 534 | buf = self._stream_bufs.pop(chat_id, None) |
| 535 | if not buf or not buf.event_id or not buf.text: |
| 536 | return |
| 537 | |
| 538 | await self._stop_typing_keepalive(chat_id, clear_typing=True) |
| 539 | |
| 540 | content = _build_matrix_text_content( |
| 541 | buf.text, |
| 542 | buf.event_id, |
| 543 | thread_relates_to=relates_to, |
| 544 | ) |
| 545 | await self._send_room_content(chat_id, content) |
| 546 | return |
| 547 | |
| 548 | buf = self._stream_bufs.get(chat_id) |
| 549 | if buf is None: |
| 550 | buf = _StreamBuf() |
| 551 | self._stream_bufs[chat_id] = buf |
| 552 | buf.text += delta |
| 553 | |
| 554 | if not buf.text.strip(): |
| 555 | return |
| 556 | |
| 557 | now = self.monotonic_time() |
| 558 | |
| 559 | if not buf.last_edit or (now - buf.last_edit) >= self._STREAM_EDIT_INTERVAL: |
| 560 | try: |
| 561 | content = _build_matrix_text_content( |
| 562 | buf.text, |
| 563 | buf.event_id, |
| 564 | thread_relates_to=relates_to, |
| 565 | ) |
| 566 | response = await self._send_room_content(chat_id, content) |
| 567 | buf.last_edit = now |
| 568 | if not buf.event_id: |
| 569 | # we are editing the same message all the time, so only the first time the event id needs to be set |
| 570 | buf.event_id = response.event_id |
| 571 | except Exception: |
| 572 | await self._stop_typing_keepalive(chat_id, clear_typing=True) |
| 573 | pass |
| 574 | |
| 575 | |
| 576 | def _register_event_callbacks(self) -> None: |
| 577 | self.client.add_event_callback(self._on_message, RoomMessageText) |
| 578 | self.client.add_event_callback(self._on_media_message, MATRIX_MEDIA_EVENT_FILTER) |
| 579 | self.client.add_event_callback(self._on_room_invite, InviteEvent) |
| 580 | |
| 581 | def _register_response_callbacks(self) -> None: |
| 582 | self.client.add_response_callback(self._on_sync_error, SyncError) |
| 583 | self.client.add_response_callback(self._on_join_error, JoinError) |
| 584 | self.client.add_response_callback(self._on_send_error, RoomSendError) |
| 585 | |
| 586 | def _log_response_error(self, label: str, response: Any) -> None: |
| 587 | """Log Matrix response errors — auth errors at ERROR level, rest at WARNING.""" |
| 588 | code = getattr(response, "status_code", None) |
| 589 | is_auth = code in {"M_UNKNOWN_TOKEN", "M_FORBIDDEN", "M_UNAUTHORIZED"} |
| 590 | is_fatal = is_auth or getattr(response, "soft_logout", False) |
| 591 | (logger.error if is_fatal else logger.warning)("Matrix {} failed: {}", label, response) |
| 592 | |
| 593 | async def _on_sync_error(self, response: SyncError) -> None: |
| 594 | self._log_response_error("sync", response) |
| 595 | |
| 596 | async def _on_join_error(self, response: JoinError) -> None: |
| 597 | self._log_response_error("join", response) |
| 598 | |
| 599 | async def _on_send_error(self, response: RoomSendError) -> None: |
| 600 | self._log_response_error("send", response) |
| 601 | |
| 602 | async def _set_typing(self, room_id: str, typing: bool) -> None: |
| 603 | """Best-effort typing indicator update.""" |
| 604 | if not self.client: |
| 605 | return |
| 606 | try: |
| 607 | response = await self.client.room_typing(room_id=room_id, typing_state=typing, |
| 608 | timeout=TYPING_NOTICE_TIMEOUT_MS) |
| 609 | if isinstance(response, RoomTypingError): |
| 610 | logger.debug("Matrix typing failed for {}: {}", room_id, response) |
| 611 | except Exception: |
| 612 | pass |
| 613 | |
| 614 | async def _start_typing_keepalive(self, room_id: str) -> None: |
| 615 | """Start periodic typing refresh (spec-recommended keepalive).""" |
| 616 | await self._stop_typing_keepalive(room_id, clear_typing=False) |
| 617 | await self._set_typing(room_id, True) |
| 618 | if not self._running: |
| 619 | return |
| 620 | |
| 621 | async def loop() -> None: |
| 622 | try: |
| 623 | while self._running: |
| 624 | await asyncio.sleep(TYPING_KEEPALIVE_INTERVAL_MS / 1000) |
| 625 | await self._set_typing(room_id, True) |
| 626 | except asyncio.CancelledError: |
| 627 | pass |
| 628 | |
| 629 | self._typing_tasks[room_id] = asyncio.create_task(loop()) |
| 630 | |
| 631 | async def _stop_typing_keepalive(self, room_id: str, *, clear_typing: bool) -> None: |
| 632 | if task := self._typing_tasks.pop(room_id, None): |
| 633 | task.cancel() |
| 634 | try: |
| 635 | await task |
| 636 | except asyncio.CancelledError: |
| 637 | pass |
| 638 | if clear_typing: |
| 639 | await self._set_typing(room_id, False) |
| 640 | |
| 641 | async def _sync_loop(self) -> None: |
| 642 | while self._running: |
| 643 | try: |
| 644 | await self.client.sync_forever(timeout=30000, full_state=True) |
| 645 | except asyncio.CancelledError: |
| 646 | break |
| 647 | except Exception: |
| 648 | await asyncio.sleep(2) |
| 649 | |
| 650 | async def _on_room_invite(self, room: MatrixRoom, event: InviteEvent) -> None: |
| 651 | if self.is_allowed(event.sender): |
| 652 | await self.client.join(room.room_id) |
| 653 | |
| 654 | def _is_direct_room(self, room: MatrixRoom) -> bool: |
| 655 | count = getattr(room, "member_count", None) |
| 656 | return isinstance(count, int) and count <= 2 |
| 657 | |
| 658 | def _is_bot_mentioned(self, event: RoomMessage) -> bool: |
| 659 | """Check m.mentions payload for bot mention.""" |
| 660 | source = getattr(event, "source", None) |
| 661 | if not isinstance(source, dict): |
| 662 | return False |
| 663 | mentions = (source.get("content") or {}).get("m.mentions") |
| 664 | if not isinstance(mentions, dict): |
| 665 | return False |
| 666 | user_ids = mentions.get("user_ids") |
| 667 | if isinstance(user_ids, list) and self.config.user_id in user_ids: |
| 668 | return True |
| 669 | return bool(self.config.allow_room_mentions and mentions.get("room") is True) |
| 670 | |
| 671 | def _should_process_message(self, room: MatrixRoom, event: RoomMessage) -> bool: |
| 672 | """Apply sender and room policy checks.""" |
| 673 | if not self.is_allowed(event.sender): |
| 674 | return False |
| 675 | if self._is_direct_room(room): |
| 676 | return True |
| 677 | policy = self.config.group_policy |
| 678 | if policy == "open": |
| 679 | return True |
| 680 | if policy == "allowlist": |
| 681 | return room.room_id in (self.config.group_allow_from or []) |
| 682 | if policy == "mention": |
| 683 | return self._is_bot_mentioned(event) |
| 684 | return False |
| 685 | |
| 686 | def _media_dir(self) -> Path: |
| 687 | return get_media_dir("matrix") |
| 688 | |
| 689 | @staticmethod |
| 690 | def _event_source_content(event: RoomMessage) -> dict[str, Any]: |
| 691 | source = getattr(event, "source", None) |
| 692 | if not isinstance(source, dict): |
| 693 | return {} |
| 694 | content = source.get("content") |
| 695 | return content if isinstance(content, dict) else {} |
| 696 | |
| 697 | def _event_thread_root_id(self, event: RoomMessage) -> str | None: |
| 698 | relates_to = self._event_source_content(event).get("m.relates_to") |
| 699 | if not isinstance(relates_to, dict) or relates_to.get("rel_type") != "m.thread": |
| 700 | return None |
| 701 | root_id = relates_to.get("event_id") |
| 702 | return root_id if isinstance(root_id, str) and root_id else None |
| 703 | |
| 704 | def _thread_metadata(self, event: RoomMessage) -> dict[str, str] | None: |
| 705 | if not (root_id := self._event_thread_root_id(event)): |
| 706 | return None |
| 707 | meta: dict[str, str] = {"thread_root_event_id": root_id} |
| 708 | if isinstance(reply_to := getattr(event, "event_id", None), str) and reply_to: |
| 709 | meta["thread_reply_to_event_id"] = reply_to |
| 710 | return meta |
| 711 | |
| 712 | @staticmethod |
| 713 | def _build_thread_relates_to(metadata: dict[str, Any] | None) -> dict[str, Any] | None: |
| 714 | if not metadata: |
| 715 | return None |
| 716 | root_id = metadata.get("thread_root_event_id") |
| 717 | if not isinstance(root_id, str) or not root_id: |
| 718 | return None |
| 719 | reply_to = metadata.get("thread_reply_to_event_id") or metadata.get("event_id") |
| 720 | if not isinstance(reply_to, str) or not reply_to: |
| 721 | return None |
| 722 | return {"rel_type": "m.thread", "event_id": root_id, |
| 723 | "m.in_reply_to": {"event_id": reply_to}, "is_falling_back": True} |
| 724 | |
| 725 | def _event_attachment_type(self, event: MatrixMediaEvent) -> str: |
| 726 | msgtype = self._event_source_content(event).get("msgtype") |
| 727 | return _MSGTYPE_MAP.get(msgtype, "file") |
| 728 | |
| 729 | @staticmethod |
| 730 | def _is_encrypted_media_event(event: MatrixMediaEvent) -> bool: |
| 731 | return (isinstance(getattr(event, "key", None), dict) |
| 732 | and isinstance(getattr(event, "hashes", None), dict) |
| 733 | and isinstance(getattr(event, "iv", None), str)) |
| 734 | |
| 735 | def _event_declared_size_bytes(self, event: MatrixMediaEvent) -> int | None: |
| 736 | info = self._event_source_content(event).get("info") |
| 737 | size = info.get("size") if isinstance(info, dict) else None |
| 738 | return size if isinstance(size, int) and size >= 0 else None |
| 739 | |
| 740 | def _event_mime(self, event: MatrixMediaEvent) -> str | None: |
| 741 | info = self._event_source_content(event).get("info") |
| 742 | if isinstance(info, dict) and isinstance(m := info.get("mimetype"), str) and m: |
| 743 | return m |
| 744 | m = getattr(event, "mimetype", None) |
| 745 | return m if isinstance(m, str) and m else None |
| 746 | |
| 747 | def _event_filename(self, event: MatrixMediaEvent, attachment_type: str) -> str: |
| 748 | body = getattr(event, "body", None) |
| 749 | if isinstance(body, str) and body.strip(): |
| 750 | if candidate := safe_filename(Path(body).name): |
| 751 | return candidate |
| 752 | return _DEFAULT_ATTACH_NAME if attachment_type == "file" else attachment_type |
| 753 | |
| 754 | def _build_attachment_path(self, event: MatrixMediaEvent, attachment_type: str, |
| 755 | filename: str, mime: str | None) -> Path: |
| 756 | safe_name = safe_filename(Path(filename).name) or _DEFAULT_ATTACH_NAME |
| 757 | suffix = Path(safe_name).suffix |
| 758 | if not suffix and mime: |
| 759 | if guessed := mimetypes.guess_extension(mime, strict=False): |
| 760 | safe_name, suffix = f"{safe_name}{guessed}", guessed |
| 761 | stem = (Path(safe_name).stem or attachment_type)[:72] |
| 762 | suffix = suffix[:16] |
| 763 | event_id = safe_filename(str(getattr(event, "event_id", "") or "evt").lstrip("$")) |
| 764 | event_prefix = (event_id[:24] or "evt").strip("_") |
| 765 | return self._media_dir() / f"{event_prefix}_{stem}{suffix}" |
| 766 | |
| 767 | async def _download_media_bytes(self, mxc_url: str) -> bytes | None: |
| 768 | if not self.client: |
| 769 | return None |
| 770 | response = await self.client.download(mxc=mxc_url) |
| 771 | if isinstance(response, DownloadError): |
| 772 | logger.warning("Matrix download failed for {}: {}", mxc_url, response) |
| 773 | return None |
| 774 | body = getattr(response, "body", None) |
| 775 | if isinstance(body, (bytes, bytearray)): |
| 776 | return bytes(body) |
| 777 | if isinstance(response, MemoryDownloadResponse): |
| 778 | return bytes(response.body) |
| 779 | if isinstance(body, (str, Path)): |
| 780 | path = Path(body) |
| 781 | if path.is_file(): |
| 782 | try: |
| 783 | return path.read_bytes() |
| 784 | except OSError: |
| 785 | return None |
| 786 | return None |
| 787 | |
| 788 | def _decrypt_media_bytes(self, event: MatrixMediaEvent, ciphertext: bytes) -> bytes | None: |
| 789 | key_obj, hashes, iv = getattr(event, "key", None), getattr(event, "hashes", None), getattr(event, "iv", None) |
| 790 | key = key_obj.get("k") if isinstance(key_obj, dict) else None |
| 791 | sha256 = hashes.get("sha256") if isinstance(hashes, dict) else None |
| 792 | if not all(isinstance(v, str) for v in (key, sha256, iv)): |
| 793 | return None |
| 794 | try: |
| 795 | return decrypt_attachment(ciphertext, key, sha256, iv) |
| 796 | except (EncryptionError, ValueError, TypeError): |
| 797 | logger.warning("Matrix decrypt failed for event {}", getattr(event, "event_id", "")) |
| 798 | return None |
| 799 | |
| 800 | async def _fetch_media_attachment( |
| 801 | self, room: MatrixRoom, event: MatrixMediaEvent, |
| 802 | ) -> tuple[dict[str, Any] | None, str]: |
| 803 | """Download, decrypt if needed, and persist a Matrix attachment.""" |
| 804 | atype = self._event_attachment_type(event) |
| 805 | mime = self._event_mime(event) |
| 806 | filename = self._event_filename(event, atype) |
| 807 | mxc_url = getattr(event, "url", None) |
| 808 | fail = _ATTACH_FAILED.format(filename) |
| 809 | |
| 810 | if not isinstance(mxc_url, str) or not mxc_url.startswith("mxc://"): |
| 811 | return None, fail |
| 812 | |
| 813 | limit_bytes = await self._effective_media_limit_bytes() |
| 814 | declared = self._event_declared_size_bytes(event) |
| 815 | if declared is not None and declared > limit_bytes: |
| 816 | return None, _ATTACH_TOO_LARGE.format(filename) |
| 817 | |
| 818 | downloaded = await self._download_media_bytes(mxc_url) |
| 819 | if downloaded is None: |
| 820 | return None, fail |
| 821 | |
| 822 | encrypted = self._is_encrypted_media_event(event) |
| 823 | data = downloaded |
| 824 | if encrypted: |
| 825 | if (data := self._decrypt_media_bytes(event, downloaded)) is None: |
| 826 | return None, fail |
| 827 | |
| 828 | if len(data) > limit_bytes: |
| 829 | return None, _ATTACH_TOO_LARGE.format(filename) |
| 830 | |
| 831 | path = self._build_attachment_path(event, atype, filename, mime) |
| 832 | try: |
| 833 | path.write_bytes(data) |
| 834 | except OSError: |
| 835 | return None, fail |
| 836 | |
| 837 | attachment = { |
| 838 | "type": atype, "mime": mime, "filename": filename, |
| 839 | "event_id": str(getattr(event, "event_id", "") or ""), |
| 840 | "encrypted": encrypted, "size_bytes": len(data), |
| 841 | "path": str(path), "mxc_url": mxc_url, |
| 842 | } |
| 843 | return attachment, _ATTACH_MARKER.format(path) |
| 844 | |
| 845 | def _base_metadata(self, room: MatrixRoom, event: RoomMessage) -> dict[str, Any]: |
| 846 | """Build common metadata for text and media handlers.""" |
| 847 | meta: dict[str, Any] = {"room": getattr(room, "display_name", room.room_id)} |
| 848 | if isinstance(eid := getattr(event, "event_id", None), str) and eid: |
| 849 | meta["event_id"] = eid |
| 850 | if thread := self._thread_metadata(event): |
| 851 | meta.update(thread) |
| 852 | return meta |
| 853 | |
| 854 | async def _on_message(self, room: MatrixRoom, event: RoomMessageText) -> None: |
| 855 | if event.sender == self.config.user_id or not self._should_process_message(room, event): |
| 856 | return |
| 857 | await self._start_typing_keepalive(room.room_id) |
| 858 | try: |
| 859 | await self._handle_message( |
| 860 | sender_id=event.sender, chat_id=room.room_id, |
| 861 | content=event.body, metadata=self._base_metadata(room, event), |
| 862 | ) |
| 863 | except Exception: |
| 864 | await self._stop_typing_keepalive(room.room_id, clear_typing=True) |
| 865 | raise |
| 866 | |
| 867 | async def _on_media_message(self, room: MatrixRoom, event: MatrixMediaEvent) -> None: |
| 868 | if event.sender == self.config.user_id or not self._should_process_message(room, event): |
| 869 | return |
| 870 | attachment, marker = await self._fetch_media_attachment(room, event) |
| 871 | parts: list[str] = [] |
| 872 | if isinstance(body := getattr(event, "body", None), str) and body.strip(): |
| 873 | parts.append(body.strip()) |
| 874 | |
| 875 | if attachment and attachment.get("type") == "audio": |
| 876 | transcription = await self.transcribe_audio(attachment["path"]) |
| 877 | if transcription: |
| 878 | parts.append(f"[transcription: {transcription}]") |
| 879 | else: |
| 880 | parts.append(marker) |
| 881 | elif marker: |
| 882 | parts.append(marker) |
| 883 | |
| 884 | await self._start_typing_keepalive(room.room_id) |
| 885 | try: |
| 886 | meta = self._base_metadata(room, event) |
| 887 | meta["attachments"] = [] |
| 888 | if attachment: |
| 889 | meta["attachments"] = [attachment] |
| 890 | await self._handle_message( |
| 891 | sender_id=event.sender, chat_id=room.room_id, |
| 892 | content="\n".join(parts), |
| 893 | media=[attachment["path"]] if attachment else [], |
| 894 | metadata=meta, |
| 895 | ) |
| 896 | except Exception: |
| 897 | await self._stop_typing_keepalive(room.room_id, clear_typing=True) |
| 898 | raise |
| 899 |