sandbox-quantum · amaudruz · Sep 6, 2026
@@ -40,7 +40,11 @@
 
 from switch_core.db.models import Message, MessageAttachment
 from switch_core.db.stores.message_store import MessageStore
-from switch_core.messages.recorded_types import MEMBERSHIP_EVENT_TYPE, should_record
+from switch_core.messages.recorded_types import (
+    MEMBERSHIP_EVENT_TYPE,
+    NOT_RECORDED_FILTER,
+    should_record,
+)
 from switch_core.transport import (
     InboundCustomEvent,
     InboundEvent,
@@ -177,7 +181,10 @@ async def backfill_room(
     start: str | None = None
     while report.pages_read < MAX_PAGES:
         page = await transport.read_history(
-            room.matrix_room_id, start=start, limit=PAGE_SIZE
+            room.matrix_room_id,
+            start=start,
+            limit=PAGE_SIZE,
+            exclude_types=NOT_RECORDED_FILTER,
         )
         report.pages_read += 1
 

@@ -316,8 +316,12 @@ async def _read_back_to(
     anchored = False
     done = False
     while not done:
+        # Unfiltered on purpose, unlike the backfill. This walk reports what
+        # it discarded (`ignored_by_type`), and that disclosure is part of the
+        # answer — a homeserver that withheld those events would make the
+        # report read as though the room had never carried any.
         page = await transport.read_history(
-            matrix_room_id, start=start, limit=PAGE_SIZE
+            matrix_room_id, start=start, limit=PAGE_SIZE, exclude_types=()
         )
         for raw in page.events:
             if not isinstance(raw, InboundEvent):

@@ -75,6 +75,21 @@
 NOT_RECORDED_PREFIXES = ("com.switch.observe.",)
 
 
+# The same denial expressed as a Matrix event filter, so a walk can ask the
+# homeserver not to send what it is only going to drop. Derived from the sets
+# above rather than restated, because the two going out of step would mean the
+# server silently withholding something the log wanted.
+#
+# A trailing `*` is the filter wildcard, which is what makes a prefix
+# expressible. Keep this a denial: an allowlist here would have the homeserver
+# skip a type nobody has classified yet, turning the deliberate "record what we
+# do not recognise" into "lose it", and the log would never know.
+NOT_RECORDED_FILTER = [
+    *sorted(NOT_RECORDED),
+    *(f"{prefix}*" for prefix in NOT_RECORDED_PREFIXES),
+]
+
+
 def should_record(event_type: str) -> bool:
     """Whether an event of this type belongs in the message log."""
     if event_type in NOT_RECORDED:

@@ -9,6 +9,7 @@
 
 import io
 import logging
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any
 
@@ -422,9 +423,25 @@ async def get_event(self, room_id: str, event_id: str) -> InboundEvent | None:
         return to_inbound(_RoomIdOnly(room_id), resp.event)
 
     async def read_history(
-        self, room_id: str, *, start: str | None, limit: int
+        self,
+        room_id: str,
+        *,
+        start: str | None,
+        limit: int,
+        exclude_types: Sequence[str],
     ) -> HistoryPage:
-        resp = await self.raw_client.room_messages(room_id, start=start, limit=limit)
+        # Filtered by the homeserver rather than after the fact. A room's
+        # history is mostly things the log does not keep, and `limit` counts
+        # what is sent — so filtering here makes each page carry a page's worth
+        # of what the caller actually wants instead of whatever survived.
+        resp = await self.raw_client.room_messages(
+            room_id,
+            start=start,
+            limit=limit,
+            message_filter={"not_types": list(exclude_types)}
+            if exclude_types
+            else None,
+        )
         if isinstance(resp, RoomMessagesError):
             raise TransportError(f"Failed to read history in {room_id}: {resp.message}")
         room = _RoomIdOnly(room_id)

@@ -12,7 +12,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass
 from typing import Any, Protocol, runtime_checkable
 
@@ -157,9 +157,22 @@ async def get_event(self, room_id: str, event_id: str) -> Any | None:
         ...
 
     async def read_history(
-        self, room_id: str, *, start: str | None, limit: int
+        self,
+        room_id: str,
+        *,
+        start: str | None,
+        limit: int,
+        exclude_types: Sequence[str],
     ) -> HistoryPage:
-        """Read one page of history backwards from `start`."""
+        """Read one page of history backwards from `start`.
+
+        `exclude_types` is a denial the transport pushes as far towards the
+        source as it can, so a caller that discards a category does not pay to
+        receive it first. Required rather than defaulted: a walk that filters
+        and a walk that counts what it drops want opposite answers, and neither
+        should get one by omission. A transport that cannot filter may ignore
+        it — the caller's own filtering is what decides correctness.
+        """
         ...
 
     # ── Rooms ─────────────────────────────────────────────────────────────────

@@ -37,6 +37,7 @@
 import asyncio
 import logging
 import uuid
+from collections.abc import Sequence
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
@@ -562,7 +563,12 @@ async def get_event(self, room_id: str, event_id: str) -> Any | None:
         return rows[0] if rows else None
 
     async def read_history(
-        self, room_id: str, *, start: str | None, limit: int
+        self,
+        room_id: str,
+        *,
+        start: str | None,
+        limit: int,
+        exclude_types: Sequence[str],
     ) -> HistoryPage:
         raise NotImplementedError(
             "PostgresTransport does not serve history: the read path already "

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import uuid
+from collections.abc import Sequence
 from datetime import UTC, datetime, timedelta
 
 from sqlalchemy import select
@@ -17,6 +18,12 @@
 from switch_core.db.models import Message, Room
 from switch_core.db.stores.message_store import MessageStore
 from switch_core.messages import backfill_room
+from switch_core.messages.recorded_types import (
+    MEMBERSHIP_EVENT_TYPE,
+    NOT_RECORDED,
+    NOT_RECORDED_FILTER,
+    should_record,
+)
 from switch_core.transport import (
     HistoryPage,
     InboundCustomEvent,
@@ -34,10 +41,17 @@ class PagingTransport:
     def __init__(self, pages: list[HistoryPage]) -> None:
         self._pages = pages
         self.reads = 0
+        self.excluded: list[Sequence[str]] = []
 
     async def read_history(
-        self, room_id: str, *, start: str | None, limit: int
+        self,
+        room_id: str,
+        *,
+        start: str | None,
+        limit: int,
+        exclude_types: Sequence[str] = (),
     ) -> HistoryPage:
+        self.excluded.append(exclude_types)
         page = self._pages[min(self.reads, len(self._pages) - 1)]
         self.reads += 1
         return page
@@ -170,6 +184,43 @@ async def test_the_timestamp_is_when_it_was_sent_not_when_it_was_written(
         assert rows[0].sent_at == _at(10)
 
 
+class TestWhatIsAskedFor:
+    async def test_the_walk_asks_the_bus_not_to_send_what_it_would_drop(
+        self, session_factory: async_sessionmaker[AsyncSession]
+    ) -> None:
+        """A room's history is mostly things the log does not keep, and `limit`
+        counts what is sent — so a page that arrives full of discards is a page
+        of nothing, paid for."""
+        async with session_factory() as session:
+            room = await _make_room(session)
+
+        transport = PagingTransport(
+            [HistoryPage(events=[_event("$a", 10, "a")], next_token=None)]
+        )
+        async with session_factory() as session:
+            room = await session.get(Room, room.id)  # type: ignore[assignment]
+            await backfill_room(transport, session_factory, room, store=MessageStore())
+
+        asked = list(transport.excluded[0])
+        assert "com.switch.agent.runtime_state" in asked
+        assert "com.switch.report.tool_call" in asked
+        assert "com.switch.task.delegate" in asked
+        # A prefix is expressible only as a wildcard.
+        assert "com.switch.observe.*" in asked
+        # Never the conversation, and never an arrival: the walk writes both.
+        assert "m.room.message" not in asked
+        assert MEMBERSHIP_EVENT_TYPE not in asked
+
+    async def test_the_denial_is_derived_from_the_denylist(self) -> None:
+        """Restating it would let the two drift, and drift here means the
+        homeserver withholding something the log wanted."""
+        for event_type in NOT_RECORDED:
+            assert event_type in NOT_RECORDED_FILTER
+        assert all(
+            not should_record(t) for t in NOT_RECORDED_FILTER if not t.endswith("*")
+        )
+
+
 class TestCompletionMark:
     """A room is marked done only by a walk that really wrote it.
 

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import uuid
+from collections.abc import Sequence
 from datetime import UTC, datetime, timedelta
 from itertools import count
 
@@ -35,7 +36,12 @@ def __init__(self, pages: list[HistoryPage]) -> None:
         self.reads = 0
 
     async def read_history(
-        self, room_id: str, *, start: str | None, limit: int
+        self,
+        room_id: str,
+        *,
+        start: str | None,
+        limit: int,
+        exclude_types: Sequence[str] = (),
     ) -> HistoryPage:
         page = self._pages[self.reads]
         self.reads += 1

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import Any
 
 from switch_core.attachments import ATTACHMENT_GROUP_KEY
@@ -227,7 +228,12 @@ async def get_event(self, room_id: str, event_id: str) -> InboundEvent | None:
         return self.events_by_id.get(event_id)
 
     async def read_history(
-        self, room_id: str, *, start: str | None, limit: int
+        self,
+        room_id: str,
+        *,
+        start: str | None,
+        limit: int,
+        exclude_types: Sequence[str] = (),
     ) -> HistoryPage:
         return self._history
 

@@ -165,6 +165,49 @@ async def test_send_event_carries_the_type_through() -> None:
     assert nio.room_send_calls[0][1] == "com.switch.task.accept"
 
 
+async def test_read_history_asks_the_homeserver_to_withhold_what_it_is_told_to() -> (
+    None
+):
+    """Filtered at the source, not after the fact.
+
+    `limit` counts what the homeserver sends, so a page filtered on arrival is
+    a page mostly wasted — the caller pays to receive and decode events it
+    already knows it will drop.
+    """
+    nio = _FakeNio()
+    seen: dict[str, Any] = {}
+
+    async def _chunk(*args: Any, **kwargs: Any) -> Any:
+        seen.update(kwargs)
+        return SimpleNamespace(chunk=[], end=None)
+
+    nio.room_messages = _chunk  # type: ignore[assignment]
+
+    await _transport(nio).read_history(
+        "!r:s", start=None, limit=10, exclude_types=("com.switch.report.tool_call",)
+    )
+
+    assert seen["message_filter"] == {"not_types": ["com.switch.report.tool_call"]}
+
+
+async def test_read_history_sends_no_filter_when_nothing_is_excluded() -> None:
+    """An empty denial is not a filter that denies nothing — it is no filter,
+    so a caller that wants everything is not relying on the server to agree
+    about what "everything" means."""
+    nio = _FakeNio()
+    seen: dict[str, Any] = {}
+
+    async def _chunk(*args: Any, **kwargs: Any) -> Any:
+        seen.update(kwargs)
+        return SimpleNamespace(chunk=[], end=None)
+
+    nio.room_messages = _chunk  # type: ignore[assignment]
+
+    await _transport(nio).read_history("!r:s", start=None, limit=10, exclude_types=())
+
+    assert seen["message_filter"] is None
+
+
 async def test_read_history_returns_neutral_events_and_its_cursor() -> None:
     nio = _FakeNio()
 
@@ -184,7 +227,9 @@ async def _chunk(*args: Any, **kwargs: Any) -> Any:
 
     nio.room_messages = _chunk  # type: ignore[assignment]
 
-    page = await _transport(nio).read_history("!r:s", start=None, limit=10)
+    page = await _transport(nio).read_history(
+        "!r:s", start=None, limit=10, exclude_types=("com.switch.report.tool_call",)
+    )
 
     assert page.next_token == "tok-2"
     assert [e.event_id for e in page.events] == ["$1"]
@@ -204,7 +249,9 @@ async def _fail(*args: Any, **kwargs: Any) -> Any:
     nio.room_messages = _fail  # type: ignore[assignment]
 
     with pytest.raises(TransportError, match="boom"):
-        await _transport(nio).read_history("!r:s", start=None, limit=10)
+        await _transport(nio).read_history(
+            "!r:s", start=None, limit=10, exclude_types=()
+        )
 
 
 def test_register_handlers_binds_only_what_was_supplied() -> None:

@@ -394,7 +394,9 @@ async def test_history_refuses_rather_than_answering_from_the_wrong_place(
         method are the walkers that compare a bus against them."""
         transport = _transport(session_factory, client_id="c", user_id="@a:test")
         with pytest.raises(NotImplementedError):
-            await transport.read_history("!r:test", start=None, limit=10)
+            await transport.read_history(
+                "!r:test", start=None, limit=10, exclude_types=()
+            )
 
 
 class TestReceiving: