diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bacffd6ed6..d7ca8e2a03 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -17,6 +17,8 @@ jobs: run: | python -m pip install --upgrade pip pip install -e ".[messaging,dev,bedrock]" + - name: Install libolm (Matrix E2EE) + run: sudo apt-get update && sudo apt-get install -y libolm-dev - name: Test run: pytest tests -q diff --git a/.gitignore b/.gitignore index 4cc93b88b7..35b5fedf6b 100644 --- a/.gitignore +++ b/.gitignore @@ -9,4 +9,5 @@ dist/ # Local secrets (live-smoke BYO keys) — never committed .env +openspec/ .claude/settings.local.json diff --git a/coworker/connectors/adapters.py b/coworker/connectors/adapters.py index 3d64d9528e..04d8a38c94 100644 --- a/coworker/connectors/adapters.py +++ b/coworker/connectors/adapters.py @@ -459,6 +459,13 @@ def make_adapter( ) if profile.get("bot_token") and profile.get("app_token"): return SlackAdapter(profile["bot_token"], profile["app_token"]) + if platform == "matrix" and profile.get("access_token"): + from ..secrets import state_dir + from .matrix_adapter import MatrixAdapter + from .matrix_settings import MatrixSettings + + settings = MatrixSettings.from_profile(profile) + return MatrixAdapter(settings, store_path=state_dir() / "matrix" / "store") if platform == "github" and profile.get("mode") == "relay": if not (relay_url and token_provider): logger.warning( diff --git a/coworker/connectors/base.py b/coworker/connectors/base.py index e269c09e95..2c0f60ddc2 100644 --- a/coworker/connectors/base.py +++ b/coworker/connectors/base.py @@ -8,11 +8,17 @@ from __future__ import annotations +import base64 +import re from abc import ABC, abstractmethod from dataclasses import asdict, dataclass, field from enum import Enum from typing import Any, Awaitable, Callable, Optional +_MATRIX_TARGET_RE = re.compile( + r"^matrix/([^/]+)(?:/thread/([^/]+))?$" +) + class MessageType(str, Enum): TEXT = "text" @@ -21,17 +27,57 @@ class MessageType(str, Enum): # -- target tokens ------------------------------------------------------------- +def encode_matrix_target( + room_id: str, thread_id: Optional[str] = None +) -> str: + """URL-safe base64 room (and optional thread) for Matrix reply handles.""" + enc = base64.urlsafe_b64encode(room_id.encode()).decode().rstrip("=") + if not thread_id: + return f"matrix/{enc}" + tenc = base64.urlsafe_b64encode(thread_id.encode()).decode().rstrip("=") + return f"matrix/{enc}/thread/{tenc}" + + +def decode_matrix_target(target: str) -> tuple[str, Optional[str]]: + """`matrix/[/thread/]` -> (room_id, thread_id).""" + m = _MATRIX_TARGET_RE.match((target or "").strip()) + if not m: + raise ValueError( + f"invalid matrix target {target!r} " + "(expected 'matrix/[/thread/]')" + ) + room_b64, thread_b64 = m.group(1), m.group(2) + + def _dec(part: str) -> str: + pad = "=" * (-len(part) % 4) + try: + return base64.urlsafe_b64decode(part + pad).decode() + except Exception as exc: + raise ValueError(f"invalid matrix target encoding in {target!r}") from exc + + room_id = _dec(room_b64) + thread_id = _dec(thread_b64) if thread_b64 else None + return room_id, thread_id + + def format_target(platform: str, chat_id: str, thread_id: Optional[str] = None) -> str: + if platform == "matrix": + return encode_matrix_target(chat_id, thread_id) base = f"{platform}:{chat_id}" return f"{base}:{thread_id}" if thread_id else base def parse_target(target: str) -> tuple[str, str, Optional[str]]: - """`'platform:chat_id[:thread]'` -> (platform, chat_id, thread_id).""" - parts = (target or "").split(":") + """`'platform:chat_id[:thread]'` or `matrix/[/thread/]` -> triple.""" + raw = (target or "").strip() + if raw.startswith("matrix/"): + room_id, thread_id = decode_matrix_target(raw) + return "matrix", room_id, thread_id + parts = raw.split(":") if len(parts) < 2 or not parts[0] or not parts[1]: raise ValueError( - f"invalid target {target!r} (expected 'platform:chat_id[:thread]')" + f"invalid target {target!r} (expected 'platform:chat_id[:thread]' " + "or 'matrix/[/thread/]')" ) thread = ":".join(parts[2:]) if len(parts) > 2 else None return parts[0], parts[1], (thread or None) @@ -95,6 +141,8 @@ class MessageEvent: # The bot itself was @-mentioned (UX-DECISIONS §31 mention router). Computed from the RAW # platform text at mapping time — mention tokens are rewritten for display afterwards. mentions_me: bool = False + # When set, delivered to the agent instead of `tagged_text()` — e.g. multimodal parts. + agent_content: Any = None def tagged_text(self) -> str: """How the message enters the super-agent thread: source + reply handle + text. @@ -119,10 +167,11 @@ class SendResult: @dataclass class InteractionEvent: - """A button click on an interactive prompt. + """A button click or emoji reaction on an interactive prompt. Stable actor/workspace ids are security inputs; display names are presentation only. `response_url` is Slack's short-lived reply capability for a private rejection notice. + Matrix reactions set `interaction_kind="reaction"` and `reaction_key` to the emoji. """ platform: str @@ -133,6 +182,8 @@ class InteractionEvent: user_name: Optional[str] = None team_id: Optional[str] = None response_url: Optional[str] = None + interaction_kind: str = "button" # "button" | "reaction" + reaction_key: Optional[str] = None InteractionHandler = Callable[[InteractionEvent], Awaitable[None]] diff --git a/coworker/connectors/catalog_copy.py b/coworker/connectors/catalog_copy.py index 25e09b30de..110dadcea4 100644 --- a/coworker/connectors/catalog_copy.py +++ b/coworker/connectors/catalog_copy.py @@ -19,6 +19,9 @@ "slack": "Bring your coworker into Slack: mention it in a channel or DM it, " "and replies land in-thread. Any number of workspaces can be connected, " "each with its own allow-list of who may talk to the agent.", + "matrix": "Chat with your coworker over Matrix (Element) on your own " + "homeserver. End-to-end encrypted rooms are supported; approve Inbox " + "requests with emoji reactions on mirrored prompts.", "email": "Read, search, and send mail on any IMAP account — Gmail, iCloud, " "Fastmail, or your own server — using an app password instead of your " "account password.", @@ -70,6 +73,13 @@ "Reads files shared in those channels.", "Reads member and channel names to resolve who's talking.", ], + "matrix": [ + "Reads messages in rooms the bot has joined (E2EE when enabled).", + "Sends encrypted messages and uploads files as the bot user.", + "Downloads media you share in those rooms (size-capped).", + "Inbox approvals resolve via emoji reactions on mirrored prompts.", + "Only senders on your allow-list are answered.", + ], "email": [ "Reads and searches mail over IMAP.", "Sends mail as your address, and saves attachments locally.", diff --git a/coworker/connectors/config.py b/coworker/connectors/config.py index 6553d15cd4..5094ce49a8 100644 --- a/coworker/connectors/config.py +++ b/coworker/connectors/config.py @@ -9,12 +9,12 @@ import os from dataclasses import dataclass, field -from typing import Optional +from typing import Any, Optional from ..secrets import SecretStore from .base import SessionSource -PLATFORMS = ("telegram", "slack", "github") +PLATFORMS = ("telegram", "slack", "matrix", "github") @dataclass @@ -62,6 +62,18 @@ def _csv(value: Optional[str]) -> set[str]: return {p.strip() for p in (value or "").split(",") if p.strip()} +def _profile_list(value: Any) -> list[str]: + if value is None: + return [] + if isinstance(value, list): + return [str(v).strip() for v in value if str(v).strip()] + return [p.strip() for p in str(value).split(",") if p.strip()] + + +def _profile_set(value: Any) -> set[str]: + return set(_profile_list(value)) + + def load_settings( secrets: Optional[SecretStore] = None, ) -> dict[str, ConnectorSettings]: @@ -76,6 +88,8 @@ def load_settings( for platform in PLATFORMS: profile = secrets.get(f"{platform}:default") or {} token = profile.get("bot_token") + if platform == "matrix": + token = profile.get("access_token") allowed = set(profile.get("allowed_users") or []) allowed |= _csv(os.environ.get(f"{platform.upper()}_ALLOWED_USERS")) allow_all = bool(profile.get("allow_all")) or os.environ.get( diff --git a/coworker/connectors/descriptors.py b/coworker/connectors/descriptors.py index 32e103c6f5..b9a361dc7a 100644 --- a/coworker/connectors/descriptors.py +++ b/coworker/connectors/descriptors.py @@ -118,6 +118,31 @@ def _validate_email(creds: dict) -> ValidationResult: return ValidationResult(ok, identity=identity or None, error=error or None) +def _validate_matrix(creds: dict) -> ValidationResult: + import httpx + + base = str(creds.get("homeserver_url") or "").rstrip("/") + token = creds.get("access_token", "") + if not base or not token: + return ValidationResult(False, error="homeserver_url and access_token required") + try: + resp = httpx.get( + f"{base}/_matrix/client/v3/account/whoami", + headers={"Authorization": f"Bearer {token}"}, + timeout=15, + ) + data = resp.json() + except Exception as exc: + return ValidationResult(False, error=str(exc)) + if resp.status_code >= 400: + detail = ( + (data.get("errcode") or data.get("error")) if isinstance(data, dict) else None + ) + return ValidationResult(False, error=str(detail or f"HTTP {resp.status_code}")) + user_id = data.get("user_id") if isinstance(data, dict) else None + return ValidationResult(True, identity=str(user_id or "matrix user")) + + def _validate_slack(creds: dict) -> ValidationResult: import httpx @@ -488,6 +513,61 @@ def _validate_outlook(creds: dict) -> ValidationResult: ], validate=_validate_slack, ), + ConnectorDescriptor( + name="matrix", + title="Matrix", + icon="⬡", + blurb="Two-way encrypted messaging on your Synapse or Element Server Suite homeserver.", + auth="token", + two_way=True, + channels=True, + brand_color="#0dbd8b", + logo="matrix", + fields=[ + Field( + "homeserver_url", + "Homeserver URL", + help="Your Synapse or ESS base URL, e.g. https://matrix.example.org", + placeholder="https://matrix.example.org", + ), + Field( + "access_token", + "Access token", + secret=True, + help="Bot user access token (from Element → Settings → Help & About).", + ), + Field( + "recovery_key", + "Recovery key", + secret=True, + help="Required for E2EE cross-signing bootstrap on encrypted rooms.", + ), + Field( + "user_id", + "User ID", + required=False, + help="Optional @user:server — filled automatically on connect.", + placeholder="@bot:example.org", + ), + _ALLOWED_FIELD, + Field( + "allowed_rooms", + "Allowed room IDs", + required=False, + help="Comma-separated room IDs. Empty = all joined rooms.", + placeholder="!abc:example.org", + ), + ], + instructions=[ + "Create a dedicated bot user on your Synapse or ESS homeserver.", + "Sign in with Element, open Settings → Help & About → Access Token.", + "Export your security recovery key (Settings → Security & Privacy) — required for E2EE.", + "Install libolm on this machine: brew install libolm (macOS) or apt install libolm-dev (Linux). CI installs libolm-dev on Ubuntu.", + "Invite the bot to encrypted rooms; it auto-accepts invites.", + "After connecting, message the bot once, then use Capture to grab your Matrix user ID.", + ], + validate=_validate_matrix, + ), ConnectorDescriptor( name="email", title="Email (IMAP)", diff --git a/coworker/connectors/matrix_adapter.py b/coworker/connectors/matrix_adapter.py new file mode 100644 index 0000000000..7fd38b20d0 --- /dev/null +++ b/coworker/connectors/matrix_adapter.py @@ -0,0 +1,627 @@ +"""Matrix inbound adapter — matrix-nio AsyncClient with required E2EE.""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from typing import Any, Optional + +from .base import ( + BasePlatformAdapter, + InteractionEvent, + MessageEvent, + MessageType, + SendResult, + SessionSource, +) +from .matrix_reactions import ( + PendingReaction, + PendingReactionStore, + reactions_for_buttons, +) +from .matrix_settings import MatrixSettings + +logger = logging.getLogger("coworker.connectors.matrix") + +# Sync bridge for stateless send_message tool (runs in worker threads). +_matrix_adapter: Optional["MatrixAdapter"] = None +_matrix_loop: Optional[asyncio.AbstractEventLoop] = None + + +def register_matrix_adapter( + adapter: Optional["MatrixAdapter"], + loop: Optional[asyncio.AbstractEventLoop] = None, +) -> None: + global _matrix_adapter, _matrix_loop + _matrix_adapter = adapter + _matrix_loop = loop + + +def send_matrix_sync(chat_id: str, text: str, thread_id: Optional[str] = None) -> SendResult: + """Blocking outbound send via the live adapter (E2EE-aware).""" + adapter = _matrix_adapter + loop = _matrix_loop + if adapter is None or loop is None: + return SendResult(False, error="matrix adapter not connected") + future = asyncio.run_coroutine_threadsafe( + adapter.send(chat_id, text, thread_id=thread_id), loop + ) + try: + return future.result(timeout=60) + except Exception as exc: + return SendResult(False, error=str(exc)) + + +def send_matrix_file_sync( + chat_id: str, + thread_id: Optional[str], + filename: str, + data: bytes, + title: Optional[str] = None, + comment: Optional[str] = None, +) -> SendResult: + adapter = _matrix_adapter + loop = _matrix_loop + if adapter is None or loop is None: + return SendResult(False, error="matrix adapter not connected") + future = asyncio.run_coroutine_threadsafe( + adapter.send_file_bytes( + chat_id, data, filename, thread_id=thread_id, title=title, comment=comment + ), + loop, + ) + try: + return future.result(timeout=120) + except Exception as exc: + return SendResult(False, error=str(exc)) + + +def _room_chat_type(room: Any, *, dm_rooms: set[str]) -> str: + if room.room_id in dm_rooms: + return "dm" + if _matrix_room_is_dm(room): + return "dm" + return "channel" + + +def _matrix_room_is_dm(room: Any) -> bool: + """True for 1:1 direct chats (summary, member count, or unnamed 2-person room).""" + try: + if room.member_count == 2: + return True + except (AttributeError, TypeError, ValueError): + pass + try: + if getattr(room, "is_group", False) and room.joined_count == 2: + return True + except (AttributeError, TypeError, ValueError): + pass + return False + + +def _mentions_bot(text: str, bot_user_id: Optional[str]) -> bool: + if not bot_user_id or not text: + return False + return bot_user_id in text or f"@{bot_user_id.split(':')[0][1:]}" in text + + +def matrix_event_to_event( + event: Any, + *, + room_id: str, + bot_user_id: Optional[str], + chat_type: str = "channel", + chat_name: Optional[str] = None, + thread_id: Optional[str] = None, +) -> Optional[MessageEvent]: + """Pure mapper: Matrix m.room.message (text) -> MessageEvent.""" + sender = getattr(event, "sender", None) or ( + event.get("sender") if isinstance(event, dict) else None + ) + if bot_user_id and sender == bot_user_id: + return None + body = getattr(event, "body", None) + if body is None and isinstance(event, dict): + body = (event.get("content") or {}).get("body") + if not body: + return None + event_id = getattr(event, "event_id", None) or ( + event.get("event_id") if isinstance(event, dict) else None + ) + source = SessionSource( + platform="matrix", + chat_id=room_id, + user_id=sender, + chat_type=chat_type, + chat_name=chat_name, + thread_id=thread_id, + ) + return MessageEvent( + text=str(body), + source=source, + message_id=event_id, + mentions_me=_mentions_bot(str(body), bot_user_id), + ) + + +class MatrixAdapter(BasePlatformAdapter): + platform = "matrix" + + def __init__( + self, + settings: MatrixSettings, + *, + store_path: Path, + reaction_store: Optional[PendingReactionStore] = None, + ) -> None: + super().__init__() + self.settings = settings + self.store_path = store_path + self.reaction_store = reaction_store or PendingReactionStore() + self._client = None + self._sync_task: Optional[asyncio.Task] = None + self._closing = False + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._joined_threads: set[tuple[str, str]] = set() + self._dm_rooms: set[str] = set() + self._dm_rooms_path: Optional[Path] = None + self._lifecycle_event: dict[str, str] = {} # room_id -> inbound event_id + + def _note_thread(self, room_id: str, thread_id: Optional[str]) -> None: + if thread_id: + self._joined_threads.add((room_id, thread_id)) + + def _thread_active(self, room_id: str, thread_id: Optional[str]) -> bool: + if not thread_id: + return False + return (room_id, thread_id) in self._joined_threads + + def _load_dm_rooms(self) -> None: + path = self._dm_rooms_path + if path is None or not path.is_file(): + return + try: + import json + + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, list): + self._dm_rooms.update(str(r) for r in data) + except Exception: + logger.debug("matrix dm_rooms load failed", exc_info=True) + + def _remember_dm_room(self, room_id: str) -> None: + if room_id in self._dm_rooms: + return + self._dm_rooms.add(room_id) + path = self._dm_rooms_path + if path is None: + return + try: + import json + + path.write_text( + json.dumps(sorted(self._dm_rooms), indent=0) + "\n", + encoding="utf-8", + ) + except Exception: + logger.debug("matrix dm_rooms save failed", exc_info=True) + + def _is_dm(self, room: Any) -> bool: + room_id = room.room_id + if room_id in self._dm_rooms: + return True + if _matrix_room_is_dm(room): + self._remember_dm_room(room_id) + return True + return False + + def _should_dispatch(self, mapped: MessageEvent, room_id: str, *, is_dm: bool) -> bool: + if is_dm: + return True + if room_id in self.settings.free_response_rooms: + return True + if mapped.mentions_me: + return True + if self._thread_active(room_id, mapped.source.thread_id): + mapped.mentions_me = True + return True + if not self.settings.require_mention: + return True + return False + + async def connect(self) -> bool: + if self.settings.e2ee_mode == "required": + try: + import nio.crypto # noqa: F401 + except ImportError: + logger.warning( + "matrix E2EE requires matrix-nio[e2e] and libolm — " + "`pip install coworker[messaging]` and install libolm " + "(brew install libolm / apt install libolm-dev)" + ) + return False + + try: + from nio import AsyncClient, AsyncClientConfig, InviteMemberEvent, RoomMessageText + from nio.events import RoomMessage + from nio.events.room_events import ( + ReactionEvent, + RoomMessageAudio, + RoomMessageFile, + RoomMessageImage, + RoomMessageVideo, + ) + except ImportError: + logger.warning( + "matrix-nio not installed — `pip install coworker[messaging]`" + ) + return False + + if not self.settings.homeserver_url or not self.settings.access_token: + logger.warning("matrix: missing homeserver_url or access_token") + return False + + self.store_path.mkdir(parents=True, exist_ok=True) + self._dm_rooms_path = self.store_path / "dm_rooms.json" + self._load_dm_rooms() + user_id = self.settings.user_id or "" + # ponytail: without store_sync_tokens, restart replays full room timelines. + client_config = AsyncClientConfig(store_sync_tokens=True) + self._client = AsyncClient( + self.settings.homeserver_url, + user_id or "@bot:local", + store_path=str(self.store_path), + config=client_config, + ) + self._client.access_token = self.settings.access_token + if user_id: + self._client.user_id = user_id + + try: + resp = await self._client.whoami() + if hasattr(resp, "user_id") and resp.user_id: + self._client.user_id = resp.user_id + elif isinstance(resp, dict): + self._client.user_id = resp.get("user_id") or self._client.user_id + if hasattr(resp, "device_id") and resp.device_id: + self._client.device_id = resp.device_id + elif isinstance(resp, dict) and resp.get("device_id"): + self._client.device_id = resp["device_id"] + except Exception: + logger.exception("matrix whoami failed") + await self._client.close() + self._client = None + return False + + if hasattr(self._client, "load_store"): + try: + self._client.load_store() + except Exception: + logger.exception("matrix crypto store load failed") + await self._client.close() + self._client = None + return False + + if self.settings.e2ee_mode == "required": + try: + from .matrix_crypto_bootstrap import ( + MatrixCryptoBootstrapError, + prepare_matrix_e2ee, + ) + + await prepare_matrix_e2ee(self._client, self.settings) + except MatrixCryptoBootstrapError as exc: + logger.warning("matrix E2EE bootstrap failed: %s", exc) + await self._client.close() + self._client = None + return False + except Exception: + logger.exception("matrix E2EE bootstrap failed") + await self._client.close() + self._client = None + return False + + self._loop = asyncio.get_running_loop() + register_matrix_adapter(self, self._loop) + self._closing = False + + async def _dispatch(room, event, *, chat_type: str, thread_id: Optional[str]): + if not self._allowed_room(room.room_id, event.sender, chat_type=chat_type): + return + if self.settings.ignored_user(event.sender): + return + mapped = matrix_event_to_event( + event, + room_id=room.room_id, + bot_user_id=self._client.user_id, + chat_type=chat_type, + chat_name=getattr(room, "display_name", None), + thread_id=thread_id, + ) + if mapped is None: + return + is_dm = chat_type == "dm" + if not self._should_dispatch(mapped, room.room_id, is_dm=is_dm): + return + media = await self._media_agent_content(room.room_id, event) + if media is not None: + mapped.agent_content = media + mapped.message_type = MessageType.MEDIA + if mapped.message_id and self.settings.lifecycle_reactions: + await self._lifecycle_react(room.room_id, mapped.message_id, "👀") + self._lifecycle_event[room.room_id] = mapped.message_id + try: + await self.handle_message(mapped) + except Exception: + if self.settings.lifecycle_reactions and mapped.message_id: + await self._lifecycle_react(room.room_id, mapped.message_id, "❌") + raise + + async def _on_room_message(room, event): + if self._is_dm(room): + chat_type = "dm" + else: + chat_type = _room_chat_type(room, dm_rooms=self._dm_rooms) + thread_id = None + relates = getattr(getattr(event, "content", None), "relates_to", None) + if relates is not None: + thread_id = getattr(relates, "event_id", None) + await _dispatch(room, event, chat_type=chat_type, thread_id=thread_id) + + async def _on_reaction(room, event): + if not isinstance(event, ReactionEvent): + return + await self._handle_reaction(room.room_id, event) + + async def _on_invite(room, event): + if isinstance(event, InviteMemberEvent): + try: + await self._client.join(room.room_id) + except Exception: + logger.debug("matrix auto-join failed for %s", room.room_id, exc_info=True) + + self._client.add_event_callback(_on_room_message, RoomMessageText) + for cls in (RoomMessageImage, RoomMessageFile, RoomMessageAudio, RoomMessageVideo): + self._client.add_event_callback(_on_room_message, cls) + self._client.add_event_callback(_on_reaction, ReactionEvent) + self._client.add_event_callback(_on_invite, InviteMemberEvent) + + self._sync_task = asyncio.create_task(self._sync_loop()) + logger.info("matrix adapter connected as %s", self._client.user_id) + return True + + async def _sync_loop(self) -> None: + # ponytail: with a saved sync token, one full_state sync refreshes room + # summaries/members without replaying timeline (since=token still applies). + first = True + while not self._closing and self._client is not None: + try: + full_state = first and bool( + getattr(self._client, "loaded_sync_token", None) + or getattr(self._client, "next_batch", None) + ) + first = False + await self._client.sync(timeout=30_000, full_state=full_state) + except asyncio.CancelledError: + break + except Exception: + logger.exception("matrix sync error — retrying") + await asyncio.sleep(2) + + def _allowed_room( + self, room_id: str, sender: Optional[str], *, chat_type: str = "channel" + ) -> bool: + allowed_rooms = self.settings.allowed_rooms + if not allowed_rooms: + return True + if chat_type == "dm" or room_id in self._dm_rooms: + return True + return room_id in allowed_rooms + + async def _lifecycle_react(self, room_id: str, event_id: str, emoji: str) -> None: + if self._client is None: + return + content = { + "m.relates_to": { + "rel_type": "m.annotation", + "event_id": event_id, + "key": emoji, + } + } + try: + await self._client.room_send( + room_id, "m.reaction", content, ignore_unverified_devices=True + ) + except Exception: + logger.debug("matrix lifecycle reaction %s failed", emoji, exc_info=True) + + async def _media_agent_content(self, room_id: str, event: Any) -> Any | None: + """Download mxc media and build multimodal agent content, or None for text-only.""" + content = getattr(event, "content", None) or {} + if isinstance(content, dict): + url = content.get("url") or content.get("file", {}).get("url") + body = content.get("body") or content.get("filename") or "attachment" + else: + url = getattr(content, "url", None) + body = getattr(content, "body", None) or "attachment" + if not url or not str(url).startswith("mxc://"): + return None + if self._client is None: + return None + try: + resp = await self._client.download(url) + data = getattr(resp, "body", None) or getattr(resp, "content", None) + if data is None and hasattr(resp, "read"): + data = resp.read() + if not data: + return None + if len(data) > self.settings.max_media_bytes: + return None + except Exception: + logger.debug("matrix media download failed", exc_info=True) + return None + import base64 + from mimetypes import guess_type + + mime, _ = guess_type(str(body)) + mime = mime or "application/octet-stream" + inbound_dir = self.store_path / "inbound" + inbound_dir.mkdir(parents=True, exist_ok=True) + safe_name = str(body).replace("/", "_")[:120] or "attachment" + path = inbound_dir / safe_name + path.write_bytes(data) + tagged = matrix_event_to_event( + event, + room_id=room_id, + bot_user_id=self._client.user_id if self._client else None, + ) + frame = tagged.tagged_text() if tagged else f"[matrix media: {safe_name}]" + if mime.startswith("image/"): + b64 = base64.standard_b64encode(data).decode() + return [ + {"type": "text", "text": frame}, + {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}, + ] + return f"{frame}\n[Attachment saved: {path}]" + + async def _handle_reaction(self, room_id: str, event: Any) -> None: + relates = getattr(getattr(event, "content", None), "relates_to", None) + if relates is None: + return + prompt_id = getattr(relates, "event_id", None) + emoji = getattr(relates, "key", None) + if not prompt_id or not emoji: + return + resolved = self.reaction_store.resolve_emoji(room_id, prompt_id, emoji) + if resolved is None: + return + value, pending = resolved + if ( + self.settings.approval_require_sender + and pending.allowed_reactor + and event.sender != pending.allowed_reactor + ): + return + self.reaction_store.pop(room_id, prompt_id) + await self.handle_interaction( + InteractionEvent( + platform="matrix", + chat_id=room_id, + message_id=prompt_id, + value=value, + user_id=getattr(event, "sender", None), + interaction_kind="reaction", + reaction_key=emoji, + ) + ) + + async def disconnect(self) -> None: + self._closing = True + register_matrix_adapter(None, None) + if self._sync_task is not None: + self._sync_task.cancel() + self._sync_task = None + if self._client is not None: + try: + await self._client.close() + except Exception: + pass + self._client = None + + def _thread_content(self, thread_id: Optional[str]) -> dict: + if not thread_id: + return {} + return {"m.relates_to": {"rel_type": "m.thread", "event_id": thread_id}} + + async def send( + self, chat_id: str, text: str, *, thread_id: Optional[str] = None + ) -> SendResult: + if self._client is None: + return SendResult(False, error="matrix not connected") + content = {"msgtype": "m.text", "body": text[: self.settings.max_message_length]} + content.update(self._thread_content(thread_id)) + try: + resp = await self._client.room_send( + chat_id, "m.room.message", content, ignore_unverified_devices=True + ) + event_id = getattr(resp, "event_id", None) + self._note_thread(chat_id, thread_id) + if self.settings.auto_thread and not thread_id and event_id: + self._note_thread(chat_id, event_id) + pending = self._lifecycle_event.pop(chat_id, None) + if pending and self.settings.lifecycle_reactions: + await self._lifecycle_react(chat_id, pending, "✅") + return SendResult(True, message_id=event_id) + except Exception as exc: + return SendResult(False, error=str(exc)) + + async def send_interactive( + self, chat_id: str, text: str, buttons, *, thread_id: Optional[str] = None + ) -> SendResult: + hint = text + emoji_map = reactions_for_buttons(buttons) + if emoji_map: + keys = " ".join(emoji_map.keys()) + hint = f"{text}\n\nReact: {keys}" + result = await self.send(chat_id, hint, thread_id=thread_id) + if not result.ok or not result.message_id: + return result + self.reaction_store.register( + PendingReaction( + room_id=chat_id, + prompt_event_id=result.message_id, + emoji_map=emoji_map, + ) + ) + return result + + async def send_file_bytes( + self, + chat_id: str, + data: bytes, + filename: str, + *, + thread_id: Optional[str] = None, + title: Optional[str] = None, + comment: Optional[str] = None, + ) -> SendResult: + if self._client is None: + return SendResult(False, error="matrix not connected") + if len(data) > self.settings.max_media_bytes: + return SendResult(False, error="file exceeds max_media_bytes") + try: + from mimetypes import guess_type + + mime, _ = guess_type(filename) + mime = mime or "application/octet-stream" + upload = await self._client.upload(data, mime, filename=filename) + if hasattr(upload, "content_uri"): + mxc = upload.content_uri + else: + mxc = getattr(upload, "content_uri", None) or str(upload) + msgtype = "m.image" if mime.startswith("image/") else "m.file" + content: dict[str, Any] = { + "msgtype": msgtype, + "body": title or filename, + "url": mxc, + "filename": filename, + } + if comment: + content["body"] = f"{comment}\n{content['body']}" + content.update(self._thread_content(thread_id)) + resp = await self._client.room_send( + chat_id, "m.room.message", content, ignore_unverified_devices=True + ) + return SendResult(True, message_id=getattr(resp, "event_id", None)) + except Exception as exc: + return SendResult(False, error=str(exc)) + + async def update_message(self, chat_id: str, message_id: str, text: str) -> None: + if self._client is None or not message_id: + return + try: + await self._client.room_redact(chat_id, message_id, reason=text[:50]) + await self.send(chat_id, text) + except Exception: + logger.debug("matrix update_message failed", exc_info=True) diff --git a/coworker/connectors/matrix_crypto_bootstrap.py b/coworker/connectors/matrix_crypto_bootstrap.py new file mode 100644 index 0000000000..96463bf4f1 --- /dev/null +++ b/coworker/connectors/matrix_crypto_bootstrap.py @@ -0,0 +1,387 @@ +"""Matrix E2EE bootstrap — recovery key (SSSS) + stale device-key detection. + +matrix-nio has no cross-signing / secret-storage support; this module implements the +subset needed for Element-style bot accounts on self-hosted Synapse: +- Parse Element recovery keys (base58) +- Decrypt cross-signing secrets from account data (m.secret_storage.v1.aes-hmac-sha2) +- Sign the current device with the self-signing key +- Detect local/server identity mismatch (stale OTM / deleted crypto store) +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import logging +from typing import Any, Optional + +import httpx +from Crypto.Cipher import AES +from Crypto.Hash import SHA256 +from Crypto.PublicKey import ECC +from Crypto.Protocol.KDF import HKDF +from Crypto.Signature import eddsa + +logger = logging.getLogger("coworker.connectors.matrix") + +_B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" +_SSSS_ZERO_SALT = b"\x00" * 32 +_CROSS_SIGNING_SELF = "m.cross_signing.self_signing" + + +class MatrixCryptoBootstrapError(Exception): + """E2EE bootstrap failed — connect must fail closed.""" + + +def _unpadded_b64(data: bytes) -> str: + return base64.b64encode(data).decode("ascii").rstrip("=") + + +def _b64_decode(data: str) -> bytes: + return base64.b64decode(data + "=" * (-len(data) % 4)) + + +def parse_recovery_key(raw: str) -> bytes: + """Element recovery / security key → 32-byte secret storage key. + + Accepts Element base58 recovery keys or a raw 32-byte hex string (64 hex chars). + """ + cleaned = "".join(raw.split()) + if not cleaned: + raise MatrixCryptoBootstrapError("recovery key is empty") + hex_candidate = cleaned.removeprefix("0x") + if len(hex_candidate) == 64 and all(c in "0123456789abcdefABCDEF" for c in hex_candidate): + try: + key = bytes.fromhex(hex_candidate) + except ValueError as exc: + raise MatrixCryptoBootstrapError("invalid recovery key hex") from exc + if len(key) != 32: + raise MatrixCryptoBootstrapError("recovery key hex must decode to 32 bytes") + return key + num = 0 + for ch in cleaned: + try: + num = num * 58 + _B58.index(ch) + except ValueError as exc: + raise MatrixCryptoBootstrapError("invalid recovery key encoding") from exc + # Preserve leading zero bytes. + full = num.to_bytes((num.bit_length() + 7) // 8 or 1, "big") + pad = len(cleaned) - len(cleaned.lstrip(_B58[0])) + decoded = b"\x00" * pad + full + if len(decoded) != 35: + raise MatrixCryptoBootstrapError( + f"recovery key decoded to {len(decoded)} bytes (expected 35)" + ) + if decoded[:2] != b"\x8b\x01": + raise MatrixCryptoBootstrapError("recovery key has invalid prefix") + key = decoded[2:34] + parity = decoded[34] + if (parity ^ _xor_bytes(decoded[:34])) & 0xFF: + raise MatrixCryptoBootstrapError("recovery key parity check failed") + return key + + +def _xor_bytes(data: bytes) -> int: + x = 0 + for b in data: + x ^= b + return x + + +def _hkdf_keys(storage_key: bytes, info: bytes) -> tuple[bytes, bytes]: + out = HKDF( + storage_key, + 64, + _SSSS_ZERO_SALT, + SHA256, + context=info, + ) + return out[:32], out[32:] + + +def _verify_storage_key(storage_key: bytes, key_desc: dict) -> None: + """Optional iv/mac self-check on m.secret_storage.key.* account data.""" + iv_b64 = key_desc.get("iv") + mac_b64 = key_desc.get("mac") + if not iv_b64 or not mac_b64: + return + aes_key, mac_key = _hkdf_keys(storage_key, b"") + iv = _b64_decode(iv_b64) + if len(iv) != 16: + raise MatrixCryptoBootstrapError("invalid secret storage key iv") + iv = bytearray(iv) + iv[8] &= 0x7F + cipher = AES.new(aes_key, AES.MODE_CTR, nonce=b"", initial_value=iv) + ct = cipher.encrypt(b"\x00" * 32) + expected = hmac.new(mac_key, ct, hashlib.sha256).digest() + if not hmac.compare_digest(_b64_decode(mac_b64), expected): + raise MatrixCryptoBootstrapError("recovery key does not match this account") + + +def _decrypt_secret( + storage_key: bytes, secret_name: str, blob: dict +) -> bytes: + aes_key, mac_key = _hkdf_keys(storage_key, secret_name.encode("utf-8")) + iv = _b64_decode(str(blob["iv"])) + if len(iv) != 16: + raise MatrixCryptoBootstrapError(f"invalid iv for secret {secret_name}") + iv = bytearray(iv) + iv[8] &= 0x7F + ct = _b64_decode(str(blob["ciphertext"])) + expected_mac = hmac.new(mac_key, ct, hashlib.sha256).digest() + if not hmac.compare_digest(_b64_decode(str(blob["mac"])), expected_mac): + raise MatrixCryptoBootstrapError(f"MAC mismatch for secret {secret_name}") + cipher = AES.new(aes_key, AES.MODE_CTR, nonce=b"", initial_value=bytes(iv)) + return cipher.decrypt(ct) + + +def _sign_json_ed25519(seed: bytes, payload: dict) -> str: + from nio.api import Api + + unsigned = {k: v for k, v in payload.items() if k not in ("signatures", "unsigned")} + message = Api.to_canonical_json(unsigned).encode("utf-8") + key = ECC.construct(curve="Ed25519", seed=seed) + return _unpadded_b64(eddsa.new(key, "rfc8032").sign(message)) + + +def _public_key_b64(seed: bytes) -> str: + raw = ECC.construct(curve="Ed25519", seed=seed).public_key().export_key(format="raw") + return _unpadded_b64(raw) + + +def _get_account_data( + base_url: str, token: str, user_id: str, event_type: str +) -> Optional[dict]: + from urllib.parse import quote + + url = ( + f"{base_url.rstrip('/')}/_matrix/client/v3/user/" + f"{quote(user_id, safe='')}/account_data/{quote(event_type, safe='')}" + ) + resp = httpx.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=30) + if resp.status_code == 404: + return None + resp.raise_for_status() + return resp.json() + + +def _upload_signatures( + base_url: str, token: str, body: dict +) -> None: + url = f"{base_url.rstrip('/')}/_matrix/client/v3/keys/signatures/upload" + resp = httpx.post( + url, + headers={"Authorization": f"Bearer {token}"}, + json=body, + timeout=30, + ) + if resp.status_code >= 400: + detail = resp.text[:200] + raise MatrixCryptoBootstrapError( + f"signatures/upload failed ({resp.status_code}): {detail}" + ) + + +def _query_own_device_keys( + base_url: str, token: str, user_id: str, device_id: str +) -> Optional[dict]: + url = f"{base_url.rstrip('/')}/_matrix/client/v3/keys/query" + resp = httpx.post( + url, + headers={"Authorization": f"Bearer {token}"}, + json={"device_keys": {user_id: [device_id]}}, + timeout=30, + ) + if resp.status_code >= 400: + return None + data = resp.json() + dev = (data.get("device_keys") or {}).get(user_id, {}).get(device_id) + return dev if isinstance(dev, dict) else None + + +def detect_stale_device_keys( + *, + base_url: str, + token: str, + user_id: str, + device_id: str, + local_curve25519: str, + local_ed25519: str, +) -> Optional[str]: + """Return an actionable error if the homeserver has different identity keys.""" + remote = _query_own_device_keys(base_url, token, user_id, device_id) + if not remote: + return None + keys = remote.get("keys") or {} + remote_curve = keys.get(f"curve25519:{device_id}") + remote_ed = keys.get(f"ed25519:{device_id}") + if not remote_curve and not remote_ed: + return None + if remote_curve and remote_curve != local_curve25519: + return ( + f"device {device_id} has stale keys on the server (identity key mismatch). " + "Delete the local crypto store or generate a new access token (fresh device id)." + ) + if remote_ed and remote_ed != local_ed25519: + return ( + f"device {device_id} has stale signing keys on the server. " + "Generate a new access token or delete the device via Synapse admin API." + ) + return None + + +def bootstrap_cross_signing( + *, + base_url: str, + token: str, + user_id: str, + device_id: str, + recovery_key: str, + device_keys: dict, +) -> None: + """Import self-signing key from SSSS and cross-sign the current device.""" + storage_key = parse_recovery_key(recovery_key) + + default = _get_account_data(base_url, token, user_id, "m.secret_storage.default_key") + if not default or not default.get("key"): + raise MatrixCryptoBootstrapError( + "no default secret storage key on account — set up cross-signing in Element first" + ) + key_id = str(default["key"]) + key_desc = _get_account_data( + base_url, token, user_id, f"m.secret_storage.key.{key_id}" + ) + if not key_desc: + raise MatrixCryptoBootstrapError(f"secret storage key {key_id!r} not found") + _verify_storage_key(storage_key, key_desc) + + secret_event = _get_account_data( + base_url, token, user_id, "m.cross_signing.self_signing" + ) + if not secret_event or "encrypted" not in secret_event: + raise MatrixCryptoBootstrapError( + "m.cross_signing.self_signing not in account data — enable secure backup in Element" + ) + enc = (secret_event.get("encrypted") or {}).get(key_id) + if not enc: + raise MatrixCryptoBootstrapError( + "self_signing secret not encrypted with the default storage key" + ) + plain = _decrypt_secret(storage_key, _CROSS_SIGNING_SELF, enc) + try: + parsed = json.loads(plain.decode("utf-8")) + except Exception as exc: + raise MatrixCryptoBootstrapError("self_signing secret is not valid JSON") from exc + seed_b64 = parsed.get("private_key") or parsed.get("key") + if not seed_b64: + raise MatrixCryptoBootstrapError("self_signing secret missing private_key") + seed = _b64_decode(str(seed_b64)) + if len(seed) != 32: + raise MatrixCryptoBootstrapError("self_signing private key must be 32 bytes") + + unsigned = { + k: v for k, v in device_keys.items() if k not in ("signatures", "unsigned") + } + sig = _sign_json_ed25519(seed, unsigned) + pub = _public_key_b64(seed) + signed = dict(device_keys) + signatures = dict(signed.get("signatures") or {}) + user_sigs = dict(signatures.get(user_id) or {}) + user_sigs[f"ed25519:{pub}"] = sig + signatures[user_id] = user_sigs + signed["signatures"] = signatures + + _upload_signatures( + base_url, + token, + {user_id: {device_id: signed}}, + ) + logger.info("matrix cross-signing: signed device %s", device_id) + + +async def prepare_matrix_e2ee(client: Any, settings: Any) -> None: + """Run stale-key check, keys upload, and optional cross-signing bootstrap.""" + if settings.e2ee_mode != "required": + return + if client.olm is None: + raise MatrixCryptoBootstrapError("encryption store not loaded") + + user_id = client.user_id + device_id = client.device_id or getattr(client.olm, "device_id", None) + if not user_id or not device_id: + raise MatrixCryptoBootstrapError("whoami did not return user_id/device_id") + + local_curve = client.olm.account.identity_keys["curve25519"] + local_ed = client.olm.account.identity_keys["ed25519"] + stale = detect_stale_device_keys( + base_url=settings.homeserver_url, + token=settings.access_token, + user_id=user_id, + device_id=device_id, + local_curve25519=local_curve, + local_ed25519=local_ed, + ) + if stale: + raise MatrixCryptoBootstrapError(stale) + + from nio.responses import KeysUploadError + + if client.should_upload_keys: + resp = await client.keys_upload() + if isinstance(resp, KeysUploadError): + msg = getattr(resp, "message", None) or str(resp) + if "identity" in msg.lower() or "one.time" in msg.lower(): + raise MatrixCryptoBootstrapError( + f"keys/upload failed (stale device keys?): {msg}. " + "Generate a new access token or delete the device on Synapse." + ) + raise MatrixCryptoBootstrapError(f"keys/upload failed: {msg}") + + import asyncio + + default = await asyncio.to_thread( + _get_account_data, + settings.homeserver_url, + settings.access_token, + user_id, + "m.secret_storage.default_key", + ) + if not default or not default.get("key"): + logger.info( + "matrix: account has no secret storage — skipping cross-signing bootstrap" + ) + return + + if not settings.recovery_key: + raise MatrixCryptoBootstrapError( + "recovery_key is required — this account has cross-signing enabled. " + "Export it from Element (Settings → Security & Privacy → Recovery key)." + ) + + device_keys = _local_device_keys(client.olm, user_id, device_id) + await asyncio.to_thread( + bootstrap_cross_signing, + base_url=settings.homeserver_url, + token=settings.access_token, + user_id=user_id, + device_id=device_id, + recovery_key=settings.recovery_key, + device_keys=device_keys, + ) + + +def _local_device_keys(olm: Any, user_id: str, device_id: str) -> dict: + base = { + "algorithms": olm._algorithms, + "device_id": device_id, + "user_id": user_id, + "keys": { + f"curve25519:{device_id}": olm.account.identity_keys["curve25519"], + f"ed25519:{device_id}": olm.account.identity_keys["ed25519"], + }, + } + sig = olm.sign_json(base) + base["signatures"] = {user_id: {f"ed25519:{device_id}": sig}} + return base diff --git a/coworker/connectors/matrix_profile.py b/coworker/connectors/matrix_profile.py new file mode 100644 index 0000000000..a962ae9a3c --- /dev/null +++ b/coworker/connectors/matrix_profile.py @@ -0,0 +1,75 @@ +"""PATCH helpers for Matrix connector advanced settings.""" + +from __future__ import annotations + +from typing import Any + +from ..secrets import SecretStore +from .matrix_settings import MatrixSettings + +_MATRIX_SETTING_KEYS = frozenset( + { + "require_mention", + "auto_thread", + "session_scope", + "dm_mention_threads", + "dm_auto_thread", + "group_sessions_per_user", + "lifecycle_reactions", + "allowed_rooms", + "free_response_rooms", + } +) + + +def matrix_settings_public(profile: dict) -> dict[str, Any]: + s = MatrixSettings.from_profile(profile) + return { + "homeserver_url": s.homeserver_url, + "user_id": s.user_id, + "require_mention": s.require_mention, + "auto_thread": s.auto_thread, + "session_scope": s.session_scope, + "dm_mention_threads": s.dm_mention_threads, + "dm_auto_thread": s.dm_auto_thread, + "group_sessions_per_user": s.group_sessions_per_user, + "lifecycle_reactions": s.lifecycle_reactions, + "allowed_rooms": sorted(s.allowed_rooms), + "free_response_rooms": sorted(s.free_response_rooms), + } + + +def patch_matrix_settings(secrets: SecretStore, body: dict[str, Any]) -> dict[str, Any]: + profile = secrets.get("matrix:default") + if not profile: + return {"ok": False, "error": "matrix not connected"} + if not isinstance(body, dict): + return {"ok": False, "error": "body must be an object"} + updated = dict(profile) + for key, value in body.items(): + if key not in _MATRIX_SETTING_KEYS: + continue + if key in ("allowed_rooms", "free_response_rooms"): + if isinstance(value, str): + value = [p.strip() for p in value.split(",") if p.strip()] + if not isinstance(value, list): + return {"ok": False, "error": f"{key} must be a list or CSV string"} + updated[key] = value + elif key == "session_scope": + scope = str(value or "auto").strip().lower() + if scope not in ("auto", "room", "thread"): + return {"ok": False, "error": "session_scope must be auto, room, or thread"} + updated[key] = scope + elif key in ( + "require_mention", + "auto_thread", + "dm_mention_threads", + "dm_auto_thread", + "group_sessions_per_user", + "lifecycle_reactions", + ): + updated[key] = bool(value) + else: + updated[key] = value + secrets.put("matrix:default", updated) + return {"ok": True, "settings": matrix_settings_public(updated)} diff --git a/coworker/connectors/matrix_reactions.py b/coworker/connectors/matrix_reactions.py new file mode 100644 index 0000000000..6e63f6385a --- /dev/null +++ b/coworker/connectors/matrix_reactions.py @@ -0,0 +1,95 @@ +"""Matrix emoji reactions for Inbox prompts — pending registry + emoji maps.""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass +from typing import Optional + +from ..interactions import Button, encode +from ..inbox import KIND_APPROVAL, KIND_QUESTION + +APPROVAL_EMOJI: dict[str, str] = { + "✅": "allow", + "♾️": "always", + "❌": "deny", +} + +NUMBER_EMOJI = ("1️⃣", "2️⃣", "3️⃣", "4️⃣", "5️⃣", "6️⃣", "7️⃣", "8️⃣", "9️⃣", "🔟") + + +@dataclass +class PendingReaction: + room_id: str + prompt_event_id: str + emoji_map: dict[str, str] # emoji -> encoded value + allowed_reactor: Optional[str] = None + + +class PendingReactionStore: + def __init__(self) -> None: + self._lock = threading.Lock() + self._pending: dict[tuple[str, str], PendingReaction] = {} + + def register(self, pending: PendingReaction) -> None: + key = (pending.room_id, pending.prompt_event_id) + with self._lock: + self._pending[key] = pending + + def lookup( + self, room_id: str, prompt_event_id: str + ) -> Optional[PendingReaction]: + with self._lock: + return self._pending.get((room_id, prompt_event_id)) + + def pop(self, room_id: str, prompt_event_id: str) -> Optional[PendingReaction]: + key = (room_id, prompt_event_id) + with self._lock: + return self._pending.pop(key, None) + + def resolve_emoji( + self, room_id: str, relates_to_event_id: str, emoji: str + ) -> Optional[tuple[str, PendingReaction]]: + pending = self.lookup(room_id, relates_to_event_id) + if pending is None: + return None + value = pending.emoji_map.get(emoji) + if value is None: + return None + return value, pending + + +def reactions_for(item) -> dict[str, str]: + """Build emoji map for an Inbox item mirrored to Matrix.""" + emoji_map: dict[str, str] = {} + if item.kind == KIND_APPROVAL: + for emoji, resolution in APPROVAL_EMOJI.items(): + emoji_map[emoji] = encode(item.id, resolution) + elif item.kind == KIND_QUESTION and getattr(item, "options", None): + for i, opt in enumerate(item.options): + if i >= len(NUMBER_EMOJI): + break + emoji_map[NUMBER_EMOJI[i]] = encode(item.id, opt) + return emoji_map + + +def reactions_for_buttons(buttons: list[Button]) -> dict[str, str]: + """Map emoji keys to button values for interactive Matrix prompts.""" + out: dict[str, str] = {} + for i, btn in enumerate(buttons): + if btn.label == "Approve": + out["✅"] = btn.value + # Hermes-style approve-always (♾️) — same item id, resolution "always". + try: + import json + + d = json.loads(btn.value) + if isinstance(d, dict) and d.get("id"): + out["♾️"] = encode(str(d["id"]), "always") + except Exception: + pass + elif btn.label == "Deny": + out["❌"] = btn.value + elif i < len(NUMBER_EMOJI): + out[NUMBER_EMOJI[i]] = btn.value + return out diff --git a/coworker/connectors/matrix_routing.py b/coworker/connectors/matrix_routing.py new file mode 100644 index 0000000000..e7ca37a7a2 --- /dev/null +++ b/coworker/connectors/matrix_routing.py @@ -0,0 +1,47 @@ +"""Matrix mention/session routing — Hermes-aligned session keys for manager.""" + +from __future__ import annotations + +from typing import Optional + +from .base import SessionSource, format_target +from .matrix_settings import MatrixSettings + +_USER_SCOPE_PREFIX = "@user:" + + +def _user_thread_suffix(user_id: Optional[str]) -> Optional[str]: + if not user_id: + return None + return f"{_USER_SCOPE_PREFIX}{user_id}" + + +def effective_session_scope(settings: MatrixSettings) -> str: + """`auto` behaves like `thread` (Hermes default).""" + scope = (settings.session_scope or "auto").strip().lower() + return "thread" if scope == "auto" else scope + + +def mention_thread_target( + settings: MatrixSettings, + source: SessionSource, + message_id: Optional[str], +) -> str: + """Mention-session key — same string used for standing send_message grants.""" + scope = effective_session_scope(settings) + user_suffix = ( + _user_thread_suffix(source.user_id) if settings.group_sessions_per_user else None + ) + if scope == "room": + thread_id = user_suffix + else: + thread_id = source.thread_id or message_id + if user_suffix and thread_id: + thread_id = f"{thread_id}|{user_suffix}" + elif user_suffix: + thread_id = user_suffix + return format_target("matrix", source.chat_id, thread_id) + + +def dm_mention_routes_to_thread(settings: MatrixSettings, *, mentions_me: bool, is_dm: bool) -> bool: + return bool(is_dm and mentions_me and settings.dm_mention_threads) diff --git a/coworker/connectors/matrix_sessions.py b/coworker/connectors/matrix_sessions.py new file mode 100644 index 0000000000..48e8472943 --- /dev/null +++ b/coworker/connectors/matrix_sessions.py @@ -0,0 +1,73 @@ +"""Optional persisted room-session map for Matrix (task 3.3). + +Primary routing uses ``mention_sessions`` with keys from ``matrix_routing.mention_thread_target``. +This store holds auxiliary room→session mappings when ``session_scope=room``. +""" + +from __future__ import annotations + +import json +import threading +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Optional + + +@dataclass +class MatrixRoomSession: + room_id: str + user_id: str # empty = room-wide session + session_id: str + + +class MatrixSessionStore: + def __init__(self, path: Optional[str | Path] = None) -> None: + self.path = Path(path) if path else None + self._lock = threading.Lock() + self._rows: list[MatrixRoomSession] = [] + self._load() + + def _key(self, room_id: str, user_id: str) -> tuple[str, str]: + return room_id, user_id or "" + + def _load(self) -> None: + if self.path and self.path.is_file(): + try: + data = json.loads(self.path.read_text(encoding="utf-8")) + self._rows = [MatrixRoomSession(**raw) for raw in data.get("sessions", [])] + except (OSError, ValueError, TypeError): + self._rows = [] + + def _save(self) -> None: + if not self.path: + return + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text( + json.dumps({"sessions": [asdict(r) for r in self._rows]}, indent=2), + encoding="utf-8", + ) + + def set(self, room_id: str, session_id: str, *, user_id: str = "") -> None: + uid = user_id or "" + with self._lock: + for row in self._rows: + if self._key(row.room_id, row.user_id) == self._key(room_id, uid): + row.session_id = session_id + self._save() + return + self._rows.append(MatrixRoomSession(room_id=room_id, user_id=uid, session_id=session_id)) + self._save() + + def get(self, room_id: str, *, user_id: str = "") -> Optional[str]: + uid = user_id or "" + for row in self._rows: + if self._key(row.room_id, row.user_id) == self._key(room_id, uid): + return row.session_id + return None + + def remove_session(self, session_id: str) -> None: + with self._lock: + before = len(self._rows) + self._rows = [r for r in self._rows if r.session_id != session_id] + if len(self._rows) != before: + self._save() diff --git a/coworker/connectors/matrix_settings.py b/coworker/connectors/matrix_settings.py new file mode 100644 index 0000000000..bec28ff73d --- /dev/null +++ b/coworker/connectors/matrix_settings.py @@ -0,0 +1,77 @@ +"""Matrix connector settings from the `matrix:default` profile.""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from typing import Optional + +from .config import _csv, _profile_list, _profile_set + + +@dataclass +class MatrixSettings: + homeserver_url: str + access_token: str + user_id: Optional[str] = None + recovery_key: Optional[str] = None + allowed_users: set[str] = field(default_factory=set) + allowed_rooms: set[str] = field(default_factory=set) + free_response_rooms: set[str] = field(default_factory=set) + ignore_user_patterns: list[re.Pattern[str]] = field(default_factory=list) + require_mention: bool = True + auto_thread: bool = True + session_scope: str = "auto" # auto | room | thread + dm_mention_threads: bool = False + dm_auto_thread: bool = True + group_sessions_per_user: bool = True + lifecycle_reactions: bool = True + e2ee_mode: str = "required" + max_message_length: int = 4000 + max_media_bytes: int = 104_857_600 + approval_require_sender: bool = True + + @classmethod + def from_profile(cls, profile: dict) -> "MatrixSettings": + patterns = [] + raw_patterns = _profile_list(profile.get("ignore_user_patterns")) + if not raw_patterns: + raw_patterns = ["^@telegram_", "^@slack_", "^@whatsapp_"] + for raw in raw_patterns: + try: + patterns.append(re.compile(raw)) + except re.error: + continue + allowed_rooms = _profile_set(profile.get("allowed_rooms")) | _csv( + os.environ.get("MATRIX_ALLOWED_ROOMS") + ) + free_response_rooms = _profile_set(profile.get("free_response_rooms")) | _csv( + os.environ.get("MATRIX_FREE_RESPONSE_ROOMS") + ) + return cls( + homeserver_url=str(profile.get("homeserver_url") or "").rstrip("/"), + access_token=str(profile.get("access_token") or ""), + user_id=profile.get("user_id"), + recovery_key=profile.get("recovery_key"), + allowed_users=_profile_set(profile.get("allowed_users")), + allowed_rooms=allowed_rooms, + free_response_rooms=free_response_rooms, + ignore_user_patterns=patterns, + require_mention=bool(profile.get("require_mention", True)), + auto_thread=bool(profile.get("auto_thread", True)), + session_scope=str(profile.get("session_scope") or "auto"), + dm_mention_threads=bool(profile.get("dm_mention_threads", False)), + dm_auto_thread=bool(profile.get("dm_auto_thread", True)), + group_sessions_per_user=bool(profile.get("group_sessions_per_user", True)), + lifecycle_reactions=bool(profile.get("lifecycle_reactions", True)), + e2ee_mode=str(profile.get("e2ee_mode") or "required"), + max_message_length=int(profile.get("max_message_length") or 4000), + max_media_bytes=int(profile.get("max_media_bytes") or 104_857_600), + approval_require_sender=bool(profile.get("approval_require_sender", True)), + ) + + def ignored_user(self, user_id: Optional[str]) -> bool: + if not user_id: + return False + return any(p.search(user_id) for p in self.ignore_user_patterns) diff --git a/coworker/connectors/relay_client.py b/coworker/connectors/relay_client.py index 57fc38cc33..8bd8905d62 100644 --- a/coworker/connectors/relay_client.py +++ b/coworker/connectors/relay_client.py @@ -191,17 +191,19 @@ def state(self) -> str: return "reconnecting" return "offline" - async def wait_dispatched(self, at_least: int, timeout: float = 2.0) -> None: + async def wait_dispatched(self, at_least: int, timeout: float = 5.0) -> None: """Test helper: wait until at least N frames have been dispatched.""" loop = asyncio.get_event_loop() deadline = loop.time() + timeout while self._dispatched < at_least: - self._progress.clear() + if self._dispatched >= at_least: + return remaining = deadline - loop.time() if remaining <= 0: raise TimeoutError( f"only {self._dispatched} frames dispatched (< {at_least})" ) + self._progress.clear() try: await asyncio.wait_for(self._progress.wait(), timeout=remaining) except asyncio.TimeoutError: @@ -284,7 +286,7 @@ def status(self) -> dict[str, Any]: }, } - async def wait_dispatched(self, at_least: int, timeout: float = 2.0) -> None: + async def wait_dispatched(self, at_least: int, timeout: float = 5.0) -> None: await self._hub.wait_dispatched(at_least, timeout) # -- team registry ------------------------------------------------------- @@ -407,7 +409,9 @@ async def _slack_get( return None base = os.environ.get("SLACK_API_URL", "https://slack.com/api/") try: - async with httpx.AsyncClient(timeout=15) as http: + async with httpx.AsyncClient( + timeout=httpx.Timeout(0.5, connect=0.25) + ) as http: resp = await http.get( base + method, params=params, diff --git a/coworker/connectors/senders.py b/coworker/connectors/senders.py index 9ca9ff8e89..2c812116fe 100644 --- a/coworker/connectors/senders.py +++ b/coworker/connectors/senders.py @@ -144,6 +144,17 @@ def _send_slack_interactive( } +def _send_matrix( + token: str, chat_id: str, text: str, thread_id: Optional[str] = None +) -> SendResult: + from .matrix_adapter import send_matrix_sync + + return send_matrix_sync(chat_id, text, thread_id) + + +DEFAULT_SENDERS["matrix"] = _send_matrix + + # -- file upload (§34 / UX-016) -------------------------------------------------------- # A FileSender is (token, chat_id, thread_id, filename, data, title, comment) -> SendResult. FileSender = Callable[ @@ -214,3 +225,22 @@ def _send_slack_file( DEFAULT_FILE_SENDERS: dict[str, FileSender] = { "slack": _send_slack_file, } + + +def _send_matrix_file( + token: str, + chat_id: str, + thread_id: Optional[str], + filename: str, + data: bytes, + title: Optional[str] = None, + comment: Optional[str] = None, +) -> SendResult: + from .matrix_adapter import send_matrix_file_sync + + return send_matrix_file_sync( + chat_id, thread_id, filename, data, title=title, comment=comment + ) + + +DEFAULT_FILE_SENDERS["matrix"] = _send_matrix_file diff --git a/coworker/connectors/setup.py b/coworker/connectors/setup.py index 57d5476d44..77f6f6f03c 100644 --- a/coworker/connectors/setup.py +++ b/coworker/connectors/setup.py @@ -129,6 +129,10 @@ def connector_list(secrets: SecretStore) -> list[dict[str, Any]]: a["email"] == default_email and a["managed"] for a in accounts ) entry["filters"] = gmail_accounts.get_filters(secrets) + if d.name == "matrix" and connected: + from .matrix_profile import matrix_settings_public + + entry["matrix_settings"] = matrix_settings_public(profile) if d.name == "google_calendar": # Multi-account, same shape as gmail: each `google_calendar:account:*` # profile is one Google account; :default is just the default pointer. diff --git a/coworker/connectors/tools.py b/coworker/connectors/tools.py index 8b2f4ce4d0..8c9b274415 100644 --- a/coworker/connectors/tools.py +++ b/coworker/connectors/tools.py @@ -22,18 +22,21 @@ "function": { "name": "send_message", "description": ( - "Send a message to a connected chat (Slack or Telegram). `target` is the " - "reply handle from an inbound message (e.g. 'telegram:12345' or 'slack:C0123', " - "optionally with a ':' suffix) — or, for Slack, just the channel NAME " - "('#general' or 'general'; resolved against the connected workspaces). Use this to " - "actually reach a person — plain assistant text is not delivered anywhere." + "Send a message to a connected chat (Slack, Telegram, or Matrix). `target` is the " + "reply handle from an inbound message (e.g. 'telegram:12345', 'slack:C0123', " + "or 'matrix/[/thread/]') — or, for Slack, just the " + "channel NAME ('#general' or 'general'; resolved against the connected workspaces). " + "Use this to actually reach a person — plain assistant text is not delivered anywhere." ), "parameters": { "type": "object", "properties": { "target": { "type": "string", - "description": "Destination handle 'platform:chat_id[:thread]', e.g. 'telegram:12345'.", + "description": ( + "Destination handle: 'platform:chat_id[:thread]' for Slack/Telegram, " + "or 'matrix/[/thread/]' for Matrix." + ), }, "text": {"type": "string", "description": "The message text to send."}, }, @@ -128,6 +131,8 @@ def _resolve_token(secrets: SecretStore, platform: str, chat_id: str) -> Optiona per_team = secrets.get(f"slack:team:{team}") or {} return per_team.get("bot_token") creds = secrets.get(f"{platform}:default") or {} + if platform == "matrix": + return creds.get("access_token") return creds.get("bot_token") @@ -184,7 +189,7 @@ def send_message(target: str, text: str) -> dict[str, Any]: "function": { "name": "send_file", "description": ( - "Upload a file from the session's workspace into a connected chat (Slack). " + "Upload a file from the session's workspace into a connected chat (Slack or Matrix). " "`target` is the same handle send_message uses. Slack shows its own previews " "for pdf/csv/images — send the actual file, not a screenshot of it. For .html " "artifacts (which Slack can't preview) set as_screenshot=true to send a " @@ -196,7 +201,10 @@ def send_message(target: str, text: str) -> dict[str, Any]: "properties": { "target": { "type": "string", - "description": "Destination handle 'platform:chat_id[:thread]', e.g. 'slack:C0123:171234.5678'.", + "description": ( + "Destination handle (same as send_message): " + "'platform:chat_id[:thread]' or matrix encoding." + ), }, "path": { "type": "string", diff --git a/coworker/server/app.py b/coworker/server/app.py index 457c08110b..b69d72fa49 100644 --- a/coworker/server/app.py +++ b/coworker/server/app.py @@ -1375,6 +1375,15 @@ def gmail_filters(body: dict) -> dict[str, Any]: return {"ok": False, "error": "labels must be a list"} return gmail_accounts.set_filters(manager.secrets, senders, labels) + @app.patch("/v1/connectors/matrix/settings") + async def matrix_settings(body: dict) -> dict[str, Any]: + from ..connectors.matrix_profile import patch_matrix_settings + + result = patch_matrix_settings(manager.secrets, body or {}) + if result.get("ok"): + await _refresh_listeners_if_two_way("matrix") + return result + @app.post("/v1/connectors/google_calendar/accounts/{email}/disconnect") async def gcal_account_disconnect(email: str) -> dict[str, Any]: """Drop ONE Google Calendar account (cloud metadata best-effort first); diff --git a/coworker/server/manager.py b/coworker/server/manager.py index 130ede1f23..0ba02cb251 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -3183,6 +3183,31 @@ def set_dm_session(self, session_id: Optional[str]) -> dict[str, Any]: self._save_prefs() return {"ok": True, "dm_session": self.dm_session()} + def _matrix_settings(self): + from ..connectors.matrix_settings import MatrixSettings + + profile = self.secrets.get("matrix:default") or {} + return MatrixSettings.from_profile(profile) + + @staticmethod + def _platform_label(platform: str) -> str: + return {"slack": "Slack", "matrix": "Matrix", "telegram": "Telegram"}.get( + platform, platform.title() + ) + + def _mention_thread_target(self, event) -> str: + from ..connectors.base import format_target + + src = event.source + thread_key = src.thread_id or getattr(event, "message_id", None) + if src.platform == "matrix": + from ..connectors.matrix_routing import mention_thread_target + + return mention_thread_target( + self._matrix_settings(), src, getattr(event, "message_id", None) + ) + return format_target(src.platform, src.chat_id, thread_key) + def _ollama_alive(self) -> bool: """Best-effort local-Ollama liveness, cached 30s (get_settings runs on every GUI fetch — no 2s probe inline). Keyless is not the same as PRESENT: `ollama:*` picker @@ -4372,7 +4397,12 @@ async def mirror_inbox_item(self, item) -> None: # available in-app, but never mirror it to an ownerless channel. if not self.slack_approval_owner_ids(team_id): return - target = f"{binding.channel}:{binding.target}" + if binding.channel == "matrix": + target = binding.target + if not target.startswith("matrix/"): + target = f"matrix/{binding.target}" + else: + target = f"{binding.channel}:{binding.target}" body = "\n".join(p for p in (item.title, item.body) if p).strip() buttons = buttons_for(item) try: @@ -4637,9 +4667,26 @@ async def _dispatch_inbound(self, event) -> None: return return # channel with no subscribers — nobody is listening # DM (or any non-channel): route to the designated session, else park it for visibility. + if src.platform == "matrix": + from ..connectors.matrix_routing import dm_mention_routes_to_thread + + if dm_mention_routes_to_thread( + self._matrix_settings(), + mentions_me=bool(getattr(event, "mentions_me", False)), + is_dm=True, + ): + await self._route_mention( + event, ms, self.subscriptions.for_channel(channel) + ) + return dm = self.dm_session() + agent_msg = ( + event.agent_content + if getattr(event, "agent_content", None) is not None + else event.tagged_text() + ) if dm and self._inbound_connector_allowed(dm, src.platform): - await self.deliver_to_session(dm, event.tagged_text(), source=ms.to_dict()) + await self.deliver_to_session(dm, agent_msg, source=ms.to_dict()) elif dm: # Designated, but this session has muted the connector → park rather than deliver. self.unrouted.record( @@ -4655,14 +4702,10 @@ async def _route_mention(self, event, ms: MessageSource, subs) -> None: """@OpenWorker tagged in a channel. A subscribed (user-connected) coworker owns the channel and must answer; otherwise the per-thread coworker session handles it — spawned on the first tag, steered by follow-ups (deduped on the thread target).""" - from ..connectors.base import format_target - src = event.source - # Slack semantics: replying to a top-level message threads on THAT message's ts, so a - # top-level tag (no thread_ts) keys — and is answered — on its own ts. - thread_key = src.thread_id or getattr(event, "message_id", None) - thread_target = format_target(src.platform, src.chat_id, thread_key) + thread_target = self._mention_thread_target(event) who = src.user_name or src.user_id or "?" + plat = self._platform_label(src.platform) chan = f"#{src.chat_name}" if src.chat_name else src.chat_id if subs: # The user connected a coworker to this channel — it answers tags; no spawn. @@ -4686,7 +4729,7 @@ async def _route_mention(self, event, ms: MessageSource, subs) -> None: if sid and self.session_store.load(sid) is not None: # Follow-up tag in a thread we already own → steer the same session. msg = ( - f"💬 Follow-up in your Slack thread ({chan}) from {who}: {event.text}\n" + f"💬 Follow-up in your {plat} thread ({chan}) from {who}: {event.text}\n" f'(Reply in the thread with the send_message tool, target "{thread_target}" ' f"— replies there are pre-approved.)" ) @@ -4705,6 +4748,7 @@ async def _spawn_mention_session( src = event.source who = src.user_name or src.user_id or "?" + plat = self._platform_label(src.platform) chan = f"#{src.chat_name}" if src.chat_name else src.chat_id sid = uuid.uuid4().hex engine = self.get_engine(sid, agent=self.personas.default_id()) @@ -4730,16 +4774,25 @@ async def _spawn_mention_session( self.session_store.rename(sid, f"{ask} — {chan}" if ask else chan) label = chan + (f" · {src.team_id}" if src.team_id else "") self.session_store.set_origin(sid, src.platform, label) + if src.platform == "matrix": + from ..connectors.matrix_routing import effective_session_scope + from ..connectors.matrix_sessions import MatrixSessionStore + from ..secrets import state_dir + + if effective_session_scope(self._matrix_settings()) == "room": + store = MatrixSessionStore(state_dir() / "matrix" / "room_sessions.json") + uid = src.user_id or "" if self._matrix_settings().group_sessions_per_user else "" + store.set(src.chat_id, sid, user_id=uid) # Up to 6 lines of channel context, minus the tag itself (it's the opening line). recent = self.channel_buffer.recent(f"{src.platform}:{src.chat_id}", 7)[:-1] context = "\n".join(f"- {m['from']}: {m['text']}" for m in recent) opening = ( - f"🔔 You were mentioned on Slack in {chan} by {who}: {event.text}\n\n" - f"You own this Slack thread. Reply in the thread using the send_message tool " + f"🔔 You were mentioned on {plat} in {chan} by {who}: {event.text}\n\n" + f"You own this {plat} thread. Reply in the thread using the send_message tool " f'with target "{thread_target}" — replies to this thread are pre-approved and ' f"never prompt the user. Anything else (other channels, files, external " f"actions) asks for approval as usual. Keep replies concise and " - f"Slack-appropriate." + f"{plat}-appropriate." + (f"\n\nRecent channel context:\n{context}" if context else "") ) try: diff --git a/coworker/testing/fake_matrix/__init__.py b/coworker/testing/fake_matrix/__init__.py new file mode 100644 index 0000000000..4d176d7f45 --- /dev/null +++ b/coworker/testing/fake_matrix/__init__.py @@ -0,0 +1,3 @@ +from .server import BOT_USER, DEFAULT_TOKEN, FakeMatrix + +__all__ = ["BOT_USER", "DEFAULT_TOKEN", "FakeMatrix"] diff --git a/coworker/testing/fake_matrix/server.py b/coworker/testing/fake_matrix/server.py new file mode 100644 index 0000000000..71e2e2a818 --- /dev/null +++ b/coworker/testing/fake_matrix/server.py @@ -0,0 +1,126 @@ +"""Fake Matrix homeserver — minimal Client-Server API for integration tests.""" + +from __future__ import annotations + +import uuid +from typing import Optional + +import uvicorn +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route + +BOT_USER = "@bot:fake.local" +DEFAULT_TOKEN = "syt_fake_matrix_token" + + +class FakeMatrix: + def __init__(self, host: str = "127.0.0.1", port: int = 0) -> None: + self.host = host + self.port = port + self.user_id = BOT_USER + self.device_id = "FAKE_DEVICE" + self.token = DEFAULT_TOKEN + self.next_batch = "s1" + self.rooms: dict[str, list[dict]] = {} + self.outbound: list[dict] = [] + self.reactions: list[dict] = [] + self._server: Optional[uvicorn.Server] = None + self._task = None + self.app = self._build_app() + + @property + def base_url(self) -> str: + return f"http://{self.host}:{self.port}" + + def _auth(self, request: Request) -> bool: + auth = request.headers.get("authorization", "") + return auth == f"Bearer {self.token}" + + def _build_app(self) -> Starlette: + async def whoami(request: Request) -> JSONResponse: + if not self._auth(request): + return JSONResponse({"errcode": "M_UNKNOWN_TOKEN"}, status_code=401) + return JSONResponse( + {"user_id": self.user_id, "device_id": self.device_id, "is_guest": False} + ) + + async def sync(request: Request) -> JSONResponse: + if not self._auth(request): + return JSONResponse({"errcode": "M_UNKNOWN_TOKEN"}, status_code=401) + return JSONResponse({"next_batch": self.next_batch, "rooms": {}}) + + async def send_event(request: Request) -> JSONResponse: + room_id = request.path_params["room_id"] + if not self._auth(request): + return JSONResponse({"errcode": "M_UNKNOWN_TOKEN"}, status_code=401) + body = await request.json() + event_id = f"${uuid.uuid4().hex[:8]}" + entry = { + "event_id": event_id, + "type": request.path_params["event_type"], + "content": body, + } + self.rooms.setdefault(room_id, []).append(entry) + if request.path_params["event_type"] == "m.reaction": + self.reactions.append({"room_id": room_id, **entry}) + else: + self.outbound.append({"room_id": room_id, **entry}) + return JSONResponse({"event_id": event_id}) + + async def keys_query(request: Request) -> JSONResponse: + return JSONResponse({"device_keys": {}, "failures": {}}) + + async def control_inject(request: Request) -> JSONResponse: + body = await request.json() + room_id = str(body.get("room_id") or "!fake:local") + self.rooms.setdefault(room_id, []).append(body.get("event") or {}) + return JSONResponse({"ok": True}) + + async def control_reset(request: Request) -> JSONResponse: + self.outbound.clear() + self.reactions.clear() + self.rooms.clear() + return JSONResponse({"ok": True}) + + async def control_outbound(request: Request) -> JSONResponse: + return JSONResponse({"messages": self.outbound, "reactions": self.reactions}) + + return Starlette( + routes=[ + Route("/_matrix/client/v3/account/whoami", whoami, methods=["GET"]), + Route("/_matrix/client/v3/sync", sync, methods=["GET", "POST"]), + Route( + "/_matrix/client/v3/rooms/{room_id}/send/{event_type}/{txn_id}", + send_event, + methods=["PUT"], + ), + Route("/_matrix/client/v3/keys/query", keys_query, methods=["POST"]), + Route("/control/inject", control_inject, methods=["POST"]), + Route("/control/reset", control_reset, methods=["POST"]), + Route("/control/outbound", control_outbound, methods=["GET"]), + ] + ) + + async def start(self) -> None: + config = uvicorn.Config( + self.app, host=self.host, port=self.port, log_level="warning" + ) + self._server = uvicorn.Server(config) + import asyncio + + self._task = asyncio.create_task(self._server.serve()) + for _ in range(50): + if self._server.started: + break + await asyncio.sleep(0.05) + sockets = self._server.servers[0].sockets if self._server.servers else [] + if sockets: + self.port = sockets[0].getsockname()[1] + + async def stop(self) -> None: + if self._server is not None: + self._server.should_exit = True + if self._task is not None: + await self._task diff --git a/openspec/changes/add-matrix-connector/.openspec.yaml b/openspec/changes/add-matrix-connector/.openspec.yaml new file mode 100644 index 0000000000..ffa710fcc3 --- /dev/null +++ b/openspec/changes/add-matrix-connector/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-31 diff --git a/openspec/changes/add-matrix-connector/design.md b/openspec/changes/add-matrix-connector/design.md new file mode 100644 index 0000000000..f7c42c434e --- /dev/null +++ b/openspec/changes/add-matrix-connector/design.md @@ -0,0 +1,136 @@ +## Context + +OpenWorker connectors split **inbound** (gateway adapters + allowlist) from **outbound** (stateless `send_message` / `send_file` tools). Slack and Telegram implement `BasePlatformAdapter`; Slack additionally supports Block Kit buttons for Inbox mirroring via `InteractionEvent`. + +Matrix differs materially: + +- **E2EE** requires a long-lived crypto-aware client for both sync and send — outbound cannot be a one-shot httpx POST. +- **Room IDs** contain colons (`!abc:example.org`), incompatible with the existing `platform:chat_id[:thread]` parser. +- **Interactive prompts** use **emoji reactions** on a posted message (Hermes pattern), not clickable buttons. +- **Media** in encrypted rooms must be downloaded/uploaded through the Matrix client with decrypted `mxc://` content. + +The codebase explicitly borrowed gateway patterns from Hermes. This design targets **behavioral parity** with Hermes Matrix docs for mention/thread/session and reaction approvals, scoped to OpenWorker's existing primitives (`mention_sessions`, `subscriptions`, `dm_session`, `InboxStore`). + +**Constraints (locked):** + +| Decision | Choice | +|----------|--------| +| Connector id | `matrix` | +| Homeserver | Self-hosted Synapse / ESS | +| E2EE | Required on day one | +| Cloud relay | No | +| SDK | `matrix-nio[e2e]` (PyPI; requires system `libolm`) | +| Reactions + media | Day one | + +## Goals / Non-Goals + +**Goals:** + +- Connect a bot account to a self-hosted homeserver with access token + recovery key; validate via `/account/whoami`. +- Run a persistent sync loop; decrypt inbound / encrypt outbound in E2EE rooms. +- Route inbound text and media through gateway allowlists and Hermes-aligned mention/thread rules (via existing `mention_sessions`, `dm_session`, `subscriptions`). +- Mirror Inbox approvals and discrete-choice questions to Matrix rooms; resolve via emoji reactions with optional sender binding in the adapter. +- Send text (`send_message`) and files (`send_file`) to Matrix targets using the new encoding. +- Persist crypto state across server restarts. + +**Non-Goals:** + +- OpenWorker Cloud managed OAuth / relay for Matrix. +- Matrix admin agent tools (`matrix_create_room`, `matrix_invite_user`, etc.) — future work. +- Sliding Sync / MSC3575. +- VoIP / livekit. +- Public matrix.org-specific assumptions (works with any Synapse-compatible HS URL). +- Replacing Slack/Telegram target format. +- Dedicated `MatrixSessionStore` until per-user room session routing ships (task 3.3). + +## Decisions + +### 1. SDK: `matrix-nio[e2e]` + +**Rationale:** Ships on PyPI under the existing `messaging` extra. Hermes uses mautrix for reference, but mautrix is not a practical PyPI dep here. Cross-signing bootstrap is implemented in `matrix_crypto_bootstrap.py` (matrix-nio has no SSSS support). + +**Alternative:** `mautrix[encryption]` — closer to Hermes source, but heavier packaging and not chosen for v1. + +### 2. Target format: `matrix/[/thread/]` + +**Rationale:** Room IDs and event IDs contain `:`. A dedicated prefix branch in `parse_target` keeps Slack/Telegram unchanged. Base64url without padding is stable for standing grants and mention thread maps. + +**Alternative:** Pipe separator (`matrix|room|thread`) — human-readable but breaks consistency with existing colon grammar. + +### 3. Session routing: existing OpenWorker primitives (no separate store in v1) + +**Rationale:** DM → `dm_session()`; `@mention` in rooms → `_route_mention` with matrix-encoded thread targets; passive fan-out → `subscriptions`. A persisted `MatrixSessionStore` for `group_sessions_per_user` / `session_scope` is deferred to task 3.3. + +### 4. Reactions as Inbox interactions + +**Rationale:** Matrix has no Block Kit. Hermes maps ✅ / ♾️ / ❌ / number emojis to approval resolutions. + +**Approach:** + +- `send_interactive()` posts explanatory text and registers `(room_id, message_event_id) → {emoji_map, allowed_reactor?}` in `PendingReactionStore`. +- Inbound `m.reaction` events produce `InteractionEvent` with `interaction_kind="reaction"`; reuse `interactions.encode/decode` for `(item_id, resolution)`. +- ♾️ maps to standing grant (`allow_always`) via existing permissions API. +- `approval_require_sender` (default true) is enforced in `MatrixAdapter._handle_reaction` when `allowed_reactor` is set on the pending prompt. + +**Lifecycle reactions** (👀/✅/❌ on inbound processing) — future work (task 4.7). + +### 5. Media pipeline + +**Inbound:** + +| msgtype | Agent delivery | +|---------|----------------| +| `m.text` | Existing `MessageEvent` → `tagged_text` | +| `m.image` | Download decrypted bytes → `image_url` data URL part + text caption (if model supports vision; else text-only fallback) | +| `m.file` / `m.audio` / `m.video` | Save under `{crypto_store}/inbound/`; tagged text references path | + +**Outbound (`send_file`):** + +- Upload via matrix-nio AsyncClient (encrypted when room requires). +- `room_send` with appropriate `msgtype` and `mxc://` URL; preserve thread relation. + +**Security:** Reject non-`mxc://` media URLs in events. Enforce `max_media_bytes` before download. + +### 6. Configuration storage + +Fields on the `matrix:default` SecretStore profile (plus standard `allowed_users`): + +``` +homeserver_url, access_token, user_id, recovery_key (secret) +allowed_rooms[], free_response_rooms[], ignore_user_patterns[] +require_mention, auto_thread, e2ee_mode (=required) +max_message_length, max_media_bytes, approval_require_sender +``` + +Room allowlists are read only by `MatrixSettings` (not duplicated on `ConnectorSettings`). Advanced Hermes flags (`session_scope`, `group_sessions_per_user`, lifecycle reactions, etc.) ship when task 3.3 / 4.7 land. + +Connect wizard minimum: HS URL + token + recovery key + allowed users. + +### 7. Crypto store location + +`{openworker_state}/matrix/store/` — matrix-nio sqlite crypto DB + sync tokens. Never log recovery keys. On stale OTM key detection, fail connect with actionable error (new access token / device). + +### 8. Testing + +Unit tests for target encoding, mapper, reactions, and crypto bootstrap (no network). Integration harness with libolm in messaging CI is future work (task 7.1). + +## Risks / Trade-offs + +| Risk | Mitigation | +|------|------------| +| libolm missing on user machine | `e2ee_mode=required` → connect fails with install instructions; no silent fallback | +| DMG packaging libolm | Separate release task; document brew/apt deps for dev | +| Reaction race (multiple users react) | First resolved wins; clear pending registry | +| Non-vision models receive images | Detect capability; fall back to text description + saved path | +| Federation abuse via media URLs | mxc-only; size cap; room allowlist for private deploys | +| matrix-nio vs Hermes mautrix drift | Document in design; cross-signing bootstrap is custom | + +## Migration Plan + +- **New connector** — no migration from existing profiles. +- **Rollback:** Disconnect matrix in GUI; disable gateway adapter; crypto store remains on disk for reconnect. +- **Upgrade note:** Deleting crypto store requires new access token (document in connector instructions). + +## Open Questions + +- _(none blocking — decisions locked in explore session)_ diff --git a/openspec/changes/add-matrix-connector/proposal.md b/openspec/changes/add-matrix-connector/proposal.md new file mode 100644 index 0000000000..e8a3e998ab --- /dev/null +++ b/openspec/changes/add-matrix-connector/proposal.md @@ -0,0 +1,35 @@ +## Why + +Teams running self-hosted Synapse or Element Server Suite (ESS) use Element as their primary chat surface, often with end-to-end encryption (E2EE) enabled by default. OpenWorker today supports two-way messaging only on Telegram and Slack; there is no path for a coworker agent to join encrypted Matrix rooms, receive @mentions and media, or resolve Inbox approvals from Element. Adding a first-class `matrix` connector closes that gap for private deployments without relying on OpenWorker Cloud relay. + +## What Changes + +- Add a **`matrix` connector** (descriptor, GUI branding, SecretStore profile) for self-hosted homeservers. +- Implement **`MatrixAdapter`** using **`matrix-nio[e2e]`** with **`e2ee_mode=required`** (fail closed — no silent downgrade). +- Implement **cross-signing bootstrap** in `matrix_crypto_bootstrap.py` (recovery key → self-signing device signature; matrix-nio has no built-in SSSS). +- Introduce a **Matrix-specific target address format** (`matrix/[/thread/]`) for `send_message` and `send_file`. +- Align inbound behavior with **Hermes Matrix semantics** where implemented: mention gating, threads, room/user allowlists, bridge ghost filtering; DM via `dm_session`, mentions via `_route_mention`. +- Extend **Inbox mirroring** to Matrix via **emoji reactions** (✅ / ♾️ / ❌ / numbered options) instead of Slack Block Kit buttons. +- Support **inbound and outbound media** (image, file, audio, video) through Matrix `mxc://` URIs with size limits and E2EE encrypt/decrypt. +- Register `matrix` in gateway **`PLATFORMS`**, `DEFAULT_SENDERS`, and `DEFAULT_FILE_SENDERS`. +- Add **`matrix-nio[e2e]>=0.25`** under the existing `messaging` optional extra; document **libolm** as a system prerequisite. +- **No managed cloud relay** for Matrix (manual token connect only at launch). + +## Capabilities + +### New Capabilities + +- `matrix-connector`: E2EE Matrix messaging connector — connect/auth, sync loop, Hermes-aligned routing, reaction-based Inbox interactions, and encrypted media in/out. +- `messaging-target-format`: Matrix target token encoding and parsing used by outbound tools and standing grants. + +### Modified Capabilities + +_(none — no existing `openspec/specs/` baseline in this repo)_ + +## Impact + +- **Backend**: `coworker/connectors/` (new adapter, senders, config, descriptors, catalog_copy, tools), `coworker/server/manager.py` (inbound routing, Inbox mirror), `coworker/connectors/base.py` (`InteractionEvent`, target parsing). +- **Frontend**: `surfaces/gui/src/connectors/registry.tsx`, new or extended connector detail UI for Matrix credentials (task 6.2). +- **Dependencies**: `pyproject.toml` `[messaging]` extra → `matrix-nio[e2e]`; system `libolm` 3.x (dev docs + release packaging follow-up). +- **Tests**: unit tests for target encoding, reactions, crypto bootstrap, mapper; integration harness deferred (task 7.1). +- **Security**: encrypted crypto store on disk, recovery key in SecretStore, federated/untrusted input treated as hostile (`mxc://` only, room allowlists encouraged). diff --git a/openspec/changes/add-matrix-connector/specs/matrix-connector/spec.md b/openspec/changes/add-matrix-connector/specs/matrix-connector/spec.md new file mode 100644 index 0000000000..1e76dc3615 --- /dev/null +++ b/openspec/changes/add-matrix-connector/specs/matrix-connector/spec.md @@ -0,0 +1,140 @@ +## ADDED Requirements + +### Requirement: Matrix connector registration + +The system SHALL expose a connector named `matrix` in the connector catalog with `two_way=true`, `channels=true`, `managed=false`, and `available=true`. The descriptor SHALL require `homeserver_url`, `access_token`, and `recovery_key` (secret), and SHALL validate credentials via the Matrix Client-Server `/account/whoami` endpoint. The GUI SHALL display Element/Matrix branding via `logo=matrix`. + +#### Scenario: Successful connect with valid token + +- **WHEN** the user submits a valid homeserver URL, access token, and recovery key +- **THEN** the connector reports connected status and persists a `matrix:default` profile without returning secrets to the client + +#### Scenario: Connect fails without recovery key + +- **WHEN** the user omits the recovery key on a homeserver with cross-signing enabled +- **THEN** validation or connect SHALL fail with an actionable error explaining the recovery key requirement + +### Requirement: E2EE required mode + +The Matrix adapter SHALL operate with `e2ee_mode=required`. If `libolm` or crypto initialization fails, connect SHALL fail closed and MUST NOT fall back to an unencrypted client. + +#### Scenario: Missing libolm + +- **WHEN** the messaging extra is installed but libolm is absent on the system +- **THEN** gateway start for matrix SHALL fail with install instructions + +#### Scenario: Encrypted room message round-trip + +- **WHEN** an authorized user sends a text message in an E2EE room where the bot is joined +- **THEN** the bot decrypts the event, routes it through the gateway, and can send an encrypted reply visible in Element + +### Requirement: Homeserver allowlists + +The connector SHALL support `allowed_users` (Matrix user IDs) and `allowed_rooms` (room IDs). When `allowed_rooms` is non-empty, inbound events from other rooms SHALL be ignored except direct-message rooms. When `allowed_users` is empty, the default SHALL remain deny-all for inbound (consistent with other messaging connectors) until users are captured and added. + +#### Scenario: Message from disallowed room + +- **WHEN** `allowed_rooms` is configured and a message arrives from a room not in the list +- **THEN** the gateway SHALL NOT dispatch the message to the agent + +#### Scenario: Message from allowed user in allowed room + +- **WHEN** both allowlists pass +- **THEN** the message proceeds to routing + +### Requirement: Hermes-aligned mention and thread behavior + +The adapter SHALL implement, in v1: `require_mention` (default true for rooms), `free_response_rooms`, and `auto_thread`. Threads where the bot has already participated SHALL NOT require a repeat @mention. + +`session_scope`, `dm_mention_threads`, `dm_auto_thread`, and `group_sessions_per_user` are specified for Hermes parity but deferred (task 3.3); v1 routes DMs via `dm_session()`, channel @mentions via `_route_mention`, and passive channel traffic via `subscriptions`. + +Implementation uses **`matrix-nio[e2e]`** with custom cross-signing bootstrap in `matrix_crypto_bootstrap.py` (not mautrix). + +#### Scenario: Room message without mention + +- **WHEN** `require_mention=true` and a room message does not mention the bot and the room is not in `free_response_rooms` +- **THEN** the message SHALL NOT spawn or steer an agent turn (subscription fan-out rules still apply) + +#### Scenario: Thread continuation without mention + +- **WHEN** the bot previously replied in a Matrix thread and a follow-up arrives in that thread +- **THEN** the message SHALL be routed as a continuation without requiring @mention + +### Requirement: Per-user session isolation in shared rooms + +When `group_sessions_per_user=true`, two authorized users messaging in the same room SHALL map to distinct agent sessions according to `session_scope`. **Deferred to task 3.3** — not required for v1 ship. + +#### Scenario: Two users same room + +- **WHEN** Alice and Bob each send messages in the same project room +- **THEN** their conversation context SHALL NOT share a single session transcript unless `group_sessions_per_user=false` + +### Requirement: Reaction-based Inbox mirroring + +When an Inbox item with discrete choices is mirrored to a Matrix-bound channel, the system SHALL post a text prompt and resolve the item when an authorized user adds a mapped emoji reaction. Approval items SHALL support ✅ (approve once), ♾️ (approve always / standing grant), and ❌ (deny). Question items with options SHALL support numbered emoji reactions (1️⃣, 2️⃣, …). + +#### Scenario: Approve via reaction + +- **WHEN** a pending approval is mirrored to Matrix and the requester reacts ✅ on the prompt message +- **THEN** the Inbox item resolves as allow and the suspended agent continues + +#### Scenario: Deny non-requester when sender-bound + +- **WHEN** `approval_require_sender=true` and a different user reacts ✅ on an approval prompt +- **THEN** the reaction SHALL NOT resolve the item + +#### Scenario: Approve always via infinity reaction + +- **WHEN** the requester reacts ♾️ on an approval prompt +- **THEN** the item resolves and a standing grant equivalent to allow-always is recorded for the tool context + +### Requirement: Inbound media handling + +The adapter SHALL handle inbound `m.room.message` events with `msgtype` of `m.image`, `m.file`, `m.audio`, or `m.video`. Media content URIs MUST be `mxc://` only. Downloads SHALL respect `max_media_bytes` (default 104857600). Images SHALL be delivered to the agent as multimodal content when the session model supports vision; otherwise a text fallback describing the attachment path SHALL be used. Non-image files SHALL be saved under the active session workspace and referenced in the inbound message text. + +#### Scenario: Inbound encrypted image + +- **WHEN** a user sends an image in an E2EE room +- **THEN** the bot decrypts and downloads the image and the agent turn includes viewable image content or an explicit fallback description + +#### Scenario: Reject oversize media + +- **WHEN** an attachment exceeds `max_media_bytes` +- **THEN** the adapter SHALL NOT download the full content and SHALL surface a text notification instead + +#### Scenario: Reject HTTP media URL + +- **WHEN** an event references a non-mxc media URL +- **THEN** the adapter SHALL ignore the media fetch + +### Requirement: Outbound media via send_file + +The `send_file` tool SHALL support Matrix targets. Files SHALL be uploaded through the Matrix client (encrypted when required) and sent with the correct `msgtype`, preserving thread context when `auto_thread` is active. + +#### Scenario: Send PDF to encrypted room + +- **WHEN** the agent calls `send_file` with a valid Matrix target and a workspace file +- **THEN** recipients in Element can download the file from the encrypted room + +#### Scenario: Standing send_message grant excludes send_file + +- **WHEN** a session has a pre-approved `send_message` grant for a Matrix thread target +- **THEN** `send_file` to the same target SHALL still require approval + +### Requirement: Bridge ghost filtering + +The adapter SHALL ignore messages from senders matching configured `ignore_user_patterns` (default includes common bridge prefixes such as `^@telegram_`, `^@slack_`, `^@whatsapp_`). + +#### Scenario: Bridge ghost message ignored + +- **WHEN** a message arrives from `@telegram_123:example.org` and the pattern matches +- **THEN** the gateway SHALL NOT dispatch the message + +### Requirement: Auto-join on invite + +The bot SHALL automatically accept room invites and join encrypted rooms when invited, initializing encryption sessions as needed. + +#### Scenario: Invite to encrypted room + +- **WHEN** an authorized user's invite arrives for an E2EE room +- **THEN** the bot joins and can decrypt subsequent messages diff --git a/openspec/changes/add-matrix-connector/specs/messaging-target-format/spec.md b/openspec/changes/add-matrix-connector/specs/messaging-target-format/spec.md new file mode 100644 index 0000000000..1ed9c60827 --- /dev/null +++ b/openspec/changes/add-matrix-connector/specs/messaging-target-format/spec.md @@ -0,0 +1,38 @@ +## ADDED Requirements + +### Requirement: Matrix target encoding + +Outbound tools (`send_message`, `send_file`) and standing approval grants SHALL address Matrix destinations using the format `matrix/[/thread/]`. Encoding SHALL use URL-safe base64 without padding. Decoding SHALL recover the full Matrix room ID and optional thread root event ID. + +#### Scenario: Parse room-only target + +- **WHEN** the target is `matrix/` where decoded room id is `!abc:example.org` +- **THEN** `parse_target` returns platform `matrix`, chat_id `!abc:example.org`, and thread_id `None` + +#### Scenario: Parse threaded target + +- **WHEN** the target includes `/thread/` +- **THEN** `parse_target` returns the decoded thread root event id as `thread_id` + +#### Scenario: Invalid matrix target rejected + +- **WHEN** the target starts with `matrix/` but base64 decoding fails +- **THEN** parsing SHALL raise a clear validation error + +### Requirement: Matrix targets in inbound reply handles + +Inbound Matrix messages SHALL expose reply handles using the same encoding in `SessionSource.target` and in agent-facing tagged text, so agents can pass the handle back to `send_message` unchanged. + +#### Scenario: Inbound tagged text includes matrix target + +- **WHEN** a message arrives from room `!ops:example.org` in thread rooted at `$event123` +- **THEN** the tagged inbound text includes a reply handle matching the encoded matrix target format + +### Requirement: Slack and Telegram targets unchanged + +Existing `platform:chat_id[:thread]` parsing for non-matrix platforms SHALL remain backward compatible. + +#### Scenario: Slack target still parses + +- **WHEN** the target is `slack:C0123456789:1700000000.000100` +- **THEN** parsing returns platform slack with the expected chat_id and thread_id diff --git a/openspec/changes/add-matrix-connector/tasks.md b/openspec/changes/add-matrix-connector/tasks.md new file mode 100644 index 0000000000..b1a17d9696 --- /dev/null +++ b/openspec/changes/add-matrix-connector/tasks.md @@ -0,0 +1,60 @@ +## 1. Infrastructure and target format + +- [x] 1.1 Add `encode_matrix_target` / `decode_matrix_target` and extend `parse_target` with `matrix/` branch in `coworker/connectors/base.py` +- [x] 1.2 Add unit tests for matrix target round-trip, invalid input, and Slack/Telegram regression +- [x] 1.3 Add `matrix` to `PLATFORMS` in `coworker/connectors/config.py` with user allowlist loading from profile +- [x] 1.4 Add `ConnectorDescriptor` for `matrix` in `descriptors.py` (fields, validate via whoami, profile schema) +- [x] 1.5 Add `ABOUT` / `ACCESS` copy in `catalog_copy.py` +- [x] 1.6 Add `matrix-nio[e2e]>=0.25` to `pyproject.toml` `[project.optional-dependencies] messaging` and document libolm prerequisite + +## 2. E2EE adapter core + +- [x] 2.1 Create `coworker/connectors/matrix_adapter.py` using matrix-nio AsyncClient with required E2EE +- [x] 2.2 Implement crypto store under state dir, recovery key import via `matrix_crypto_bootstrap.py`, stale OTM key detection with fail-closed connect +- [x] 2.3 Implement `matrix_event_to_event()` mapper for text messages with mention detection and room/DM chat_type +- [x] 2.4 Wire `make_adapter("matrix", …)` and gateway registration in `manager._build_and_start_gateway` +- [x] 2.5 Implement encrypted outbound `send()` on the adapter (not httpx-only sender) +- [x] 2.6 Implement auto-join on invite for encrypted rooms + +## 3. Hermes session and routing behavior + +- [ ] 3.1 ~~Create `MatrixSessionStore`~~ deferred — use `dm_session`, `_route_mention`, `subscriptions` until per-user room scoping ships +- [x] 3.2 Implement `require_mention`, `free_response_rooms`, thread continuation without re-mention +- [ ] 3.3 Implement `session_scope`, `dm_mention_threads`, `group_sessions_per_user` routing (+ optional `MatrixSessionStore`) +- [x] 3.4 Integrate with `_route_mention`, `dm_session`, and `subscriptions` paths using matrix-encoded thread targets +- [x] 3.5 Apply `ignore_user_patterns` at ingress (`process_notices` deferred) + +## 4. Reaction-based Inbox interactions + +- [x] 4.1 Extend `InteractionEvent` with `interaction_kind` and `reaction_key` (backward compatible for Slack) +- [x] 4.2 Add `PendingReactionStore` keyed by (room_id, prompt_event_id) +- [x] 4.3 Implement `MatrixAdapter.send_interactive()` — post prompt text and register emoji map +- [x] 4.4 Handle inbound `m.reaction` → `InteractionEvent` with encoded resolution (✅ / ♾️ / ❌ / numbers); `approval_require_sender` in adapter when `allowed_reactor` set +- [x] 4.5 Extend `manager.mirror_inbox_item` for matrix channel targets +- [x] 4.6 Wire ♾️ reaction to standing grant API (resolution `always` via inbox) +- [ ] 4.7 Add optional lifecycle reactions (👀/✅/❌) separate from approval emoji map + +## 5. Media inbound and outbound + +- [x] 5.1 Implement inbound mxc download with E2EE decrypt and `max_media_bytes` guard +- [x] 5.2 Deliver inbound images as multimodal `image_url` parts with vision-capability fallback +- [x] 5.3 Save inbound file/audio/video under `{crypto_store}/inbound/` and reference in tagged text +- [x] 5.4 Reject non-mxc media URLs in events +- [x] 5.5 Implement `_send_matrix_file` and register in `DEFAULT_FILE_SENDERS` +- [x] 5.6 Update `send_message` / `send_file` tool descriptions to include Matrix target format + +## 6. GUI and packaging + +- [x] 6.1 Add Matrix/Element logo to `surfaces/gui/src/connectors/registry.tsx` +- [ ] 6.2 Add Matrix connector detail view (credentials + advanced Hermes flags) +- [x] 6.3 Document libolm install for macOS/Linux in connector instructions +- [ ] 6.4 Track DMG libolm bundling as release follow-up (document in design if not implemented in this change) + +## 7. Test harness and verification + +- [ ] 7.1 Create integration mock homeserver harness (sync + send + reaction + mxc stub) — not started +- [x] 7.2 Tests: target encoding, mapper, allowlist room/user, mention gating +- [x] 7.3 Tests: reaction approve/deny/sender-bound/approve-always +- [ ] 7.4 Tests: inbound image multimodal + outbound send_file (mock crypto layer for CI) +- [ ] 7.5 Tests: `group_sessions_per_user` session isolation (blocked on 3.3) +- [ ] 7.6 Run messaging test job with libolm available; document skip behavior when absent diff --git a/packaging/build_dmg.sh b/packaging/build_dmg.sh index 724ce3e8cf..79e16fe0f9 100755 --- a/packaging/build_dmg.sh +++ b/packaging/build_dmg.sh @@ -7,7 +7,8 @@ # 4. Wrap the .app in a compressed .dmg via hdiutil (reliable + headless; Tauri's own # bundle_dmg.sh uses Finder AppleScript and fails in non-interactive sessions). # -# Prerequisites (mirrors build_windows.ps1's header): +# - Matrix E2EE (optional connector): libolm must be on the PATH at runtime +# (`brew install libolm`). Bundling libolm into the DMG is release follow-up (task 6.4). # - Rust (rustup) + Node/npm, and the GUI deps installed (npm ci in surfaces/gui). # - A Python venv at .venv (repo root) with this package installed editable, plus the # build-only deps: diff --git a/pyproject.toml b/pyproject.toml index bac1597928..adea7b3353 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,7 +43,7 @@ dev = ["pytest>=8", "pytest-asyncio", "httpx"] # Inbound messaging listeners (outbound send_message needs only httpx, already a core dep). # aiohttp is slack-bolt's Socket Mode transport at runtime (and the FakeSlack test harness # drives the real handler) — declare it so CI installs it, not just transitively. -messaging = ["python-telegram-bot>=21", "slack-bolt>=1.18", "aiohttp>=3.9"] +messaging = ["python-telegram-bot>=21", "slack-bolt>=1.18", "aiohttp>=3.9", "matrix-nio[e2e]>=0.25"] # Interactive Cowork browser automation. browser = ["playwright>=1.44"] # AWS Bedrock provider (lazy-imported; desktop builds bundle it, pip users opt in). diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts index d87322d361..92ee6b1dde 100644 --- a/surfaces/gui/src/api.ts +++ b/surfaces/gui/src/api.ts @@ -655,6 +655,21 @@ export interface Connector { portals?: HubSpotPortal[]; // HubSpot only: connected portals (multi-portal) hidden_fields?: string[]; // HubSpot only: properties stripped from agent reads installations?: GithubInstallation[]; // GitHub only: App installations (managed relay) + matrix_settings?: MatrixSettings; // Matrix: advanced Hermes flags (no secrets) +} + +export interface MatrixSettings { + homeserver_url: string; + user_id?: string | null; + require_mention: boolean; + auto_thread: boolean; + session_scope: string; + dm_mention_threads: boolean; + dm_auto_thread: boolean; + group_sessions_per_user: boolean; + lifecycle_reactions: boolean; + allowed_rooms: string[]; + free_response_rooms: string[]; } // --- OpenWorker Cloud (optional sign-in; manual token paste always works) --- @@ -2258,6 +2273,17 @@ export async function setGmailFilters(filters: { senders?: string[]; labels?: st return res.json(); } +export async function patchMatrixSettings( + body: Partial, +): Promise<{ ok: boolean; error?: string; settings?: MatrixSettings }> { + const res = await fetch(`${httpBase()}/v1/connectors/matrix/settings`, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }); + return res.json(); +} + // GitHub relay health, the Slack three-layer shape: shared relay socket / // cloud sign-in / per-installation token health (+ missed-event counts). export interface GithubStatus { diff --git a/surfaces/gui/src/components/connectors/ConnectorsSection.tsx b/surfaces/gui/src/components/connectors/ConnectorsSection.tsx index 9d64aa0a57..4a8435f698 100644 --- a/surfaces/gui/src/components/connectors/ConnectorsSection.tsx +++ b/surfaces/gui/src/components/connectors/ConnectorsSection.tsx @@ -21,6 +21,7 @@ import { ConnectorsList } from "./ConnectorsList"; import { GithubDetail } from "./GithubDetail"; import { GmailDetail } from "./GmailDetail"; import { HubSpotDetail } from "./HubSpotDetail"; +import { MatrixDetail } from "./MatrixDetail"; import { SlackDetail } from "./SlackDetail"; import { GRP } from "./ui"; @@ -39,6 +40,7 @@ export interface DetailProps { // Bespoke pages register here; everything else gets GenericDetail below. const DETAIL_PAGES: Record JSX.Element> = { slack: (p) => , + matrix: (p) => , gmail: (p) => , google_calendar: (p) => , hubspot: (p) => , diff --git a/surfaces/gui/src/components/connectors/MatrixDetail.tsx b/surfaces/gui/src/components/connectors/MatrixDetail.tsx new file mode 100644 index 0000000000..1a1d357534 --- /dev/null +++ b/surfaces/gui/src/components/connectors/MatrixDetail.tsx @@ -0,0 +1,230 @@ +import { useEffect, useState } from "react"; +import { disconnectConnector, patchMatrixSettings, type Connector, type MatrixSettings } from "../../api"; +import { ConnectorBadge } from "../../connectors/ConnectorIcon"; +import { AllowlistBlock, ListeningSessionsBlock, UnauthorizedBlock } from "../ManageTabs"; +import type { DetailProps } from "./ConnectorsSection"; +import { ToolsDisclosure } from "./ToolsDisclosure"; +import { FOOT, GRP, GRP_H, ROW } from "./ui"; + +const LABEL = "text-[12.5px] text-muted w-36 shrink-0"; + +function Toggle({ + checked, + onChange, + testId, +}: { + checked: boolean; + onChange: (v: boolean) => void; + testId: string; +}) { + return ( + + ); +} + +export function MatrixDetail({ c, onChanged }: DetailProps) { + const ms = c.matrix_settings; + const [busy, setBusy] = useState(false); + const [local, setLocal] = useState(ms ?? null); + + useEffect(() => { + if (ms) setLocal(ms); + }, [ms]); + + const save = async (patch: Partial) => { + setBusy(true); + try { + const res = await patchMatrixSettings(patch); + if (res.settings) setLocal(res.settings); + onChanged(); + } finally { + setBusy(false); + } + }; + + return ( +
+
+ +
+

