Matrix threads in the transport (M2.1, PR A)

Transport.post(text, thread=...) returns the event ID, which can root a
thread; a threaded post carries m.thread with the reply fallback for
clients without threads. Inbound messages carry their event ID and thread
root: a message in a thread gives its root, a plain reply joins the
thread of the message it replies to when known, and the quoted lines of a
rich reply are no longer part of the text. The in-memory transport plays
threads for tests.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

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}",
-                json={"msgtype": "m.text", "body": text},
+                f"/rooms/{room}/send/m.room.message/{txn}", json=content
             )
         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