diff --git a/docs/design-notes.md b/docs/design-notes.md index baf0d09..fce46bf 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 74c1534..cb60916 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 71f403c..aa99753 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 3601b28..a6b1e4f 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 1150750..75eec6a 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.""" @@ -1185,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, @@ -1211,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}, ) @@ -1552,48 +1608,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 8427f08..ae487a5 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 0000000..528beac --- /dev/null +++ b/tests/stacklets/test_archivist_corrections.py @@ -0,0 +1,543 @@ +"""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"]} + + +# ── 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" diff --git a/tests/stacklets/test_archivist_routing.py b/tests/stacklets/test_archivist_routing.py index 1301fcc..582c7d9 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 00cd041..f26eb62 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 ────────────────────────────