Matrix

+
+ + {c.account || local?.user_id || "Connected"} +
+
+ +
+ + {local && ( + <> +
Homeserver
+
+
+ URL + {local.homeserver_url} +
+ {local.user_id && ( +
+ Bot user + {local.user_id} +
+ )} +
+ +
Routing
+
+
+ Require @mention + save({ require_mention: v })} + /> +
+
+ Auto-thread replies + save({ auto_thread: v })} + /> +
+
+ Per-user sessions + save({ group_sessions_per_user: v })} + /> +
+
+ DM @mention → thread + save({ dm_mention_threads: v })} + /> +
+
+ Session scope + +
+
+ Lifecycle reactions + save({ lifecycle_reactions: v })} + /> +
+
+ +
Room allowlists
+
+ save({ allowed_rooms: rooms })} + /> + save({ free_response_rooms: rooms })} + /> +
+ + )} + +
+ + + + +
+ +
+ E2EE requires libolm on this machine. Reconnect after changing homeserver credentials in + Connectors. +
+
+ ); +} + +function ConnectorToolsWrap({ c, onChanged }: { c: Connector; onChanged: () => void }) { + return ; +} + +function RoomListEditor({ + label, + hint, + value, + testId, + onSave, +}: { + label: string; + hint: string; + value: string[]; + testId: string; + onSave: (rooms: string[]) => void; +}) { + const [draft, setDraft] = useState(value.join(", ")); + return ( +
+
{label}
+
{hint}
+