nisse/nisse-dev/src/nisse_dev/matrix.py

"""MatrixTransport (architecture L5): Matrix as strictly a UI channel.

Implements the Transport interface directly over the Matrix client-server HTTP
API (token auth, /sync long-poll, /send) with the httpx client already in the
dependency tree — no Matrix SDK. The surface used is four endpoints; an SDK
earns its keep when E2E arrives, and bot rooms are created unencrypted until
then, so encrypted rooms are not supported here.

Inbound handling follows L5's intake posture: on homeserver errors the sync
loop backs off (capped) and retries indefinitely — an outage delays reactions
but never fails anything.
"""

import html
import re
import threading
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from urllib.parse import unquote

import httpx

from nisse_dev.transport import InboundMessage

# A mention pill at the start of an HTML body: the link holds the user ID,
# the link text is what the plain body starts with.
_LEADING_PILL = re.compile(
    r'\s*<a href="https://matrix\.to/#/([^"?]+)[^"]*">(.*?)</a>', re.DOTALL
)
# A user ID typed at the start of a plain body, e.g. "@alice:example: ...".
_LEADING_USER_ID = re.compile(r"\s*(@[^\s:]+:[^\s:]+)")


class TransportError(RuntimeError):
    pass


@dataclass(frozen=True)
class MatrixConfig:
    homeserver: str
    user_id: str
    token: str
    room_id: str


class MatrixTransport:
    def __init__(
        self,
        config: MatrixConfig,
        *,
        poll_timeout_ms: int = 30000,
        http_transport: httpx.BaseTransport | None = None,
    ) -> None:
        self._config = config
        self._poll_timeout_ms = poll_timeout_ms
        self._http = httpx.Client(
            base_url=f"{config.homeserver}/_matrix/client/v3",
            headers={"Authorization": f"Bearer {config.token}"},
            timeout=(poll_timeout_ms / 1000) + 30,
            transport=http_transport,
        )
        self._callback: Callable[[InboundMessage], None] | None = None
        # The agent's display name in the room, learnt from its own member
        # event; clients write it in front of messages addressed to it.
        self._display_name: str | None = None
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    # -- outbound ---------------------------------------------------------

    def send_message(self, text: str) -> None:
        self._send(text)

    def send_approval_request(self, text: str) -> None:
        self._send(text)

    def ask_question(self, text: str) -> None:
        self._send(text)

    def _send(self, text: str) -> None:
        room = self._config.room_id
        txn = uuid.uuid4().hex
        try:
            response = self._http.put(
                f"/rooms/{room}/send/m.room.message/{txn}",
                json={"msgtype": "m.text", "body": text},
            )
        except httpx.HTTPError as exc:
            raise TransportError(f"matrix send failed: {exc}") from exc
        if response.status_code != 200:
            raise TransportError(
                f"matrix send returned {response.status_code}: {response.text[:300]}"
            )

    # -- inbound ----------------------------------------------------------

    def on_message(self, callback: Callable[[InboundMessage], None]) -> None:
        self._callback = callback

    def start(self) -> None:
        """Begin delivering new room messages to the registered callback.
        History from before start() is never delivered (Matrix room history
        must not become load-bearing context, L5)."""
        if self._callback is None:
            raise TransportError("no message handler registered; call on_message first")
        self._stop.clear()
        self._thread = threading.Thread(target=self._sync_loop, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=(self._poll_timeout_ms / 1000) + 5)
            self._thread = None

    def _sync_loop(self) -> None:
        since: str | None = None
        backoff = 1.0
        while not self._stop.is_set():
            params: dict[str, str] = {}
            if since is None:
                params["timeout"] = "0"  # first sync: learn position, skip backlog
            else:
                params["timeout"] = str(self._poll_timeout_ms)
                params["since"] = since
            try:
                response = self._http.get("/sync", params=params)
                response.raise_for_status()
            except httpx.HTTPError:
                self._stop.wait(backoff)
                backoff = min(backoff * 2, 60.0)
                continue
            backoff = 1.0
            data = response.json()
            self._learn_display_name(data)
            if since is not None:
                self._deliver(data)
            since = data.get("next_batch", since)

    def _room(self, sync_data: dict) -> dict:
        return sync_data.get("rooms", {}).get("join", {}).get(self._config.room_id, {})

    def _learn_display_name(self, sync_data: dict) -> None:
        room = self._room(sync_data)
        events = room.get("state", {}).get("events", []) + room.get("timeline", {}).get(
            "events", []
        )
        for event in events:
            if (
                event.get("type") == "m.room.member"
                and event.get("state_key") == self._config.user_id
            ):
                name = event.get("content", {}).get("displayname")
                if name:
                    self._display_name = name

    def _deliver(self, sync_data: dict) -> None:
        for event in self._room(sync_data).get("timeline", {}).get("events", []):
            if event.get("type") != "m.room.message":
                continue
            sender = event.get("sender", "")
            if sender == self._config.user_id:
                continue
            content = event.get("content", {})
            body = content.get("body")
            if not body:
                continue
            addressee, text = self._addressing(content)
            if addressee is not None and addressee != self._config.user_id:
                # Addressed to someone else: in a shared room, "Nisse Ops:
                # approve" must not reach this agent as rejection feedback.
                continue
            assert self._callback is not None
            self._callback(
                InboundMessage(
                    sender=sender,
                    text=text,
                    original=body if text != body else None,
                )
            )

    def _addressing(self, content: dict) -> tuple[str | None, str]:
        """Who a leading mention addresses, and the body without a leading
        mention of this agent. Only the start of the message counts: a
        mention later on ("approve, cc @alice") is part of the text. The
        sender is never read from here; it comes from the event (L3)."""
        body: str = content["body"]
        own = self._config.user_id
        if content.get("format") == "org.matrix.custom.html":
            pill = _LEADING_PILL.match(content.get("formatted_body", ""))
            if pill:
                addressee = unquote(pill.group(1))
                label = html.unescape(re.sub(r"<[^>]+>", "", pill.group(2)))
                if addressee == own and label and body.startswith(label):
                    return own, _after_prefix(body, len(label))
                return addressee, body
        if body.startswith(own):
            return own, _after_prefix(body, len(own))
        name = self._display_name
        if name and body.startswith(name) and body[len(name) :].startswith(":"):
            return own, _after_prefix(body, len(name))
        typed = _LEADING_USER_ID.match(body)
        if typed and body[typed.end() :].startswith(":"):
            return typed.group(1), body
        return None, body

    def close(self) -> None:
        self.stop()
        self._http.close()


def _after_prefix(body: str, length: int) -> str:
    """The body past a leading mention and the ':' clients put after it."""
    return body[length:].lstrip().removeprefix(":").strip()