diff --git a/core/switch_core/messages/backfill.py b/core/switch_core/messages/backfill.py index 099d0cc2..e2e7d8f5 100644 --- a/core/switch_core/messages/backfill.py +++ b/core/switch_core/messages/backfill.py @@ -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 diff --git a/core/switch_core/messages/reconcile.py b/core/switch_core/messages/reconcile.py index cf1af75f..ac2c2188 100644 --- a/core/switch_core/messages/reconcile.py +++ b/core/switch_core/messages/reconcile.py @@ -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): diff --git a/core/switch_core/messages/recorded_types.py b/core/switch_core/messages/recorded_types.py index 0ab7bffd..27d9137b 100644 --- a/core/switch_core/messages/recorded_types.py +++ b/core/switch_core/messages/recorded_types.py @@ -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: diff --git a/core/switch_core/transport/matrix.py b/core/switch_core/transport/matrix.py index 08267222..8aa3092c 100644 --- a/core/switch_core/transport/matrix.py +++ b/core/switch_core/transport/matrix.py @@ -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) diff --git a/core/switch_core/transport/port.py b/core/switch_core/transport/port.py index cd4e866b..6c8997eb 100644 --- a/core/switch_core/transport/port.py +++ b/core/switch_core/transport/port.py @@ -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 ───────────────────────────────────────────────────────────────── diff --git a/core/switch_core/transport/postgres.py b/core/switch_core/transport/postgres.py index 10313f6c..5557a83b 100644 --- a/core/switch_core/transport/postgres.py +++ b/core/switch_core/transport/postgres.py @@ -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 " diff --git a/core/tests/switch_core/messages/test_backfill.py b/core/tests/switch_core/messages/test_backfill.py index c74e087a..f8f1df9c 100644 --- a/core/tests/switch_core/messages/test_backfill.py +++ b/core/tests/switch_core/messages/test_backfill.py @@ -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. diff --git a/core/tests/switch_core/messages/test_reconcile.py b/core/tests/switch_core/messages/test_reconcile.py index a93a8c46..6d653d20 100644 --- a/core/tests/switch_core/messages/test_reconcile.py +++ b/core/tests/switch_core/messages/test_reconcile.py @@ -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 diff --git a/core/tests/switch_core/transport/fake.py b/core/tests/switch_core/transport/fake.py index 760699a7..3ddfc3bb 100644 --- a/core/tests/switch_core/transport/fake.py +++ b/core/tests/switch_core/transport/fake.py @@ -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 diff --git a/core/tests/switch_core/transport/test_matrix_transport.py b/core/tests/switch_core/transport/test_matrix_transport.py index 01e0bb6e..1182ffb5 100644 --- a/core/tests/switch_core/transport/test_matrix_transport.py +++ b/core/tests/switch_core/transport/test_matrix_transport.py @@ -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: diff --git a/core/tests/switch_core/transport/test_postgres_transport.py b/core/tests/switch_core/transport/test_postgres_transport.py index f05de0c1..4b12ca63 100644 --- a/core/tests/switch_core/transport/test_postgres_transport.py +++ b/core/tests/switch_core/transport/test_postgres_transport.py @@ -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: