Consolidate MSC4242 state DAG checks via a TypeIs helper (#19774)

The reason for the change is to make it easier to support these checks
when porting event class to Rust.

Previously, code that needed to access `prev_state_events` had to
combine a `room_version.msc4242_state_dags` boolean check with an
`isinstance(event, FrozenEventVMSC4242)` cast (or `cast()`) for the type
checker. Introduce `supports_msc4242_state_dag()` in a new
`synapse/events/py_protocol.py` which does both in one step via
`TypeIs[MSC4242Event]`, removing the need to import the concrete
`FrozenEventVMSC4242` class at every call site.

`MSC4242Event` is an `EventBase` subclass used purely for type narrowing
— it's marked with a metaclass that rejects `isinstance()` to make
accidental runtime use loud.

No behavioural change: callers continue to gate on the same room version
flag and access the same `prev_state_events` attribute.

changelog.d/19774.misc

@@ -0,0 +1 @@
+Refactor MSC4242 state DAG checks behind a single `TypeIs` helper to avoid scattered `isinstance` casts.

synapse/event_auth.py

@@ -61,7 +61,8 @@ from synapse.api.room_versions import (
     EventFormatVersions,
     RoomVersion,
 )
-from synapse.events import FrozenEventVMSC4242, is_creator
+from synapse.events import is_creator
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.state import CREATE_KEY
 from synapse.storage.databases.main.events_worker import EventRedactBehaviour
 from synapse.types import (
@@ -187,8 +188,8 @@ async def check_state_independent_auth_rules(
         return
 
     # State DAGs 2. Considering the event's prev_state_events:
-    if event.room_version.msc4242_state_dags:
-        prev_state_events_ids = set(cast(FrozenEventVMSC4242, event).prev_state_events)
+    if supports_msc4242_state_dag(event):
+        prev_state_events_ids = set(event.prev_state_events)
         # Fetch all of the `prev_state_events`
         prev_state_events = {}
         # Try to load the `prev_state_events` from `batched_auth_events` initially as
@@ -515,8 +516,7 @@ def _check_create(event: "EventBase") -> None:
         raise AuthError(403, "Create event has prev events")
 
     # State DAGs 1.2 If it has any prev_state_events, reject.
-    if event.room_version.msc4242_state_dags:
-        assert isinstance(event, FrozenEventVMSC4242)
+    if supports_msc4242_state_dag(event):
         if len(event.prev_state_events) > 0:
             raise AuthError(403, "Create event has prev state events")
 

synapse/events/py_protocol.py

@@ -0,0 +1,88 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+#
+"""Type-narrowing helpers for `EventBase`.
+
+`EventBase` subclasses are split by room version (e.g. `FrozenEventV4`,
+`FrozenEventVMSC4242`), and certain attributes — such as `prev_state_events`
+on MSC4242 events — only exist on a subset of those subclasses. Branching on
+`room_version.<flag>` at runtime tells us *which* subclass we have, but the
+type checker can't see that link without an `isinstance` cast at every call
+site.
+
+This module provides "marker" subclasses of `EventBase` (`MSC4242Event`,
+etc.) paired with `TypeIs`-returning predicates (`supports_msc4242_state_dag`,
+etc.). A single call to the predicate both performs the room-version check
+and narrows the type — replacing the `if room_version.foo: assert
+isinstance(event, FrozenEventV...)` idiom.
+
+The marker classes are *type-only*: their metaclass raises on `isinstance`
+so they cannot be misused as real runtime classes. Add new markers and
+predicates here when a new room-version feature gates access to additional
+attributes.
+"""
+
+import abc
+from typing import (
+    TYPE_CHECKING,
+    Sequence,
+)
+
+from typing_extensions import TypeIs
+
+from synapse.events import EventBase
+
+if TYPE_CHECKING:
+    from synapse.events.snapshot import EventContext, EventPersistencePair
+
+
+class _DisableIsInstance(abc.ABCMeta):
+    """Metaclass which disables isinstance checks on classes which use it, by
+    making isinstance() raise NotImplementedError.
+
+    This is used to prevent isinstance checks on EventProtocol, which is a
+    helper class used for type narrowing of EventBase objects, but which should
+    not be used for isinstance checks itself (as its purely type annotation
+    rather than a real class).
+    """
+
+    def __instancecheck__(cls, instance: object) -> bool:
+        raise NotImplementedError("Instance cannot be used.")
+
+
+class EventProtocol(EventBase, metaclass=_DisableIsInstance):
+    """Helper subclass that allows type narrowing for `EventBase` objects."""
+
+
+class MSC4242Event(EventProtocol):
+    """Marker protocol for events in MSC4242 rooms. This allows us to narrow the
+    type of events."""
+
+    prev_state_events: list[str]
+
+
+def supports_msc4242_state_dag(event: EventBase) -> TypeIs[MSC4242Event]:
+    """Returns true if the given event is in a room that supports state DAGs
+    (MSC4242)"""
+
+    return event.room_version.msc4242_state_dags
+
+
+def all_supports_msc4242_state_dag(
+    obj: Sequence["EventPersistencePair"],
+) -> TypeIs[Sequence[tuple[MSC4242Event, "EventContext"]]]:
+    """Returns true if the given sequence of events are all in a room that
+    supports state DAGs (MSC4242)"""
+
+    return all(event.room_version.msc4242_state_dags for event, _ in obj)

synapse/handlers/admin.py

@@ -32,7 +32,8 @@ import attr
 
 from synapse.api.constants import Direction, EventTypes, Membership
 from synapse.api.errors import SynapseError
-from synapse.events import EventBase, FrozenEventVMSC4242
+from synapse.events import EventBase
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.events.utils import FilteredEvent
 from synapse.types import (
     JsonMapping,
@@ -495,8 +496,7 @@ class AdminHandler:
 
                 try:
                     prev_state_events = None
-                    if room_version.msc4242_state_dags:
-                        assert isinstance(event, FrozenEventVMSC4242)
+                    if supports_msc4242_state_dag(event):
                         prev_state_events = event.prev_state_events
                         assert prev_state_events is not None, (
                             "Parent event of redaction has no `prev_state_events` which should be impossible as `prev_state_events` is a required field in MSC4242 rooms"

synapse/handlers/message.py

@@ -53,8 +53,9 @@ from synapse.api.errors import (
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
 from synapse.api.urls import ConsentURIBuilder
 from synapse.event_auth import validate_event_for_room_version
-from synapse.events import EventBase, FrozenEventVMSC4242, relation_from_event
+from synapse.events import EventBase, relation_from_event
 from synapse.events.builder import EventBuilder
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.events.snapshot import (
     EventContext,
     EventPersistencePair,
@@ -1603,8 +1604,7 @@ class EventCreationHandler:
                         auth_event = event_id_to_event.get(event_id)
                         if auth_event:
                             batched_auth_events[event_id] = auth_event
-                    if event.room_version.msc4242_state_dags:
-                        assert isinstance(event, FrozenEventVMSC4242)
+                    if supports_msc4242_state_dag(event):
                         # State DAG rooms will check that the prev_state_events are not rejected.
                         # To do that, we need to make sure we pass in the prev_state_events as
                         # batched_auth_events, else we will fail the event due to the
@@ -1873,7 +1873,7 @@ class EventCreationHandler:
             state_entry = await self.state.resolve_state_groups_for_events(
                 event.room_id,
                 event_ids=event.prev_state_events
-                if isinstance(event, FrozenEventVMSC4242)
+                if supports_msc4242_state_dag(event)
                 else event.prev_event_ids(),
             )
 

synapse/state/__init__.py

@@ -37,7 +37,8 @@ from prometheus_client import Counter, Histogram
 
 from synapse.api.constants import EventTypes
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, StateResolutionVersions
-from synapse.events import EventBase, FrozenEventVMSC4242
+from synapse.events import EventBase
+from synapse.events.py_protocol import supports_msc4242_state_dag
 from synapse.events.snapshot import (
     EventContext,
     UnpersistedEventContext,
@@ -315,7 +316,7 @@ class StateHandler:
             # might redundantly recalculate the state for this event later.)
             prev_event_ids = frozenset(
                 event.prev_state_events
-                if isinstance(event, FrozenEventVMSC4242)
+                if supports_msc4242_state_dag(event)
                 else event.prev_event_ids()
             )
             incomplete_prev_events = await self.store.get_partial_state_events(

synapse/storage/controllers/persist_events.py

@@ -34,6 +34,7 @@ from typing import (
     Generator,
     Generic,
     Iterable,
+    Sequence,
     TypeVar,
     cast,
 )
@@ -46,7 +47,14 @@ from twisted.internet import defer
 from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import SynapseError
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
-from synapse.events import EventBase, FrozenEventVMSC4242, event_exists_in_state_dag
+from synapse.events import (
+    EventBase,
+    event_exists_in_state_dag,
+)
+from synapse.events.py_protocol import (
+    MSC4242Event,
+    all_supports_msc4242_state_dag,
+)
 from synapse.events.snapshot import EventContext, EventPersistencePair
 from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
 from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
@@ -637,7 +645,6 @@ class EventsPersistenceStorageController:
         # Get the room version for the first event. This room version is the same for all events
         # as events_and_contexts is all for one room.
         assert len(events_and_contexts) > 0
-        room_version = events_and_contexts[0][0].room_version
 
         for chunk in chunks:
             # We can't easily parallelize these since different chunks
@@ -648,22 +655,18 @@ class EventsPersistenceStorageController:
             new_state_dag_extrems = None
 
             if not backfilled:
-                if room_version.msc4242_state_dags:
+                if all_supports_msc4242_state_dag(chunk):
                     with Measure(
                         self._clock,
                         name="_process_state_dag_forward_extremities_and_state_delta",
                         server_name=self.server_name,
                     ):
-                        assert all(
-                            isinstance(ev, FrozenEventVMSC4242) for ev, _ in chunk
-                        )
                         (
                             new_forward_extremities,  # for prev_events
                             state_delta_for_room,  # for state groups
                             new_state_dag_extrems,  # for prev_state_events
                         ) = await self._process_state_dag_forward_extremities_and_state_delta(
-                            room_id,
-                            cast(list[tuple[FrozenEventVMSC4242, EventContext]], chunk),
+                            room_id, chunk
                         )
                 else:
                     with Measure(
@@ -840,7 +843,7 @@ class EventsPersistenceStorageController:
     async def _process_state_dag_forward_extremities_and_state_delta(
         self,
         room_id: str,
-        event_contexts: list[tuple[FrozenEventVMSC4242, EventContext]],
+        event_contexts: Sequence[tuple[MSC4242Event, EventContext]],
     ) -> tuple[set[str] | None, DeltaState | None, set[str] | None]:
         """Process the forwards extremities for state DAG rooms.
         Returns:
@@ -933,7 +936,7 @@ class EventsPersistenceStorageController:
         self,
         room_id: str,
         existing_fwd_extrems: frozenset[str],
-        event_contexts: list[tuple[FrozenEventVMSC4242, EventContext]],
+        event_contexts: Sequence[tuple[MSC4242Event, EventContext]],
     ) -> set[str]:
         """Calculate the new state dag forward extremities. Modifies existing_fwd_extrems.
 

synapse/storage/databases/main/events.py

@@ -48,12 +48,12 @@ from synapse.api.errors import PartialStateConflictError
 from synapse.api.room_versions import RoomVersions
 from synapse.events import (
     EventBase,
-    FrozenEventVMSC4242,
     StrippedStateEvent,
     event_exists_in_state_dag,
     is_creator,
     relation_from_event,
 )
+from synapse.events.py_protocol import MSC4242Event, supports_msc4242_state_dag
 from synapse.events.snapshot import EventPersistencePair
 from synapse.events.utils import parse_stripped_state_event
 from synapse.logging.opentracing import trace
@@ -2897,10 +2897,7 @@ class PersistEventsStore:
 
             self._handle_event_relations(txn, event)
 
-            if event.room_version.msc4242_state_dags and event_exists_in_state_dag(
-                event
-            ):
-                assert isinstance(event, FrozenEventVMSC4242)
+            if supports_msc4242_state_dag(event) and event_exists_in_state_dag(event):
                 self._store_state_dag_edges(txn, event)
 
             # Store the labels for this event.
@@ -2980,7 +2977,7 @@ class PersistEventsStore:
         txn.call_after(local_prefill)
 
     def _store_state_dag_edges(
-        self, txn: LoggingTransaction, event: FrozenEventVMSC4242
+        self, txn: LoggingTransaction, event: MSC4242Event
     ) -> None:
         # the create event has no edge but we still need to persist it as get_state_dag just
         # yanks all rows in this table. It's a bit gross to store NULL as the prev_state_event_id

tests/events/test_py_protocol.py

@@ -0,0 +1,82 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+#
+
+from unittest.mock import Mock
+
+from synapse.api.room_versions import RoomVersion, RoomVersions
+from synapse.events import EventBase, FrozenEvent, make_event_from_dict
+from synapse.events.py_protocol import (
+    EventProtocol,
+    MSC4242Event,
+    all_supports_msc4242_state_dag,
+    supports_msc4242_state_dag,
+)
+
+from tests.unittest import TestCase
+
+
+def _make_event(room_version: RoomVersion) -> EventBase:
+    """Helper to make an EventBase with the given room version."""
+    event_dict = {
+        "content": {},
+        "sender": "@user:example.com",
+        "type": "m.room.message",
+        "room_id": "!room:example.com",
+    }
+    if room_version.msc4242_state_dags:
+        event_dict["prev_state_events"] = []
+    return make_event_from_dict(event_dict, room_version=room_version)
+
+
+class TestMetaClass(TestCase):
+    def test_is_instance(self) -> None:
+        """Test that isinstance checks on EventProtocol raise
+        NotImplementedError, but that isinstance checks on EventBase and
+        FrozenEvent still work as normal.
+        """
+        # EventBase and FrozenEvent should work as normal
+        self.assertFalse(isinstance(object(), EventBase))
+        self.assertFalse(isinstance(object(), FrozenEvent))
+
+        with self.assertRaises(NotImplementedError):
+            isinstance(object(), EventProtocol)
+
+        with self.assertRaises(NotImplementedError):
+            isinstance(object(), MSC4242Event)
+
+
+class SupportsMSC4242StateDagTestCase(TestCase):
+    def test_single_event_msc4242(self) -> None:
+        """A single event in an MSC4242 room is recognised."""
+        ev = _make_event(RoomVersions.MSC4242v12)
+        self.assertTrue(supports_msc4242_state_dag(ev))
+
+    def test_single_event_non_msc4242(self) -> None:
+        """A single event in a non-MSC4242 room is not recognised."""
+        ev = _make_event(RoomVersions.V11)
+        self.assertFalse(supports_msc4242_state_dag(ev))
+
+    def test_sequence_all_msc4242(self) -> None:
+        """A sequence of MSC4242 (event, context) pairs is recognised."""
+        pairs = [(_make_event(RoomVersions.MSC4242v12), Mock()) for _ in range(3)]
+        self.assertTrue(all_supports_msc4242_state_dag(pairs))
+
+    def test_sequence_mixed(self) -> None:
+        """A sequence containing any non-MSC4242 event is not recognised."""
+        pairs = [
+            (_make_event(RoomVersions.MSC4242v12), Mock()),
+            (_make_event(RoomVersions.V11), Mock()),
+        ]
+        self.assertFalse(all_supports_msc4242_state_dag(pairs))

tests/storage/test_msc4242_state_dag.py

@@ -19,7 +19,10 @@ from twisted.test.proto_helpers import MemoryReactor
 from synapse.api.constants import EventTypes
 from synapse.api.errors import SynapseError
 from synapse.api.room_versions import RoomVersions
-from synapse.events import FrozenEventVMSC4242, make_event_from_dict
+from synapse.events import (
+    make_event_from_dict,
+)
+from synapse.events.py_protocol import MSC4242Event, supports_msc4242_state_dag
 from synapse.events.snapshot import EventContext
 from synapse.rest.client import room
 from synapse.server import HomeServer
@@ -152,7 +155,7 @@ class MSC4242EventPersistenceStateDagsStoreTestCase(HomeserverTestCase):
         id: str,
         prev_state_events: list[str],
         rejected: bool = False,
-    ) -> tuple[FrozenEventVMSC4242, EventContext]:
+    ) -> tuple[MSC4242Event, EventContext]:
         ev = make_event_from_dict(
             {
                 "prev_state_events": prev_state_events,
@@ -166,8 +169,8 @@ class MSC4242EventPersistenceStateDagsStoreTestCase(HomeserverTestCase):
             },
             room_version=RoomVersions.MSC4242v12,
         )
-        assert isinstance(ev, FrozenEventVMSC4242)
-        ev._event_id = id
+        ev._event_id = id  # type: ignore[attr-defined]
+        assert supports_msc4242_state_dag(ev)
         ctx = Mock()
         ctx.rejected = rejected
         return ev, ctx
@@ -175,7 +178,7 @@ class MSC4242EventPersistenceStateDagsStoreTestCase(HomeserverTestCase):
     def _test(
         self,
         current_fwds: list[str],
-        new_events: list[tuple[FrozenEventVMSC4242, EventContext]],
+        new_events: list[tuple[MSC4242Event, EventContext]],
         want_new_extrems: set[str],
         want_raises: bool = False,
     ) -> None: