From 4c7f99fdda4e26225d58e897be147d23164ccf77 Mon Sep 17 00:00:00 2001 From: Arthur Date: Sun, 2 Aug 2026 09:24:38 +0200 Subject: [PATCH 1/2] fix(docs): answering in a filing's thread corrects it, not searches The archivist files a document and replies in a thread. Answering in that thread is how you tell it that it got something wrong, but the message came back as search results instead. Replying to your own original upload did the same. The bot had learned to answer in a thread without learning to read one: it looked at a single reply pointer, and inside a thread that pointer is a rendering aid aimed at the newest message there, not at the filing. Anything posted after the filing, a todo link or a status line, broke the connection. It now asks the thread which filing it belongs to, so both ways of replying reach the classifier. Mentioning the bot by name is still a question, not a correction. Needs matrix-nio 0.25 for reading a thread's events. --- docs/design-notes.md | 20 + docs/user-guide.md | 2 + stacklets/core/bot-runner/microbot.py | 46 ++ stacklets/core/bot-runner/requirements.txt | 4 +- stacklets/docs/bot/archivist.py | 230 +++++---- tests/integration/test_archivist_e2e.py | 109 +++++ tests/stacklets/test_archivist_corrections.py | 449 ++++++++++++++++++ tests/stacklets/test_archivist_routing.py | 6 +- tests/stacklets/test_microbot.py | 97 ++++ 9 files changed, 858 insertions(+), 105 deletions(-) create mode 100644 tests/stacklets/test_archivist_corrections.py diff --git a/docs/design-notes.md b/docs/design-notes.md index baf0d095..fce46bf7 100644 --- a/docs/design-notes.md +++ b/docs/design-notes.md @@ -33,6 +33,26 @@ third tier (chat-triggered rebuild) is a card. bot-runner service concept (one consumer), bot-runner image reuse (the curator uses 2 of its 10 deps; slim image won). +## Both halves of a Matrix relation (thread corrections, 2026-08-02) + +The archivist learned to *answer* in a thread (`_answer`, `REPLY_IN_THREAD`) +but kept *reading* one `m.in_reply_to` hop, so a correction typed in a +filing's thread was routed to search. A client's reply pointer inside a +thread is a falling-back rendering aid at the newest event there, not the +message the human aimed at, so the pointer stopped naming our filing the +moment anything followed it (a todo link, a status line). + +The general shape, worth checking whenever a bot gains a new relation +type: **whoever teaches the sender a relation owns teaching the reader +the same one.** Half a relation is worse than none, because the send +side looks right in the client while routing silently degrades. + +Also why it survived a release: the intent spec that would have caught it +(`tests/integration/test_room_modes_e2e.py`) is marked +`xfail(strict=False)`, which is green when broken and green when fixed. +A non-strict xfail is not coverage. Making it `strict=True`, skipping it +with a reason, or deleting it are all better; the action is FAM-6. + ## Surviving upstream drift: the `wait_task` pattern When Paperless-ngx 3.0 redesigned its task API, the fix that held up was diff --git a/docs/user-guide.md b/docs/user-guide.md index 74c15340..cb60916a 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -153,6 +153,8 @@ Reply to the bot's filing confirmation with what's off: The archivist re-reads the document with your correction, re-files it, and confirms again. Corrections chain: if the second attempt is still off, reply to the new confirmation. Each reply carries the whole conversation back to the original document, so you never start over. +Anything you write in the filing's thread counts as a correction, so you can also just type in the thread the bot answered in, or reply to your own original message. You do not have to quote the confirmation. + This works for captures too: reply to a bookmark or note confirmation and it gets re-filed with your hint. --- diff --git a/stacklets/core/bot-runner/microbot.py b/stacklets/core/bot-runner/microbot.py index 71f403c4..aa997537 100644 --- a/stacklets/core/bot-runner/microbot.py +++ b/stacklets/core/bot-runner/microbot.py @@ -70,6 +70,7 @@ RoomMessagesResponse, SyncResponse, ) +from nio.api import RelationshipType from room_context import RoomContext, context_for @@ -786,6 +787,51 @@ async def _reply_parent_envelope(self, room_id: str, event) -> dict | None: envelope = parent.source.get("content", {}).get(self.FAMSTACK_EVENT_KEY) return envelope if isinstance(envelope, dict) else None + async def _thread_envelopes( + self, room_id: str, root_event_id: str, *, limit: int = 10, + ) -> list[tuple[str, dict]]: + """Our own famstack envelopes in the thread rooted at + ``root_event_id``, newest first. + + The threaded sibling of ``_reply_parent_envelope``. Someone + writing inside a thread is answering what the thread holds, and + their client's ``m.in_reply_to`` does not say which message that + is: for a threaded message the relation is a *falling back* + pointer at the newest event in the thread (Matrix v1.4), there + so thread-blind clients still render some context. Following it + lands on whatever we happened to post last. So we ask the + homeserver for the thread's children instead — Matrix is the + ledger, the bot keeps no thread bookkeeping — and hand back + every envelope we ourselves posted there. + + Newest first, because a bot's later message in a thread + supersedes its earlier one. Bounded to ``limit`` events + examined so one chat message can never turn into unbounded API + calls. Returns ``(event_id, envelope)`` pairs; as with the + single-hop sibling, the caller decides what an envelope + *means*. An id that roots no thread simply has none. + """ + envelopes: list[tuple[str, dict]] = [] + try: + examined = 0 + async for related in self._client.room_get_event_relations( + room_id, root_event_id, RelationshipType.thread, + ): + examined += 1 + if examined > limit: + break + if getattr(related, "sender", None) != self.user_id: + continue + content = (getattr(related, "source", None) or {}).get("content", {}) + envelope = content.get(self.FAMSTACK_EVENT_KEY) + event_id = getattr(related, "event_id", None) + if isinstance(envelope, dict) and event_id: + envelopes.append((event_id, envelope)) + except Exception as e: + logger.debug("[{}] thread relations fetch failed for {}: {}", + self.name, root_event_id, e) + return envelopes + def _ensure_http(self) -> aiohttp.ClientSession: """The shared aiohttp session, created on first use. diff --git a/stacklets/core/bot-runner/requirements.txt b/stacklets/core/bot-runner/requirements.txt index 3601b287..a6b1e4fa 100644 --- a/stacklets/core/bot-runner/requirements.txt +++ b/stacklets/core/bot-runner/requirements.txt @@ -1,4 +1,6 @@ -matrix-nio[e2e]>=0.24,<1.0 +# 0.25 is the floor: reading a thread's events (`room_get_event_relations`, +# how the bot finds the filing a correction belongs to) arrived there. +matrix-nio[e2e]>=0.25,<1.0 aiohttp>=3.9,<4.0 loguru>=0.7,<1.0 Pillow>=10.0,<12.0 diff --git a/stacklets/docs/bot/archivist.py b/stacklets/docs/bot/archivist.py index 11507509..7a915438 100644 --- a/stacklets/docs/bot/archivist.py +++ b/stacklets/docs/bot/archivist.py @@ -1016,47 +1016,113 @@ def _event_date(event) -> str | None: ts_ms / 1000, tz=_dt.timezone.utc, ).date().isoformat() - async def _reply_target_doc_id(self, room_id: str, event) -> int | None: - """Return the paperless_id when `event` replies to one of OUR filings. - - The framework's `_reply_parent_envelope` does the transport work - — fetch the replied-to parent, confirm it's ours, read off its - `dev.famstack.event` envelope. Here we keep only the archivist's - domain reading: any envelope that carries a `paperless_id` is a - valid correction target. That covers the initial `document.filed` - message AND a later `document.reclassified` -- the user can chain - corrections by replying to the most recent classification reply, - not just to the original filing. + @staticmethod + def _is_filing_envelope(envelope: dict) -> bool: + """Whether an envelope announces a filed item the user can correct. + + Any `*.filed` (the original) or `*.reclassified` (the result of + an earlier correction round), generic over filing kind so + documents and captures share one reading. Accepting the + reclassified form is what lets a user chain corrections: the + target of the next one is the most recent classification, not + just the original filing. """ + return envelope.get("type", "").endswith((".filed", ".reclassified")) + + async def _correction_anchor( + self, room_id: str, event, + ) -> tuple[str, dict] | None: + """The filing `event` is a correction to, as `(event_id, envelope)`. + + A correction is aimed at an *item*, not at a Matrix event, and + the family member's client decides how to express "about this + one". Three shapes reach the same filing, tried in the order + that respects the user's most explicit gesture first: + + 1. They quoted one of our classification messages: the reply + parent carries the envelope, so take it. Deliberate, and the + only shape available in a room without threads. + 2. They typed inside the filing's thread. The thread relation + is the deliberate part; the `m.in_reply_to` a client folds + in alongside it points at the newest event in the thread + (see `_thread_envelopes`), which may well be a todo line or + a status update with no envelope on it. + 3. They replied to the message that started it all -- their own + upload -- from the main timeline. That event carries no + envelope of ours, but it roots the thread our filing answered + in, so the thread hanging off it identifies the item. + + Returns None when nothing in reach is a filing of ours, which is + the common case: most messages are searches. + """ + parent_id = self._in_reply_to_id(event) envelope = await self._reply_parent_envelope(room_id, event) - if not envelope or envelope.get("type") not in ( - "document.filed", "document.reclassified", - ): + if envelope is not None and self._is_filing_envelope(envelope): + return (parent_id, envelope) + + roots = [r for r in (self.get_thread_root(event), parent_id) if r] + for root in dict.fromkeys(roots): + for event_id, candidate in await self._thread_envelopes(room_id, root): + if self._is_filing_envelope(candidate): + return (event_id, candidate) + return None + + @staticmethod + def _correction_target_doc_id(envelope: dict) -> int | None: + """The paperless_id a document filing envelope points at.""" + if envelope.get("type") not in ("document.filed", "document.reclassified"): return None paperless_id = envelope.get("data", {}).get("paperless_id") return paperless_id if isinstance(paperless_id, int) else None - async def _reply_target_capture_path( - self, room_id: str, event, - ) -> str | None: - """Return the capture's vault path when `event` replies to one - of OUR captures. - - Mirrors `_reply_target_doc_id` exactly -- accepts both the - initial `capture.filed` envelope and any later - `capture.reclassified` so chained corrections work the same - way they do for documents. - """ - envelope = await self._reply_parent_envelope(room_id, event) - if not envelope or envelope.get("type") not in ( - "capture.filed", "capture.reclassified", - ): + @staticmethod + def _correction_target_capture_path(envelope: dict) -> str | None: + """The vault path a capture filing envelope points at.""" + if envelope.get("type") not in ("capture.filed", "capture.reclassified"): return None vault_path = envelope.get("data", {}).get("vault_path") return vault_path if isinstance(vault_path, str) and vault_path else None + async def _handle_correction( + self, room_id: str, event, anchor: tuple[str, dict], reply_to: str, + ) -> bool: + """Re-run classification for the item `anchor` names, using the + user's message as an authoritative hint. Returns whether the + message was consumed as a correction. + + One entry point for both filing kinds: the envelope says whether + the item is a Paperless document or a vault capture, and the + correction chain is collected the same way for either. A message + that turns out to carry no correction (a bare quote with no words + of its own) is handed back to normal routing. + """ + anchor_id, envelope = anchor + hint, initial = await self._collect_correction_chain( + room_id, event, anchor_id, envelope, + ) + if not hint: + return False + + doc_id = self._correction_target_doc_id(envelope) + if doc_id is not None: + await self._handle_reply_reprocess( + room_id, doc_id, hint, reply_to, + date_filed=self._event_date(event), + initial_classification=initial, + ) + return True + + capture_path = self._correction_target_capture_path(envelope) + if capture_path is not None: + await self._handle_reply_capture_reprocess( + room_id, capture_path, hint, event.sender, reply_to, + initial_classification=initial, + ) + return True + return False + async def _collect_correction_chain( - self, room_id: str, event, + self, room_id: str, event, anchor_id: str, anchor_envelope: dict, ) -> tuple[str, dict | None]: """Walk the reply chain back to the original filing; return both the joined human-correction hint AND the latest classification @@ -1069,15 +1135,16 @@ async def _collect_correction_chain( that boundary the human's words belong to the upload's caption, not to a correction. - The latest envelope (the IMMEDIATE parent the user just - replied to) carries the post-correction-N classification under - ``data``. That's the state the human saw on screen when they - typed their note, so it's the right anchor for "apply this - correction as a delta": the LLM works against the same picture - the user saw, not against the LLM's untouched first pass. - Each step (state_N + correction_(N+1) → state_(N+1)) is - deterministic; chaining them gives a deterministic transform - from the initial filing to the current correction. + The anchor envelope (`_correction_anchor`'s answer: the latest + classification of this item the user could see) carries the + post-correction-N state under ``data``. That's the picture the + human had on screen when they typed their note, so it's the + right anchor for "apply this correction as a delta": the LLM + works against what the user saw, not against the LLM's + untouched first pass. Each step (state_N + correction_(N+1) → + state_(N+1)) is deterministic; chaining them gives a + deterministic transform from the initial filing to the current + correction. Returned hint: numbered list when more than one correction is present, plain string for a single correction. @@ -1087,21 +1154,9 @@ async def _collect_correction_chain( if current_body: bodies.append(current_body) - # The IMMEDIATE parent is the latest classification the user - # replied to. Grab its envelope BEFORE walking back so we can - # hand it to the prompt as the delta-anchor; the walker itself - # only needs the chain of human turns. - latest_state: dict | None = None - parent_id = self._in_reply_to_id(event) - immediate = ( - await self._fetch_event(room_id, parent_id) if parent_id else None - ) - if immediate is not None and getattr(immediate, "sender", None) == self.user_id: - envelope = immediate.source.get("content", {}).get(self.FAMSTACK_EVENT_KEY) - if isinstance(envelope, dict): - data = envelope.get("data") - if isinstance(data, dict): - latest_state = data + data = anchor_envelope.get("data") + latest_state: dict | None = data if isinstance(data, dict) else None + parent_id = anchor_id # Each loop iteration consumes one (bot, prior-user) pair from # the chain. Bounded by the framework's in_reply_to depth and a @@ -1146,12 +1201,6 @@ async def _collect_correction_chain( ) return (hint, latest_state) - async def _collect_correction_hint(self, room_id: str, event) -> str: - """Back-compat wrapper -- some callers (and older tests) just - want the hint string. Internally delegates to the chain walker.""" - hint, _initial = await self._collect_correction_chain(room_id, event) - return hint - @staticmethod def _in_reply_to_id(event) -> str | None: """The event_id this event replies to, or None if it isn't a reply.""" @@ -1552,48 +1601,27 @@ async def _on_text(self, room, event: RoomMessageText) -> None: query = "help" query_lower = query.lower() - # ── Reply-to-classification: user is correcting a prior filing ── - # When the user replies to a bot's filing message WITHOUT - # @-mentioning the bot, that's a correction: trace back the - # target from the parent event's envelope, re-run the - # classifier with the user's message as an authoritative - # hint. An @-mention is a different intent -- the user is - # addressing the bot conversationally (search, help), so we - # skip the reprocess path and let the dispatcher route on - # query content instead. Some clients (Element X) attach - # `m.in_reply_to` to mentioned messages; without this guard - # those searches would be eaten by the reprocess path. + # ── Correcting a prior filing ──────────────────────────────── + # A message that lands on one of our filings -- quoting it, or + # written in the thread it lives in -- WITHOUT @-mentioning the + # bot is a correction: `_correction_anchor` finds which item, + # and the classifier re-runs with the user's words as an + # authoritative hint. + # + # An @-mention is a different intent: the user is addressing + # the bot conversationally (search, help), and that stays true + # inside a filing thread, where "@archivist what else is from + # Duff?" is a question about the archive rather than a note on + # the document. Deliberate address beats ambient context, and + # it keeps the guard some clients need -- Element X attaches an + # `m.in_reply_to` to mentioned messages the user never aimed, + # and without this those searches would be eaten by reprocess. if not mentioned: - doc_id = await self._reply_target_doc_id(room.room_id, event) - if doc_id is not None: - hint, initial = await self._collect_correction_chain( - room.room_id, event, - ) - if hint: - await self._handle_reply_reprocess( - room.room_id, doc_id, hint, reply_to, - date_filed=self._event_date(event), - initial_classification=initial, - ) - return - - # Same shape for captures: a reply to a `capture.filed` or - # `capture.reclassified` confirmation reaches the capture - # pipeline's reprocess. The chain walker is generic over - # filing kind, so it Just Works for either side. - capture_path = await self._reply_target_capture_path( - room.room_id, event, - ) - if capture_path is not None: - hint, initial = await self._collect_correction_chain( - room.room_id, event, - ) - if hint: - await self._handle_reply_capture_reprocess( - room.room_id, capture_path, hint, event.sender, reply_to, - initial_classification=initial, - ) - return + anchor = await self._correction_anchor(room.room_id, event) + if anchor is not None and await self._handle_correction( + room.room_id, event, anchor, reply_to, + ): + return if query_lower in HELP_COMMANDS: # Same per-room welcome the bot posted on first encounter diff --git a/tests/integration/test_archivist_e2e.py b/tests/integration/test_archivist_e2e.py index 8427f082..ae487a58 100644 --- a/tests/integration/test_archivist_e2e.py +++ b/tests/integration/test_archivist_e2e.py @@ -308,6 +308,115 @@ async def test_homer_replies_to_filing_and_archivist_reprocesses( bdd.ok(f"reclassified #{paperless_id} with hint {hint!r}") +async def test_homer_corrects_a_filing_by_writing_in_its_thread( + bdd, openai, paperless, paperless_scope, homer, sample_invoice_pdf, +): + """Homer writes his correction in the filing's thread instead of + quoting the filing message, and the archivist still reprocesses. + + Scenario + -------- + Given the archivist filed Homer's invoice, answering in a thread + hung off his upload + When Homer types a correction in that thread, his client pointing + the reply fallback at his own upload (what Element sends: the + deliberate relation is the thread, the `m.in_reply_to` is a + rendering aid aimed at another event in it) + Then the archivist reprocesses the document rather than reading the + message as a search + + Why this needs the rig: the failure was a disagreement between the + bot's thread-aware sender and its reply-only reader, and only a real + homeserver both stores the relations and answers the relations query + the reader now makes. + """ + scope = paperless_scope + bdd.scenario("Homer corrects a filing from inside its thread") + + title = scope.tag("Duff Insurance - Kfz-Versicherung thread") + expected_correspondent = scope.tag("Duff Insurance") + bdd.given("the OpenAI mock will classify, reformat, then reclassify") + stub_classify(openai, { + "title": title, "topics": [scope.tag("Insurance")], "persons": ["Homer"], + "correspondent": expected_correspondent, "document_type": "Invoice", + "date": "2026-03-15", "summary": "Car insurance renewal.", + "facts": ["EUR 340.00/year"], "action_items": [], + }) + stub_reformat(openai, "# Kfz-Versicherung\n\nDuff Insurance.") + stub_classify(openai, { + "title": title, "topics": [scope.tag("Insurance")], "persons": ["Marge"], + "correspondent": expected_correspondent, "document_type": "Invoice", + "date": "2026-03-15", "summary": "Reclassified: this one is Marge's.", + "facts": ["EUR 340.00/year"], "action_items": [], + }) + + bdd.given("the #documents room exists and Homer has access") + room_id = await resolve_room(homer, DOCS_ROOM_ALIAS) + await ensure_joined(homer, room_id) + + # Front door: the document arrives as a family member sends it, so + # what gets corrected is a real filing with tags, correspondent, type + # and title. A document POSTed straight into Paperless has none of + # that, and a reclassification assertion against it means nothing. + bdd.when("Homer uploads the invoice and the archivist files it") + upload_event_id = await upload_and_send_file( + homer, room_id, sample_invoice_pdf, filename="invoice.pdf", + mime_type="application/pdf", msgtype="m.file", + ) + filing = await _wait_for_reply( + homer, room_id, + predicate=lambda e: ( + event_type(e) == "m.room.message" + and (_envelope(e) or {}).get("type") == "document.filed" + and (_envelope(e) or {}).get("data", {}).get("title") == title + ), + ) + assert filing, "archivist never posted a document.filed envelope" + paperless_id = _envelope(filing)["data"]["paperless_id"] + bdd.ok(f"filed doc #{paperless_id}, event {filing.event_id}") + + bdd.and_("the filing answered in a thread rooted at Homer's upload") + relation = filing.source.get("content", {}).get("m.relates_to", {}) + assert relation.get("rel_type") == "m.thread", \ + f"filing is not threaded, so this scenario cannot arise: {relation!r}" + assert relation.get("event_id") == upload_event_id, relation + bdd.detail(f"thread root = {upload_event_id}") + + bdd.when("Homer types a correction inside that thread") + hint = "this one is Marge's, not mine" + await homer.room_send( + room_id, "m.room.message", + { + "msgtype": "m.text", "body": hint, + # The shape a client sends for a message typed in a thread: + # the thread relation is what the user chose, and the reply + # fallback points at another event in the thread (here the + # upload) purely so thread-blind clients show context. + "m.relates_to": { + "rel_type": "m.thread", + "event_id": upload_event_id, + "is_falling_back": True, + "m.in_reply_to": {"event_id": upload_event_id}, + }, + }, + ) + + bdd.then("the archivist posts a document.reclassified confirmation") + reclassified = await _wait_for_reply( + homer, room_id, + predicate=lambda e: ( + (_envelope(e) or {}).get("type") == "document.reclassified" + and (_envelope(e) or {}).get("data", {}).get("paperless_id") + == paperless_id + ), + ) + assert reclassified, \ + "archivist never reclassified — an in-thread correction was read as a search" + env = _envelope(reclassified) + assert env["data"].get("user_hint") == hint, env + bdd.ok(f"reclassified #{paperless_id} with hint {hint!r}") + + # ── DM: reacts without a mention ────────────────────────────────────────── diff --git a/tests/stacklets/test_archivist_corrections.py b/tests/stacklets/test_archivist_corrections.py new file mode 100644 index 00000000..ceaf1caf --- /dev/null +++ b/tests/stacklets/test_archivist_corrections.py @@ -0,0 +1,449 @@ +"""Correcting a filing from chat: which messages mean "you got it wrong". + +The archivist answers a filing in a thread hung off the uploaded +message. When a family member then writes in that thread ("this is +Marge's, not Homer's"), they are correcting the filing, and the bot +must re-run classification with their words as an authoritative hint. +Anything else free-typed in the documents room is a search. + +Telling the two apart is a reading of the Matrix relation on the +incoming event, so these tests drive `_on_text` from the outside with +relation shapes taken from the **spec**, not from what our own sender +happens to produce: + + * a message typed inside a thread carries `rel_type: m.thread` with + the thread root, and clients that support rich replies also add a + *falling back* `m.in_reply_to` pointing at the newest event in the + thread, flagged `is_falling_back: true` (Matrix v1.4, threads). + That pointer is a rendering aid for thread-blind clients, not the + event the human chose; + * some clients send the thread relation without any `m.in_reply_to`; + * a plain rich reply carries only `m.in_reply_to`. + +A fixture built from our own `_send` output would only prove we agree +with ourselves. The bug these tests pin was exactly a disagreement +between our sender (thread-aware) and our reader (single reply hop). +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +_REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(_REPO_ROOT / "lib")) +sys.path.insert(0, str(_REPO_ROOT / "stacklets" / "core" / "bot-runner")) +sys.path.insert(0, str(_REPO_ROOT / "stacklets" / "docs" / "bot")) + +from archivist import ArchivistBot # noqa: E402 + + +BOT_ID = "@archivist-bot:server" +HOMER = "@homer:server" +ROOM_ID = "!docs:server" +DOC_ID = 42 + + +# ── Matrix wire shapes ─────────────────────────────────────────────────── +# +# Written from the spec rather than from `MicroBot._send`, so a change in +# how we *send* can never quietly make these agree with how we *read*. + + +def _thread_relation(root: str, *, falls_back_to: str | None = None) -> dict: + """`m.relates_to` for a message typed inside the thread at `root`. + + `falls_back_to` is the client's reply fallback: Element points it at + the newest event in the thread so thread-blind clients still show + the message in context, and marks it `is_falling_back`. Leaving it + None is the equally valid shape other clients send. + """ + relation = {"rel_type": "m.thread", "event_id": root} + if falls_back_to: + relation["is_falling_back"] = True + relation["m.in_reply_to"] = {"event_id": falls_back_to} + return {"m.relates_to": relation} + + +def _reply_relation(target: str) -> dict: + """`m.relates_to` for a rich reply to `target`, no thread involved.""" + return {"m.relates_to": {"m.in_reply_to": {"event_id": target}}} + + +def _message(event_id, sender, body, *, content=None, envelope=None): + """A minimal `m.room.message` as the bot receives it from nio.""" + payload = {"msgtype": "m.text", "body": body, **(content or {})} + if envelope is not None: + payload["dev.famstack.event"] = envelope + return SimpleNamespace( + event_id=event_id, sender=sender, body=body, + server_timestamp=1_700_000_000_000, + source={"content": payload}, + ) + + +def _filed(paperless_id=DOC_ID, **data): + return { + "source": "docs", "type": "document.filed", + "data": {"paperless_id": paperless_id, **data}, + } + + +def _reclassified(paperless_id=DOC_ID, **data): + return { + "source": "docs", "type": "document.reclassified", + "data": {"paperless_id": paperless_id, **data}, + } + + +# ── Homeserver stand-in ────────────────────────────────────────────────── + + +class FakeMatrix: + """The slice of `nio.AsyncClient` the correction path reads. + + Holds a room's events and, per thread root, the ids of the events + hanging off it. `room_get_event_relations` is an async iterator that + yields newest-first, which is what Synapse returns for the default + backwards direction. + """ + + def __init__(self): + self.events: dict[str, object] = {} + self.threads: dict[str, list[str]] = {} + self.relation_calls: list[str] = [] + + def add(self, event, *, thread_root: str | None = None): + self.events[event.event_id] = event + if thread_root: + self.threads.setdefault(thread_root, []).append(event.event_id) + return event + + async def room_get_event(self, room_id, event_id): + return SimpleNamespace(event=self.events.get(event_id)) + + async def room_get_event_relations( + self, room_id, event_id, rel_type=None, **kwargs, + ): + self.relation_calls.append(event_id) + for child_id in reversed(self.threads.get(event_id, [])): + yield self.events[child_id] + + +# ── Bot under test ─────────────────────────────────────────────────────── + + +@pytest.fixture +def bot(tmp_path): + """An archivist wired to a fake homeserver, recording where each + message was routed. Handlers are replaced with recorders: the + dispatch decision is what these tests pin, and the handlers have + their own coverage.""" + bot = ArchivistBot( + homeserver="http://homeserver", user_id=BOT_ID, password="x", + session_dir=tmp_path, + ) + bot._client = FakeMatrix() + bot.routed: list[tuple] = [] + + async def _search(room_id, query, reply_to=None, *, sender=None): + bot.routed.append(("search", query)) + + async def _reprocess(room_id, doc_id, user_hint, reply_to, *, + date_filed=None, initial_classification=None): + bot.routed.append(("reprocess", doc_id, user_hint, initial_classification)) + + async def _capture_reprocess(room_id, vault_path, user_hint, sender_mxid, + reply_to, *, initial_classification=None): + bot.routed.append(("capture_reprocess", vault_path, user_hint)) + + async def _text_capture(room_id, text, sender, reply_to=None, *, capture_id=None): + bot.routed.append(("capture_text", text)) + + async def _send(room_id, text, *a, **kw): + bot.routed.append(("send", text)) + + async def _noop(*a, **kw): + return None + + bot._handle_search = _search + bot._handle_reply_reprocess = _reprocess + bot._handle_reply_capture_reprocess = _capture_reprocess + bot._handle_text_capture = _text_capture + bot._send = _send + # The per-room welcome and the room-mode read run ahead of routing + # and are covered elsewhere; keep the recorded list to routing only. + bot._send_room_welcome_if_needed = _noop + bot._room_mode_allows_react = lambda _ctx: _true_coro() + return bot + + +async def _true_coro(): + return True + + +def _docs_room(): + return SimpleNamespace( + room_id=ROOM_ID, + canonical_alias="#documents:server", + name=None, + users={uid: object() for uid in (BOT_ID, HOMER, "@marge:server")}, + ) + + +@pytest.fixture +def filed_thread(bot): + """The room after a filing: Homer's upload, the archivist's threaded + confirmation carrying the `document.filed` envelope, and a later bot + message in the same thread that carries no envelope. + + That trailing message is the reported trigger: it is what a client's + reply fallback points at, so a reader that only follows one reply hop + lands on a message with nothing to correct. + """ + client = bot._client + client.add(_message("$upload", HOMER, "invoice.pdf")) + client.add( + _message("$filed", BOT_ID, "Filed: Duff Insurance invoice (#42)", + content=_thread_relation("$upload"), + envelope=_filed(topics=["Insurance"], persons=["Homer"])), + thread_root="$upload", + ) + client.add( + _message("$todo", BOT_ID, "Added 1 todo: pay by 2026-03-15", + content=_thread_relation("$upload", falls_back_to="$filed")), + thread_root="$upload", + ) + return client + + +# ── Corrections typed inside the filing thread ─────────────────────────── + + +class TestCorrectionInsideAThread: + """A message in the filing's thread is about that filing. The thread + is the relation the user's client states deliberately; the reply + fallback inside it is not.""" + + @pytest.mark.asyncio + async def test_reply_fallback_pointing_at_a_later_message_still_corrects( + self, bot, filed_thread, + ): + """The reported bug. Homer types in the thread; Element attaches a + falling-back `m.in_reply_to` to the newest event there, which is + the bot's todo line, not the filing. Following that one hop finds + no envelope. The thread does.""" + event = _message( + "$correction", HOMER, "this is Marge's, not Homer's", + content=_thread_relation("$upload", falls_back_to="$todo"), + ) + await bot._on_text(_docs_room(), event) + assert bot.routed == [ + ("reprocess", DOC_ID, "this is Marge's, not Homer's", + {"paperless_id": DOC_ID, "topics": ["Insurance"], "persons": ["Homer"]}), + ] + + @pytest.mark.asyncio + async def test_thread_message_without_a_reply_relation_corrects( + self, bot, filed_thread, + ): + """Clients may send the thread relation with no `m.in_reply_to` at + all. There is no reply hop to follow, so a reader built on one + cannot see this message; the thread relation is enough.""" + event = _message( + "$correction", HOMER, "wrong year, it is 2025", + content=_thread_relation("$upload"), + ) + await bot._on_text(_docs_room(), event) + assert [r[0] for r in bot.routed] == ["reprocess"] + assert bot.routed[0][2] == "wrong year, it is 2025" + + @pytest.mark.asyncio + async def test_reply_to_the_users_own_upload_corrects(self, bot, filed_thread): + """Replying to the message that started it all is the obvious + gesture, and the one the reporter used. The upload is Homer's own + event and carries no envelope, so only the thread hanging off it + identifies the document.""" + event = _message( + "$correction", HOMER, "the correspondent is Globex", + content=_thread_relation("$upload", falls_back_to="$upload"), + ) + await bot._on_text(_docs_room(), event) + assert [r[0] for r in bot.routed] == ["reprocess"] + + @pytest.mark.asyncio + async def test_plain_reply_to_the_upload_corrects(self, bot, filed_thread): + """Same gesture from the main timeline: a client that replies to + the upload without joining the thread sends only `m.in_reply_to`. + The replied-to event is the root of the filing's thread, so the + filing is still reachable.""" + event = _message( + "$correction", HOMER, "the correspondent is Globex", + content=_reply_relation("$upload"), + ) + await bot._on_text(_docs_room(), event) + assert [r[0] for r in bot.routed] == ["reprocess"] + + @pytest.mark.asyncio + async def test_latest_classification_in_the_thread_wins(self, bot, filed_thread): + """A correction applies to the state the user is looking at. When + the thread already holds a reclassification, that is the anchor + handed to the pipeline, not the original filing.""" + filed_thread.add( + _message("$reclass", BOT_ID, "Reclassified (#42)", + content=_thread_relation("$upload", falls_back_to="$todo"), + envelope=_reclassified(persons=["Marge"])), + thread_root="$upload", + ) + filed_thread.add( + _message("$todo2", BOT_ID, "Todo updated", + content=_thread_relation("$upload", falls_back_to="$reclass")), + thread_root="$upload", + ) + event = _message( + "$correction", HOMER, "and the type is Contract", + content=_thread_relation("$upload", falls_back_to="$todo2"), + ) + await bot._on_text(_docs_room(), event) + assert bot.routed[0][0] == "reprocess" + assert bot.routed[0][3] == {"paperless_id": DOC_ID, "persons": ["Marge"]} + + @pytest.mark.asyncio + async def test_capture_thread_reaches_the_capture_pipeline(self, bot): + """Captures thread the same way and correct the same way; the + archivist reads the envelope kind, not the room.""" + client = bot._client + client.add(_message("$note", HOMER, "a long pasted note")) + client.add( + _message("$capfiled", BOT_ID, "Saved as a note", + content=_thread_relation("$note"), + envelope={"source": "docs", "type": "capture.filed", + "data": {"vault_path": "homer/notes/x.md"}}), + thread_root="$note", + ) + event = _message( + "$correction", HOMER, "file this under school, not work", + content=_thread_relation("$note"), + ) + await bot._on_text(_docs_room(), event) + assert bot.routed == [ + ("capture_reprocess", "homer/notes/x.md", + "file this under school, not work"), + ] + + +# ── What must keep working ─────────────────────────────────────────────── + + +class TestNotACorrection: + """The other direction of the same decision. Reading the thread must + not turn ordinary messages into corrections.""" + + @pytest.mark.asyncio + async def test_plain_message_in_the_documents_room_is_a_search( + self, bot, filed_thread, + ): + """No relation at all: the documents room's default is recall.""" + event = _message("$q", HOMER, "Duff Insurance") + await bot._on_text(_docs_room(), event) + assert bot.routed == [("search", "Duff Insurance")] + + @pytest.mark.asyncio + async def test_thread_without_a_filing_is_a_search(self, bot): + """A thread hung off an ordinary conversation has nothing to + correct, so its messages route normally.""" + client = bot._client + client.add(_message("$chat", HOMER, "did we ever insure the car?")) + client.add( + _message("$answer", BOT_ID, "I found 3 documents", + content=_thread_relation("$chat")), + thread_root="$chat", + ) + event = _message("$q", HOMER, "what about the boat", + content=_thread_relation("$chat")) + await bot._on_text(_docs_room(), event) + assert bot.routed == [("search", "what about the boat")] + + @pytest.mark.asyncio + async def test_mention_inside_a_filing_thread_is_a_search( + self, bot, filed_thread, + ): + """An @-mention is the user addressing the bot on purpose, and it + outranks the ambient thread: inside a filing thread, "@archivist + what else is from Duff?" is a question, not a correction. + + This also keeps the existing guard honest. Element X attaches an + `m.in_reply_to` to mentioned messages the user never aimed, so + mention-means-conversation is what stops ordinary searches being + swallowed by the reprocess path. + """ + event = _message( + "$q", HOMER, f"{BOT_ID} what else is from Duff Insurance", + content={ + "m.mentions": {"user_ids": [BOT_ID]}, + **_thread_relation("$upload", falls_back_to="$todo"), + }, + ) + await bot._on_text(_docs_room(), event) + assert bot.routed == [("search", "what else is from Duff Insurance")] + + @pytest.mark.asyncio + async def test_reply_to_another_users_message_is_not_a_correction(self, bot): + """Only our own filing messages are correction targets. A reply to + a family member's message must not reach the pipeline even when + that message carries a look-alike envelope.""" + client = bot._client + client.add( + _message("$spoof", "@bart:server", "Filed: homework (#99)", + envelope=_filed(99)), + ) + event = _message("$q", HOMER, "nice try", + content=_reply_relation("$spoof")) + await bot._on_text(_docs_room(), event) + assert bot.routed == [("search", "nice try")] + + +class TestChainedCorrections: + """Correcting a correction. Each round adds a user turn and a bot + confirmation; the pipeline gets every human turn back to the original + filing, plus the classification the user was looking at.""" + + @pytest.mark.asyncio + async def test_direct_reply_to_the_filing_still_corrects(self, bot, filed_thread): + """The single reply hop is still the path for a client that quotes + the filing message itself.""" + event = _message("$c1", HOMER, "this is Marge's", + content=_reply_relation("$filed")) + await bot._on_text(_docs_room(), event) + assert bot.routed == [ + ("reprocess", DOC_ID, "this is Marge's", + {"paperless_id": DOC_ID, "topics": ["Insurance"], "persons": ["Homer"]}), + ] + + @pytest.mark.asyncio + async def test_reply_to_the_latest_reclassification_folds_earlier_turns( + self, bot, filed_thread, + ): + """Round two. Homer replies to the confirmation of round one; the + hint the pipeline sees carries both of his turns, most recent + first, and the anchor is round one's classification.""" + client = bot._client + client.add(_message("$c1", HOMER, "this is Marge's", + content=_reply_relation("$filed"))) + client.add(_message("$reclass", BOT_ID, "Reclassified (#42)", + content=_reply_relation("$c1"), + envelope=_reclassified(persons=["Marge"]))) + + event = _message("$c2", HOMER, "and it is a contract", + content=_reply_relation("$reclass")) + await bot._on_text(_docs_room(), event) + + kind, doc_id, hint, initial = bot.routed[0] + assert (kind, doc_id) == ("reprocess", DOC_ID) + assert "and it is a contract" in hint + assert "this is Marge's" in hint + assert hint.index("and it is a contract") < hint.index("this is Marge's") + assert initial == {"paperless_id": DOC_ID, "persons": ["Marge"]} diff --git a/tests/stacklets/test_archivist_routing.py b/tests/stacklets/test_archivist_routing.py index 1301fcc1..582c7d9b 100644 --- a/tests/stacklets/test_archivist_routing.py +++ b/tests/stacklets/test_archivist_routing.py @@ -597,9 +597,9 @@ async def _record_send(*a, **kw): bot._handle_text_capture = _record_text_capture bot._handle_url = _record_url bot._send = _record_send - # Reply-to lookup needs the client; short-circuit it. - bot._reply_target_doc_id = lambda *_a, **_kw: _none_coro() - bot._reply_target_capture_path = lambda *_a, **_kw: _none_coro() + # Correction lookup needs the client; short-circuit it. What it + # resolves is pinned in test_archivist_corrections.py. + bot._correction_anchor = lambda *_a, **_kw: _none_coro() # The per-room welcome path runs ahead of routing decisions in # `_on_text` / `_on_file`. These tests focus on the routing # dispatch, not the welcome -- stub it out so the recorded diff --git a/tests/stacklets/test_microbot.py b/tests/stacklets/test_microbot.py index 00cd041a..f26eb624 100644 --- a/tests/stacklets/test_microbot.py +++ b/tests/stacklets/test_microbot.py @@ -44,6 +44,11 @@ def __init__(self): # key yields a response whose `.event` is None (parent not found). self.parent_events: dict[str, object] = {} self.get_event_raises: BaseException | None = None + # Thread children keyed by root event id, in timeline order. The + # relations endpoint hands them back newest-first (the default + # backwards direction), so the stub reverses on the way out. + self.thread_children: dict[str, list] = {} + self.relations_raise: BaseException | None = None # Drain surface: the timeline (newest-first), the rooms map, and a # sync token. room_messages returns the whole timeline in one page. self.next_batch = "END" @@ -80,6 +85,13 @@ async def room_get_event(self, room_id, event_id): raise self.get_event_raises return SimpleNamespace(event=self.parent_events.get(event_id)) + async def room_get_event_relations(self, room_id, event_id, rel_type=None, + **kwargs): + if self.relations_raise is not None: + raise self.relations_raise + for event in reversed(self.thread_children.get(event_id, [])): + yield event + def _build_bot(tmp_path, *, handler) -> tuple[MicroBot, FakeClient]: """Construct a minimal MicroBot subclass with a recording client. @@ -680,6 +692,91 @@ async def test_none_when_fetch_raises(self, tmp_path): assert await bot._reply_parent_envelope("!r:server", self._reply_to("$x")) is None +# ── Thread envelopes ─────────────────────────────────────────────────────── + + +class TestThreadEnvelopes: + """`_thread_envelopes` is the threaded half of the same job: given a + thread root, hand back every `dev.famstack.event` the bot itself + posted in that thread, newest first. + + It exists because a message typed inside a thread does not say which + event it answers -- the `m.in_reply_to` a client folds in alongside + the thread relation is a fallback pointer at the newest event there. + Reading the thread from the homeserver is what makes the bot's own + filing findable regardless of what followed it.""" + + @staticmethod + def _bot_message(event_id, envelope=None, sender="@test-bot:server"): + content = {"msgtype": "m.text", "body": "…"} + if envelope is not None: + content["dev.famstack.event"] = envelope + return SimpleNamespace( + event_id=event_id, sender=sender, source={"content": content}, + ) + + @pytest.mark.asyncio + async def test_returns_our_envelopes_newest_first(self, tmp_path): + bot, client = _bare_bot(tmp_path) + client.thread_children["$root"] = [ + self._bot_message("$a", {"type": "document.filed"}), + self._bot_message("$b", {"type": "document.reclassified"}), + ] + got = await bot._thread_envelopes("!r:server", "$root") + assert got == [ + ("$b", {"type": "document.reclassified"}), + ("$a", {"type": "document.filed"}), + ] + + @pytest.mark.asyncio + async def test_skips_messages_we_did_not_send(self, tmp_path): + # Same ownership rule as the single-hop sibling: another user's + # message carrying a look-alike envelope is not ours to act on. + bot, client = _bare_bot(tmp_path) + client.thread_children["$root"] = [ + self._bot_message("$a", {"type": "document.filed"}, sender="@homer:server"), + ] + assert await bot._thread_envelopes("!r:server", "$root") == [] + + @pytest.mark.asyncio + async def test_skips_our_messages_without_an_envelope(self, tmp_path): + # The reason the single reply hop was not enough: plain bot + # messages (a todo line, a status update) share the thread. + bot, client = _bare_bot(tmp_path) + client.thread_children["$root"] = [ + self._bot_message("$a", {"type": "document.filed"}), + self._bot_message("$b"), + ] + assert await bot._thread_envelopes("!r:server", "$root") == [ + ("$a", {"type": "document.filed"}), + ] + + @pytest.mark.asyncio + async def test_empty_when_the_event_roots_no_thread(self, tmp_path): + bot, _ = _bare_bot(tmp_path) + assert await bot._thread_envelopes("!r:server", "$loner") == [] + + @pytest.mark.asyncio + async def test_bounded_by_events_examined(self, tmp_path): + """A long thread must not turn one chat message into unbounded + paging, so the scan stops after `limit` events even if the + envelope it wants sits further back.""" + bot, client = _bare_bot(tmp_path) + client.thread_children["$root"] = [ + self._bot_message("$old", {"type": "document.filed"}), + *[self._bot_message(f"$chat{i}") for i in range(5)], + ] + assert await bot._thread_envelopes("!r:server", "$root", limit=3) == [] + + @pytest.mark.asyncio + async def test_empty_when_the_fetch_fails(self, tmp_path): + # Transport trouble reads as "no envelopes", never an exception + # out of the routing path. + bot, client = _bare_bot(tmp_path) + client.relations_raise = ConnectionError("synapse down") + assert await bot._thread_envelopes("!r:server", "$root") == [] + + # ── Per-room config + emoji + !config command ──────────────────────────── From 4ddb578800c4e42a6bf2d86d9c7239a4810ae004 Mon Sep 17 00:00:00 2001 From: Arthur Date: Sun, 2 Aug 2026 09:49:02 +0200 Subject: [PATCH 2/2] fix(docs): answer a correction inside the thread it came from The reply confirming a re-file went to the main timeline instead of the thread the correction was typed in. On its own that is just noise, but corrections chain by asking the thread what the newest classification is, so a confirmation posted outside it left every later correction working from the original filing again. Fixing one thing then fixing a second undid the first. The capture side already answered in the thread; only documents did not. The tests for reading a thread were handed one that already contained a re-filed confirmation, so they would have agreed with a bot that never posts one there. They now also check the bot puts it there. --- stacklets/docs/bot/archivist.py | 13 ++- tests/stacklets/test_archivist_corrections.py | 94 +++++++++++++++++++ 2 files changed, 104 insertions(+), 3 deletions(-) diff --git a/stacklets/docs/bot/archivist.py b/stacklets/docs/bot/archivist.py index 7a915438..75eec6a0 100644 --- a/stacklets/docs/bot/archivist.py +++ b/stacklets/docs/bot/archivist.py @@ -1234,18 +1234,25 @@ async def _handle_reply_reprocess( maps it to a chat reply. The reclassified confirmation carries a fresh `document.reclassified` envelope so the user can chain another correction by replying to it. + + Answers go back through `_answer`, so they land in the same + thread the correction came from. That placement is load-bearing + rather than tidiness: the next correction resolves its anchor by + asking the thread for the newest classification, so a + confirmation posted outside the thread would leave every later + correction anchored to the original filing. """ o = await self._pipeline.reprocess( doc_id=doc_id, user_hint=user_hint, date_filed=date_filed, initial_classification=initial_classification, ) if o.status == "doc_missing": - await self._send( + await self._answer( room_id, self.t("reprocess_doc_missing", doc_id=doc_id), reply_to, ) elif o.status == "llm_error": kind, detail = o.llm_error - await self._send( + await self._answer( room_id, self.t("reprocess_llm_error", doc_id=doc_id, kind=kind, detail=detail), reply_to, @@ -1260,7 +1267,7 @@ async def _handle_reply_reprocess( resolved_type=o.resolved_type, resolved_correspondent=o.resolved_correspondent, ) - await self._send( + await self._answer( room_id, reply, reply_to, metadata={"dev.famstack.event": o.envelope}, ) diff --git a/tests/stacklets/test_archivist_corrections.py b/tests/stacklets/test_archivist_corrections.py index ceaf1caf..528beac1 100644 --- a/tests/stacklets/test_archivist_corrections.py +++ b/tests/stacklets/test_archivist_corrections.py @@ -447,3 +447,97 @@ async def test_reply_to_the_latest_reclassification_folds_earlier_turns( assert "this is Marge's" in hint assert hint.index("and it is a contract") < hint.index("this is Marge's") assert initial == {"paperless_id": DOC_ID, "persons": ["Marge"]} + + +# ── The other half of the relation: where the answer lands ─────────────── + + +class TestTheAnswerStaysInTheThread: + """A correction is answered inside the thread it came from. + + The reader tests above are handed a thread that already contains a + `document.reclassified` message. This class checks the bot actually + produces one there. Without it those tests would agree with a bot + that answers outside the thread: the fixture would supply the + placement the implementation never creates, and chained corrections + would silently anchor to the original filing forever. + """ + + @pytest.fixture + def bot(self, tmp_path): + """An archivist with the REAL reprocess handler and a stub + pipeline, recording every send with its relation.""" + bot = ArchivistBot( + homeserver="http://homeserver", user_id=BOT_ID, password="x", + session_dir=tmp_path, + ) + bot._client = FakeMatrix() + bot.sent: list[dict] = [] + + async def _room_send(room_id, message_type, content, **kw): + bot.sent.append(content) + return SimpleNamespace(event_id="$answer") + + bot._client.room_send = _room_send + bot._pipeline = SimpleNamespace() + return bot + + def _outcome(self, status="reclassified"): + return SimpleNamespace( + status=status, doc_id=DOC_ID, llm_error=("timeout", "took too long"), + title="Auto Insurance Policy 2026", resolved_topics=["insurance"], + resolved_persons=["Marge"], resolved_type="contract", + resolved_correspondent="Duff Insurance", + envelope=_reclassified(persons=["Marge"]), + ) + + @pytest.mark.asyncio + async def test_reclassified_confirmation_joins_the_correction_thread(self, bot): + """The confirmation carries an `m.thread` relation rooted where + the correction was, so the next correction can find it.""" + bot._client.add(_message("$upload", HOMER, "policy.pdf")) + bot._client.add( + _message("$correction", HOMER, "this is Marge's", + content=_thread_relation("$upload", falls_back_to="$filed")), + thread_root="$upload", + ) + + async def _reprocess(**kw): + return self._outcome() + + bot._pipeline.reprocess = _reprocess + await bot._handle_reply_reprocess( + "!docs:server", DOC_ID, "this is Marge's", "$correction", + ) + + assert bot.sent, "the reprocess produced no message at all" + relation = bot.sent[-1].get("m.relates_to", {}) + assert relation.get("rel_type") == "m.thread", ( + "the confirmation was posted outside the thread, so the next " + "correction would anchor to the original filing" + ) + assert relation.get("event_id") == "$upload" + + @pytest.mark.asyncio + async def test_a_failed_reprocess_answers_in_the_thread_too(self, bot): + """An error is a reply to what the user just typed, so it belongs + where they typed it. Otherwise a correction that failed looks + like nothing happened.""" + bot._client.add(_message("$upload", HOMER, "policy.pdf")) + bot._client.add( + _message("$correction", HOMER, "this is Marge's", + content=_thread_relation("$upload", falls_back_to="$filed")), + thread_root="$upload", + ) + + async def _reprocess(**kw): + return self._outcome(status="llm_error") + + bot._pipeline.reprocess = _reprocess + await bot._handle_reply_reprocess( + "!docs:server", DOC_ID, "this is Marge's", "$correction", + ) + + relation = bot.sent[-1].get("m.relates_to", {}) + assert relation.get("rel_type") == "m.thread" + assert relation.get("event_id") == "$upload"