From 6dff36f458b2454f98b7c63f583087b9bd650b84 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Fri, 2 Oct 2026 23:40:24 +0000 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE --- nisse-dev/src/nisse_dev/matrix.py | 90 +++++++++++++++++-- nisse-dev/src/nisse_dev/transport.py | 44 ++++++++-- nisse-dev/tests/test_matrix.py | 122 ++++++++++++++++++++++++++ tests/integration/test_matrix_live.py | 22 +++++ 4 files changed, 266 insertions(+), 12 deletions(-) diff --git a/nisse-dev/src/nisse_dev/matrix.py b/nisse-dev/src/nisse_dev/matrix.py index 76ee235..43608c5 100644 --- a/nisse-dev/src/nisse_dev/matrix.py +++ b/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() diff --git a/nisse-dev/src/nisse_dev/transport.py b/nisse-dev/src/nisse_dev/transport.py index 48914ce..69d9475 100644 --- a/nisse-dev/src/nisse_dev/transport.py +++ b/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 diff --git a/nisse-dev/tests/test_matrix.py b/nisse-dev/tests/test_matrix.py index f127e59..54b2e85 100644 --- a/nisse-dev/tests/test_matrix.py +++ b/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) diff --git a/tests/integration/test_matrix_live.py b/tests/integration/test_matrix_live.py index 018cf55..a1d3216 100644 --- a/tests/integration/test_matrix_live.py +++ b/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()