Merge pull request 'Matrix threads in the transport (M2.1, PR A)' (#96) from matrix-threads into main

Reviewed-on: dongar/nisse#96
Reviewed-by: Don Garrett <dgarrett@acm.org>

nisse-dev/src/nisse_dev/matrix.py

@@ -9,12 +9,18 @@ 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.
+
+Threads (L5): each work item's conversation is a Matrix thread. A message in a
+thread carries its root's event ID; a plain reply from a client without
+thread support is routed to the thread of the message it replies to, when
+that is known. The quoted lines such replies carry are not part of the text.
 """
 
 import html
 import re
 import threading
 import uuid
+from collections import OrderedDict
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from urllib.parse import unquote
@@ -30,6 +36,8 @@ _LEADING_PILL = re.compile(
 )
 # A user ID typed at the start of a plain body, e.g. "@alice:example: ...".
 _LEADING_USER_ID = re.compile(r"\s*(@[^\s:]+:[^\s:]+)")
+# How many event-to-thread mappings to remember, for routing plain replies.
+_REMEMBERED_EVENTS = 2000
 
 
 class TransportError(RuntimeError):
@@ -66,25 +74,54 @@ class MatrixTransport:
         self._display_name: str | None = None
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
+        # event ID -> the root of the thread it is in (a root maps to
+        # itself), for routing replies from clients without thread support.
+        self._threads: OrderedDict[str, str] = OrderedDict()
+        self._threads_lock = threading.Lock()
+
+    def _remember(self, event_id: str, root: str) -> None:
+        with self._threads_lock:
+            self._threads[event_id] = root
+            self._threads.move_to_end(event_id)
+            while len(self._threads) > _REMEMBERED_EVENTS:
+                self._threads.popitem(last=False)
+
+    def _thread_of(self, event_id: str) -> str | None:
+        with self._threads_lock:
+            return self._threads.get(event_id)
 
     # -- outbound ---------------------------------------------------------
 
+    def post(self, text: str, *, thread: str | None = None) -> str:
+        content: dict = {"msgtype": "m.text", "body": text}
+        if thread is not None:
+            content["m.relates_to"] = {
+                "rel_type": "m.thread",
+                "event_id": thread,
+                # For clients without threads: show it as a reply to the root.
+                "is_falling_back": True,
+                "m.in_reply_to": {"event_id": thread},
+            }
+        event_id = self._send(content)
+        # A top-level post may root a thread; one in a thread belongs to it.
+        self._remember(event_id, thread or event_id)
+        return event_id
+
     def send_message(self, text: str) -> None:
-        self._send(text)
+        self.post(text)
 
     def send_approval_request(self, text: str) -> None:
-        self._send(text)
+        self.post(text)
 
     def ask_question(self, text: str) -> None:
-        self._send(text)
+        self.post(text)
 
-    def _send(self, text: str) -> None:
+    def _send(self, content: dict) -> str:
         room = self._config.room_id
         txn = uuid.uuid4().hex
         try:
             response = self._http.put(
-                f"/rooms/{room}/send/m.room.message/{txn}",
+                f"/rooms/{room}/send/m.room.message/{txn}", json=content
-                json={"msgtype": "m.text", "body": text},
             )
         except httpx.HTTPError as exc:
             raise TransportError(f"matrix send failed: {exc}") from exc
@@ -92,6 +129,7 @@ class MatrixTransport:
             raise TransportError(
                 f"matrix send returned {response.status_code}: {response.text[:300]}"
             )
+        return str(response.json().get("event_id", ""))
 
     # -- inbound ----------------------------------------------------------
 
@@ -160,12 +198,23 @@ class MatrixTransport:
             if event.get("type") != "m.room.message":
                 continue
             sender = event.get("sender", "")
+            content = event.get("content", {})
+            event_id = str(event.get("event_id", ""))
+            thread = self._thread_for(content)
+            if event_id:
+                self._remember(event_id, thread or event_id)
             if sender == self._config.user_id:
                 continue
-            content = event.get("content", {})
             body = content.get("body")
             if not body:
                 continue
+            if "m.in_reply_to" in content.get("m.relates_to", {}):
+                # A rich reply quotes what it replies to; the quote isn't
+                # what the sender said.
+                body = _strip_reply_fallback(body)
+                content = {**content, "body": 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:
@@ -177,9 +226,22 @@ class MatrixTransport:
                     sender=sender,
                     text=text,
                     original=body if text != body else None,
+                    event_id=event_id,
+                    thread=thread,
                 )
             )
 
+    def _thread_for(self, content: dict) -> str | None:
+        """The thread root a message belongs to, or None at top level."""
+        relates = content.get("m.relates_to") or {}
+        if relates.get("rel_type") == "m.thread" and relates.get("event_id"):
+            return str(relates["event_id"])
+        replied = (relates.get("m.in_reply_to") or {}).get("event_id")
+        if replied:
+            # A plain reply: in the thread of whatever it replies to, if known.
+            return self._thread_of(str(replied))
+        return 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
@@ -210,6 +272,20 @@ class MatrixTransport:
         self._http.close()
 
 
+def _strip_reply_fallback(body: str) -> str:
+    """The body of a rich reply without the quoted lines clients prepend
+    ("> <@alice:example> original text", then a blank line)."""
+    lines = body.split("\n")
+    if not lines or not lines[0].startswith("> "):
+        return body
+    index = 0
+    while index < len(lines) and lines[index].startswith(">"):
+        index += 1
+    if index < len(lines) and not lines[index].strip():
+        index += 1
+    return "\n".join(lines[index:]).strip()
+
+
 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()

nisse-dev/src/nisse_dev/transport.py

@@ -18,15 +18,27 @@ class InboundMessage:
     text: str
     # The message as sent, when that differs from text; the record keeps it.
     original: str | None = None
+    # The platform's ID for this message, and the root of the thread it was
+    # sent in (None at top level). A work item's conversation is a thread
+    # (L5), so this is how a reply finds its item.
+    event_id: str = ""
+    thread: str | None = None
 
 
 @dataclass(frozen=True)
 class OutboundMessage:
     kind: str  # "message" | "approval_request" | "question"
     text: str
+    thread: str | None = None
+    event_id: str = ""
 
 
 class Transport(Protocol):
+    def post(self, text: str, *, thread: str | None = None) -> str:
+        """Send a message, in a thread when one is given; returns its ID,
+        which can root a new thread."""
+        ...
+
     def send_message(self, text: str) -> None: ...
 
     def send_approval_request(self, text: str) -> None: ...
@@ -42,15 +54,27 @@ class InMemoryTransport:
     def __init__(self) -> None:
         self.outbound: list[OutboundMessage] = []
         self._callback: Callable[[InboundMessage], None] | None = None
+        self._next_id = 0
+
+    def _id(self) -> str:
+        self._next_id += 1
+        return f"$event{self._next_id}"
+
+    def post(self, text: str, *, thread: str | None = None) -> str:
+        event_id = self._id()
+        self.outbound.append(OutboundMessage("message", text, thread, event_id))
+        return event_id
 
     def send_message(self, text: str) -> None:
-        self.outbound.append(OutboundMessage("message", text))
+        self.outbound.append(OutboundMessage("message", text, None, self._id()))
 
     def send_approval_request(self, text: str) -> None:
-        self.outbound.append(OutboundMessage("approval_request", text))
+        self.outbound.append(
+            OutboundMessage("approval_request", text, None, self._id())
+        )
 
     def ask_question(self, text: str) -> None:
-        self.outbound.append(OutboundMessage("question", text))
+        self.outbound.append(OutboundMessage("question", text, None, self._id()))
 
     def on_message(self, callback: Callable[[InboundMessage], None]) -> None:
         self._callback = callback
@@ -61,6 +85,16 @@ class InMemoryTransport:
     def close(self) -> None:
         """Nothing to disconnect from."""
 
-    def receive(self, sender: str, text: str, original: str | None = None) -> None:
+    def receive(
+        self,
+        sender: str,
+        text: str,
+        original: str | None = None,
+        *,
+        thread: str | None = None,
+    ) -> str:
+        """Deliver a message as if the operator sent it; returns its ID."""
         assert self._callback is not None, "no message handler registered"
-        self._callback(InboundMessage(sender, text, original))
+        event_id = self._id()
+        self._callback(InboundMessage(sender, text, original, event_id, thread))
+        return event_id

nisse-dev/tests/test_matrix.py

@@ -237,3 +237,125 @@ def test_a_mention_later_in_the_message_is_left_alone() -> None:
     assert received == [
         InboundMessage(sender="@don:test", text="approve, cc @alice:test")
     ]
+
+
+# --- threads (L5): a work item's conversation is a thread --------------------
+
+
+def threaded_transport() -> tuple[MatrixTransport, list[dict], list[InboundMessage]]:
+    sent: list[dict] = []
+    ids = iter(f"$sent{n}" for n in range(1, 100))
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        sent.append(json.loads(request.content))
+        return httpx.Response(200, json={"event_id": next(ids)})
+
+    transport = MatrixTransport(CONFIG, http_transport=httpx.MockTransport(handler))
+    received: list[InboundMessage] = []
+    transport.on_message(received.append)
+    return transport, sent, received
+
+
+def event(event_id: str, body: str, relates: dict | None = None) -> dict:
+    content: dict = {"msgtype": "m.text", "body": body}
+    if relates is not None:
+        content["m.relates_to"] = relates
+    return {
+        "type": "m.room.message",
+        "sender": "@don:test",
+        "event_id": event_id,
+        "content": content,
+    }
+
+
+def deliver(transport: MatrixTransport, *events: dict) -> None:
+    transport._deliver(sync_body(list(events), "b"))
+
+
+def test_post_returns_the_event_id_and_threads_a_reply() -> None:
+    transport, sent, _ = threaded_transport()
+    root = transport.post("queued sandbox#14")
+    transport.post("As I understand it: ...", thread=root)
+
+    assert root == "$sent1"
+    assert "m.relates_to" not in sent[0]
+    assert sent[1]["m.relates_to"] == {
+        "rel_type": "m.thread",
+        "event_id": "$sent1",
+        # Clients without threads show it as a reply to the root.
+        "is_falling_back": True,
+        "m.in_reply_to": {"event_id": "$sent1"},
+    }
+
+
+def test_a_message_in_a_thread_carries_its_root() -> None:
+    transport, _, received = threaded_transport()
+    deliver(
+        transport,
+        event("$top", "status"),
+        event("$in", "go", {"rel_type": "m.thread", "event_id": "$root"}),
+    )
+    assert [(m.text, m.event_id, m.thread) for m in received] == [
+        ("status", "$top", None),
+        ("go", "$in", "$root"),
+    ]
+
+
+def test_a_plain_reply_to_a_threaded_message_joins_its_thread() -> None:
+    # Clients without thread support send an ordinary reply, quoting.
+    transport, _, received = threaded_transport()
+    root = transport.post("queued sandbox#14")
+    in_thread = transport.post("As I understand it: ...", thread=root)
+    deliver(
+        transport,
+        event(
+            "$r1",
+            "> <@nisse-dev:test> As I understand it: ...\n\nyes, go",
+            {"m.in_reply_to": {"event_id": in_thread}},
+        ),
+        event(
+            "$r2",
+            "> <@nisse-dev:test> queued sandbox#14\n\nthanks",
+            {"m.in_reply_to": {"event_id": root}},
+        ),
+    )
+    assert [(m.text, m.thread) for m in received] == [
+        ("yes, go", root),  # the quote is not what was said
+        ("thanks", root),  # a reply to the root joins its thread
+    ]
+
+
+def test_a_reply_to_an_unknown_message_stays_at_top_level() -> None:
+    transport, _, received = threaded_transport()
+    deliver(
+        transport,
+        event("$r", "> <@x:test> old\n\nhm", {"m.in_reply_to": {"event_id": "$old"}}),
+    )
+    assert [(m.text, m.thread) for m in received] == [("hm", None)]
+
+
+def test_a_reply_to_the_operators_own_threaded_message_joins_the_thread() -> None:
+    transport, _, received = threaded_transport()
+    deliver(
+        transport,
+        event("$t", "first", {"rel_type": "m.thread", "event_id": "$root"}),
+        event(
+            "$r",
+            "> <@don:test> first\n\nand also",
+            {"m.in_reply_to": {"event_id": "$t"}},
+        ),
+    )
+    assert [m.thread for m in received] == ["$root", "$root"]
+
+
+def test_in_memory_transport_plays_threads() -> None:
+    from nisse_dev.transport import InMemoryTransport
+
+    transport = InMemoryTransport()
+    seen: list[InboundMessage] = []
+    transport.on_message(seen.append)
+    root = transport.post("queued #14")
+    transport.post("restated", thread=root)
+    reply = transport.receive("@don:test", "go", thread=root)
+    assert transport.outbound[-1].thread == root
+    assert (seen[0].thread, seen[0].event_id) == (root, reply)

tests/integration/test_matrix_live.py

@@ -51,3 +51,25 @@ def test_send_lands_in_test_room(live_config: AgentConfig) -> None:
         assert marker in bodies
     finally:
         transport.close()
+
+
+def test_a_threaded_post_is_kept_in_its_thread(live_config: AgentConfig) -> None:
+    """The homeserver accepts the thread relation and lists the reply under
+    its root (L5: a work item's conversation is a thread)."""
+    config = live_config.matrix
+    marker = uuid.uuid4().hex
+    transport = MatrixTransport(config)
+    try:
+        _assert_testing_identity(transport, config)
+        root = transport.post(f"thread smoke test root {marker}")
+        reply = transport.post(f"thread smoke test reply {marker}", thread=root)
+        # The relations API is v1, not the transport's v3.
+        response = transport._http.get(
+            f"{config.homeserver}/_matrix/client/v1/rooms/{config.room_id}"
+            f"/relations/{root}/m.thread"
+        )
+        response.raise_for_status()
+        in_thread = [e["event_id"] for e in response.json().get("chunk", [])]
+        assert reply in in_thread
+    finally:
+        transport.close()