diff --git a/rust/src/events/serialize.rs b/rust/src/events/serialize.rs
index 100bc8383c..73e51c4c8f 100644
--- a/rust/src/events/serialize.rs
+++ b/rust/src/events/serialize.rs
@@ -28,7 +28,7 @@
 //! batch and then serializes each event — recursing entirely in Rust into
 //! redactions and bundled aggregations via [`serialize_event`].
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use pyo3::{
     exceptions::{PyTypeError, PyValueError},
@@ -132,12 +132,29 @@ pub struct SerializeEventConfig {
     /// `true`, the sender is repeated in `unsigned` as
     /// `org.matrix.msc4428.stable_id`.
     msc4428_enabled: bool,
+    /// MSC4430: the requester's member keys. In a member key room the `sender`
+    /// of the requester's own events is their key in that room rather than
+    /// their user ID, so sender-only fields (`transaction_id`, `delay_id`)
+    /// are also gated on these.
+    requester_member_keys: Option<HashSet<String>>,
 }
 
 #[pymethods]
 impl SerializeEventConfig {
     #[new]
     #[allow(clippy::too_many_arguments)]
+    #[pyo3(signature = (
+        *,
+        as_client_event,
+        event_format,
+        requester,
+        event_field_allowlist,
+        include_stripped_room_state,
+        include_admin_metadata,
+        msc4354_enabled,
+        msc4428_enabled,
+        requester_member_keys = None,
+    ))]
     fn new(
         as_client_event: bool,
         event_format: EventFormat,
@@ -147,8 +164,10 @@ impl SerializeEventConfig {
         include_admin_metadata: bool,
         msc4354_enabled: bool,
         msc4428_enabled: bool,
+        requester_member_keys: Option<Vec<String>>,
     ) -> PyResult<Self> {
         let requester = requester.map(|r| r.get().clone());
+        let requester_member_keys = requester_member_keys.map(HashSet::from_iter);
 
         Ok(Self {
             as_client_event,
@@ -159,6 +178,7 @@ impl SerializeEventConfig {
             include_admin_metadata,
             msc4354_enabled,
             msc4428_enabled,
+            requester_member_keys,
         })
     }
 }
@@ -437,7 +457,12 @@ fn serialize_event_value(
     // was sent by the same session (or, where appropriate, the same sender) as
     // the one requesting the event.
     if let Some(requester) = &config.requester {
-        if requester.user_id == event.sender() {
+        let sent_by_requester = requester.user_id == event.sender()
+            || config
+                .requester_member_keys
+                .as_ref()
+                .is_some_and(|keys| keys.contains(event.sender()));
+        if sent_by_requester {
             if let Some(txn_id) = event.internal_metadata.txn_id()? {
                 if let Some(event_device_id) = event.internal_metadata.device_id()? {
                     if Some(event_device_id.as_str()) == requester.device_id.as_deref() {
diff --git a/synapse/events/utils.py b/synapse/events/utils.py
index 5668ba2294..c2f37e4e2f 100644
--- a/synapse/events/utils.py
+++ b/synapse/events/utils.py
@@ -183,7 +183,8 @@ class EventClientSerializer:
         Helper method that sets the `include_admin_metadata` field based on
         whether the requester is a server admin if it is not explicitly
         provided. Also sets the `msc4354_enabled` and `msc4428_enabled` fields
-        based on the homeserver config.
+        based on the homeserver config, and, with MSC4430 on, the requester's
+        member keys (the senders of their own events in member key rooms).
 
         Args:
             as_client_event: Whether to serialize the events as client events.
@@ -210,6 +211,14 @@ class EventClientSerializer:
             else:
                 include_admin_metadata = False
 
+        requester_member_keys: list[str] | None = None
+        if requester is not None and self._config.experimental.msc4430_enabled:
+            keys_by_room = await self._store.get_local_member_keys_for_user(
+                requester.user.to_string()
+            )
+            if keys_by_room:
+                requester_member_keys = list(keys_by_room.values())
+
         return SerializeEventConfig(
             as_client_event=as_client_event,
             event_format=event_format,
@@ -219,6 +228,7 @@ class EventClientSerializer:
             include_admin_metadata=include_admin_metadata,
             msc4354_enabled=self._config.experimental.msc4354_enabled,
             msc4428_enabled=self._config.experimental.msc4428_enabled,
+            requester_member_keys=requester_member_keys,
         )
 
     async def serialize_event(
diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py
index 6826c36745..725d54bb7c 100644
--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -717,6 +717,15 @@ class DeviceHandler:
 
         return device_list_updates
 
+    async def _users_behind_members(
+        self, room_id: str, members: Iterable[str]
+    ) -> set[str]:
+        """The users behind a room's membership state keys: the keys themselves
+        in rooms without member keys; in an MSC4430 member key room, the users
+        the keys map to (keys that map to nobody are left out)."""
+        mapped = await self.store.get_user_ids_for_identities(room_id, members)
+        return {user_id for user_id in mapped.values() if user_id is not None}
+
     async def generate_sync_entry_for_device_list(
         self,
         user_id: str,
@@ -774,7 +783,10 @@ class DeviceHandler:
         # Step 1b, check for newly joined rooms
         for room_id in newly_joined_rooms:
             joined_users = await self.store.get_users_in_room(room_id)
-            newly_joined_or_invited_or_knocked_users.update(joined_users)
+            # (Member keys in an MSC4430 room; device lists are about users.)
+            newly_joined_or_invited_or_knocked_users.update(
+                await self._users_behind_members(room_id, joined_users)
+            )
 
         # TODO: Check that these users are actually new, i.e. either they
         # weren't in the previous sync *or* they left and rejoined.
@@ -788,7 +800,9 @@ class DeviceHandler:
         # Now find users that we no longer track
         for room_id in newly_left_rooms:
             left_users = await self.store.get_users_in_room(room_id)
-            newly_left_users.update(left_users)
+            newly_left_users.update(
+                await self._users_behind_members(room_id, left_users)
+            )
 
         # Remove any users that we still share a room with.
         left_users_rooms = await self.store.get_rooms_for_users(newly_left_users)
diff --git a/synapse/handlers/initial_sync.py b/synapse/handlers/initial_sync.py
index 56f4d86d41..1a163918ca 100644
--- a/synapse/handlers/initial_sync.py
+++ b/synapse/handlers/initial_sync.py
@@ -466,7 +466,14 @@ class InitialSyncHandler:
             if not self.hs.config.server.presence_enabled:
                 return []
 
-            states = await presence_handler.get_states([m.sender for m in room_members])
+            # (The members as users: in an MSC4430 member key room the
+            # senders are their keys.)
+            member_users = await self.store.get_user_ids_for_identities(
+                room_id, [m.sender for m in room_members]
+            )
+            states = await presence_handler.get_states(
+                [u for u in member_users.values() if u is not None]
+            )
 
             return [
                 {
diff --git a/synapse/handlers/member_info.py b/synapse/handlers/member_info.py
index 1400ff7023..ad124dcf0e 100644
--- a/synapse/handlers/member_info.py
+++ b/synapse/handlers/member_info.py
@@ -13,11 +13,14 @@
 #
 """MSC4428: the per-room `member_info` map.
 
-`member_info` maps a room member's stable ID (today: their user ID, carried on
-events as `unsigned.org.matrix.msc4428.stable_id`) to the user ID it currently
-stands for, which may be `None` once the user is erased. Entries whose user ID
-equals the stable ID are omitted, as the MSC allows, so in today's rooms the map
-only ever names erased users.
+`member_info` maps a room member's stable ID (carried on events as
+`unsigned.org.matrix.msc4428.stable_id`) to the user ID it currently stands
+for, which may be `None`. In rooms without member keys the stable ID is the
+user ID itself and the entry is omitted, as the MSC allows, unless the user is
+erased. In MSC4430 member key rooms the stable ID is the member key, and every
+key gets an entry: the user behind a key we hold (`None` once erased), the user
+a verified link names for another server's key, and `None` for a key whose
+link is unverified or unknown.
 
 Most changes to the map come with an event in the room (a join, a leave) and
 are delivered alongside it. Changes without such an anchor, erasure today, are
@@ -41,6 +44,7 @@ class MemberInfoHandler:
         self._store = hs.get_datastores().main
         self._state_storage_controller = hs.get_storage_controllers().state
         self._notifier = hs.get_notifier()
+        self._member_key_handler = hs.get_member_key_handler()
         self.enabled = hs.config.experimental.msc4428_enabled
 
     async def on_user_erased(self, user_id: str) -> None:
@@ -54,8 +58,11 @@ class MemberInfoHandler:
         if not room_ids:
             return
 
+        # The stable ID that changed: the user's member key in an MSC4430
+        # member key room, their user ID elsewhere.
+        member_keys = await self._store.get_local_member_keys_for_user(user_id)
         token = await self._store.add_member_info_changes(
-            dict.fromkeys(room_ids, (user_id,))
+            {room_id: (member_keys.get(room_id, user_id),) for room_id in room_ids}
         )
         self._notifier.on_new_event(StreamKeyType.MEMBER_INFO, token, rooms=room_ids)
 
@@ -96,7 +103,10 @@ class MemberInfoHandler:
         """The `member_info` entries for the given stable IDs.
 
         Only stable IDs whose mapping differs from the stable ID itself get an
-        entry; today that is erased users, whose user ID is `None`.
+        entry: a user ID gets one only once the user is erased (`None`); a
+        member key always gets one, naming the user behind it (see
+        `MemberKeyHandler.user_ids_for_identities`), `None` if that user is
+        erased or the key cannot be attributed to anyone.
 
         Args:
             room_id: The room the stable IDs belong to.
@@ -108,12 +118,19 @@ class MemberInfoHandler:
         if not stable_ids:
             return {}
 
-        erased = await self._store.are_users_erased(stable_ids)
-        return {
-            stable_id: {MemberInfo.USER_ID: None}
-            for stable_id, is_erased in erased.items()
-            if is_erased
-        }
+        user_ids = await self._member_key_handler.user_ids_for_identities(
+            room_id, stable_ids
+        )
+        erased = await self._store.are_users_erased(
+            {user_id for user_id in user_ids.values() if user_id is not None}
+        )
+        result: JsonDict = {}
+        for stable_id, user_id in user_ids.items():
+            if user_id is not None and erased.get(user_id):
+                user_id = None
+            if user_id != stable_id:
+                result[stable_id] = {MemberInfo.USER_ID: user_id}
+        return result
 
     async def member_info_for_response(
         self,
diff --git a/synapse/handlers/member_keys.py b/synapse/handlers/member_keys.py
index 0c95333b00..1f78051f6d 100644
--- a/synapse/handlers/member_keys.py
+++ b/synapse/handlers/member_keys.py
@@ -37,7 +37,7 @@ vouch for linkage information that is not ours.
 """
 
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Iterable
 
 from prometheus_client import Counter
 from signedjson.key import generate_signing_key
@@ -162,15 +162,14 @@ class MemberKeyHandler:
         behind one of our member keys, or the *verified* link of another
         server's key. `None` for a key we hold no verified link for: such a key
         maps to no user, whatever it claims."""
-        if not is_member_key(identity):
-            return identity
-        local_user_id = await self.local_user_id_for_identity(room_id, identity)
-        if local_user_id is not None:
-            return local_user_id
-        link = await self._store.get_member_link(room_id, identity)
-        if link is None:
-            return None
-        return link.verified_user_id
+        return (await self.user_ids_for_identities(room_id, [identity]))[identity]
+
+    async def user_ids_for_identities(
+        self, room_id: str, identities: Iterable[str]
+    ) -> dict[str, str | None]:
+        """Batched `user_id_for_identity`: every identity given maps to a user ID
+        or to `None`."""
+        return await self._store.get_user_ids_for_identities(room_id, identities)
 
     async def domain_for_identity(self, room_id: str, identity: str) -> str | None:
         """The server a room member identity belongs to, for routing only: the
diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py
index 3aa0d91eb2..415c79a12b 100644
--- a/synapse/handlers/sliding_sync/__init__.py
+++ b/synapse/handlers/sliding_sync/__init__.py
@@ -107,6 +107,7 @@ class SlidingSyncHandler:
         self.event_sources = hs.get_event_sources()
         self.relations_handler = hs.get_relations_handler()
         self.member_info_handler = hs.get_member_info_handler()
+        self._member_key_handler = hs.get_member_key_handler()
         self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync
         self.is_mine_id = hs.is_mine_id
 
@@ -611,6 +612,11 @@ class SlidingSyncHandler:
             is_dm: Whether the room is a DM room
         """
         user = sync_config.user
+        # The user as this room knows them: their member key, in an MSC4430
+        # member key room. It is what `$ME` and the heroes are about.
+        user_room_identity = await self._member_key_handler.room_identity_for_user(
+            room_id, user.to_string()
+        )
 
         set_tag(
             SynapseTags.FUNC_ARG_PREFIX + "membership",
@@ -1009,7 +1015,7 @@ class SlidingSyncHandler:
                 # TODO: Reverse/rewind back to the `to_token`
 
             hero_user_ids = extract_heroes_from_room_summary(
-                room_membership_summary, me=user.to_string()
+                room_membership_summary, me=user_room_identity
             )
 
         # Fetch the membership counts for rooms we're joined to.
@@ -1180,7 +1186,7 @@ class SlidingSyncHandler:
                             # their user ID.
                             normalized_state_key = state_key
                             if state_key == StateValues.ME:
-                                normalized_state_key = user.to_string()
+                                normalized_state_key = user_room_identity
 
                             if state_type == EventTypes.Member:
                                 # Also track explicitly requested member state for
@@ -1320,7 +1326,7 @@ class SlidingSyncHandler:
                 # Check if there are any changes to the required state config
                 # that we need to handle.
                 changes_return = _required_state_changes(
-                    user.to_string(),
+                    user_room_identity,
                     prev_required_state_map=prev_room_sync_config.required_state_map,
                     request_required_state_map=room_sync_config.required_state_map,
                     previously_returned_lazy_user_ids=previously_returned_user_to_last_seen.keys(),
diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py
index 7f3ee4877a..91d93a6cf1 100644
--- a/synapse/handlers/sliding_sync/room_lists.py
+++ b/synapse/handlers/sliding_sync/room_lists.py
@@ -45,7 +45,7 @@ from synapse.storage.databases.main.state import (
     Sentinel as StateSentinel,
 )
 from synapse.storage.databases.main.stream import CurrentStateDeltaMembership
-from synapse.storage.invite_rule import InviteRule
+from synapse.storage.invite_rule import InviteRule, InviteRulesConfig
 from synapse.storage.roommember import (
     RoomsForUser,
     RoomsForUserSlidingSync,
@@ -189,6 +189,7 @@ def filter_membership_for_sync(
     user_id: str,
     room_membership_for_user: RoomsForUserType,
     newly_left: bool,
+    member_keys_by_room: Mapping[str, str] | None = None,
 ) -> bool:
     """
     Returns True if the membership event should be included in the sync response,
@@ -214,13 +215,20 @@ def filter_membership_for_sync(
     # from the room without a corresponding leave event. We can just remove the rooms
     # since they are no longer relevant to the user but will still appear if they are
     # `newly_left`.
+    # The user as the room knows them: in an MSC4430 member key room, their
+    # member key there (`member_keys_by_room`, from
+    # `get_local_member_keys_for_user`), under which their own leave is sent.
+    own_identities: tuple[str | None, ...] = (user_id, None)
+    if member_keys_by_room:
+        own_identities += (member_keys_by_room.get(room_membership_for_user.room_id),)
+
     return (
         # Anything except leave events
         membership != Membership.LEAVE
         # Unless...
         or newly_left
         # Allow kicks
-        or (membership == Membership.LEAVE and sender not in (user_id, None))
+        or (membership == Membership.LEAVE and sender not in own_identities)
     )
 
 
@@ -234,6 +242,40 @@ class SlidingSyncRoomLists:
         self.is_mine_id = hs.is_mine_id
         self._clock = hs.get_clock()
         self._member_info_handler = hs.get_member_info_handler()
+        self._msc4430_enabled = hs.config.experimental.msc4430_enabled
+
+    async def _invite_is_ignored(
+        self,
+        room_id: str,
+        inviter: str,
+        ignored_users: AbstractSet[str],
+        invite_config: InviteRulesConfig | None,
+    ) -> bool:
+        """Whether an invite is hidden because of who sent it: the inviter is on
+        the user's ignore list, or (when given) the user's invite rules ignore
+        them. In an MSC4430 member key room the inviter is a member key, which
+        may itself be ignored, and which maps to a user (if to anyone) for the
+        user-keyed checks; a key that maps to nobody is not ignored by them."""
+        if inviter in ignored_users:
+            return True
+        inviter_user_id = (
+            await self.store.get_user_ids_for_identities(room_id, [inviter])
+        )[inviter]
+        if inviter_user_id is None:
+            return False
+        if inviter_user_id in ignored_users:
+            return True
+        return (
+            invite_config is not None
+            and invite_config.get_invite_rule(inviter_user_id) == InviteRule.IGNORE
+        )
+
+    async def _member_keys_by_room(self, user_id: str) -> Mapping[str, str]:
+        """The user's member keys per MSC4430 member key room: the identities
+        their own membership events carry there. Empty with the feature off."""
+        if not self._msc4430_enabled:
+            return {}
+        return await self.store.get_local_member_keys_for_user(user_id)
 
     async def compute_interested_rooms(
         self,
@@ -281,6 +323,7 @@ class SlidingSyncRoomLists:
     ) -> SlidingSyncInterestedRooms:
         """Implementation of `compute_interested_rooms` using new sliding sync db tables."""
         user_id = sync_config.user.to_string()
+        member_keys_by_room = await self._member_keys_by_room(user_id)
 
         # Assemble sliding window lists
         lists: dict[str, SlidingSyncResult.SlidingWindowList] = {}
@@ -331,12 +374,11 @@ class SlidingSyncRoomLists:
                 if (
                     room_for_user_sliding_sync.membership == Membership.INVITE
                     and room_for_user_sliding_sync.sender
-                    and (
-                        room_for_user_sliding_sync.sender in ignored_users
-                        or invite_config.get_invite_rule(
-                            room_for_user_sliding_sync.sender
-                        )
-                        == InviteRule.IGNORE
+                    and await self._invite_is_ignored(
+                        room_id,
+                        room_for_user_sliding_sync.sender,
+                        ignored_users,
+                        invite_config,
                     )
                 ):
                     room_membership_for_user_map.pop(room_id, None)
@@ -377,6 +419,7 @@ class SlidingSyncRoomLists:
                         user_id=user_id,
                         room_membership_for_user=room_for_user,
                         newly_left=room_id in newly_left_room_map,
+                        member_keys_by_room=member_keys_by_room,
                     ):
                         room_membership_for_user_map[room_id] = room_for_user
                     else:
@@ -412,6 +455,7 @@ class SlidingSyncRoomLists:
                         user_id=user_id,
                         room_membership_for_user=newly_left_room_for_user_sliding_sync,
                         newly_left=room_id in newly_left_room_map,
+                        member_keys_by_room=member_keys_by_room,
                     ):
                         room_membership_for_user_map[room_id] = (
                             newly_left_room_for_user_sliding_sync
@@ -438,6 +482,7 @@ class SlidingSyncRoomLists:
                             user_id=user_id,
                             room_membership_for_user=room_for_user,
                             newly_left=room_id in newly_left_room_map,
+                            member_keys_by_room=member_keys_by_room,
                         ):
                             room_membership_for_user_map[room_id] = room_for_user
                         else:
@@ -478,6 +523,7 @@ class SlidingSyncRoomLists:
                         user_id=user_id,
                         room_membership_for_user=room_for_user,
                         newly_left=room_id in newly_left_room_map,
+                        member_keys_by_room=member_keys_by_room,
                     ):
                         room_membership_for_user_map[room_id] = room_for_user
                     else:
@@ -1200,14 +1246,20 @@ class SlidingSyncRoomLists:
         # Remove invites from ignored users
         ignored_users = await self.store.ignored_users(user_id)
         if ignored_users:
-            room_for_user_list = [
-                room_for_user
-                for room_for_user in room_for_user_list
-                if not (
+            kept_room_for_user_list = []
+            for room_for_user in room_for_user_list:
+                if (
                     room_for_user.membership == Membership.INVITE
-                    and room_for_user.sender in ignored_users
-                )
-            ]
+                    and await self._invite_is_ignored(
+                        room_for_user.room_id,
+                        room_for_user.sender,
+                        ignored_users,
+                        None,
+                    )
+                ):
+                    continue
+                kept_room_for_user_list.append(room_for_user)
+            room_for_user_list = kept_room_for_user_list
 
         (
             newly_joined_room_ids,
@@ -1502,6 +1554,7 @@ class SlidingSyncRoomLists:
             with membership information in that room at the time of `to_token`.
         """
         user_id = user.to_string()
+        member_keys_by_room = await self._member_keys_by_room(user_id)
 
         # Filter rooms to only what we're interested to sync with
         filtered_sync_room_map = {
@@ -1511,6 +1564,7 @@ class SlidingSyncRoomLists:
                 user_id=user_id,
                 room_membership_for_user=room_membership_for_user,
                 newly_left=room_id in newly_left_room_ids,
+                member_keys_by_room=member_keys_by_room,
             )
         }
 
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 94c65192bf..493704bd18 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -64,7 +64,7 @@ from synapse.metrics import SERVER_NAME_LABEL
 from synapse.storage.databases.main.event_push_actions import RoomNotifCounts
 from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
 from synapse.storage.databases.main.stream import PaginateFunction
-from synapse.storage.invite_rule import InviteRule
+from synapse.storage.invite_rule import InviteRule, InviteRulesConfig
 from synapse.storage.roommember import MemberSummary
 from synapse.types import (
     DeviceListUpdates,
@@ -429,6 +429,8 @@ class SyncHandler:
         self._storage_controllers = hs.get_storage_controllers()
         self._state_storage_controller = self._storage_controllers.state
         self._member_info_handler = hs.get_member_info_handler()
+        self._member_key_handler = hs.get_member_key_handler()
+        self._msc4430_enabled = hs.config.experimental.msc4430_enabled
         self._device_handler = hs.get_device_handler()
         self._task_scheduler = hs.get_task_scheduler()
 
@@ -1163,7 +1165,10 @@ class SyncHandler:
             for user_id, event_id in details.get(membership, empty_ms).members:
                 member_ids[user_id] = event_id
 
-        me = sync_config.user.to_string()
+        # The user as the room knows them (their member key, in an MSC4430 room).
+        me = await self._member_key_handler.room_identity_for_user(
+            room_id, sync_config.user.to_string()
+        )
         summary["m.heroes"] = extract_heroes_from_room_summary(details, me)
 
         if not sync_config.filter_collection.lazy_load_members():
@@ -1500,7 +1505,15 @@ class SyncHandler:
             # is no guarantee that our membership will be in the auth events of
             # timeline events when the room is partial stated.
             state_filter = StateFilter.from_lazy_load_member_list(
-                members_to_fetch.union((sync_config.user.to_string(),))
+                members_to_fetch.union(
+                    (
+                        # Under the identity the room knows us by (our member
+                        # key, in an MSC4430 room).
+                        await self._member_key_handler.room_identity_for_user(
+                            room_id, sync_config.user.to_string()
+                        ),
+                    )
+                )
             )
 
             # We are happy to use partial state to compute the `/sync` response.
@@ -2060,6 +2073,16 @@ class SyncHandler:
                     newly_joined_or_invited_or_knocked_users,
                     newly_left_users,
                 ) = sync_result_builder.calculate_user_changes()
+                # Those are membership state keys: in an MSC4430 member key
+                # room, keys. Presence and device lists are about users.
+                newly_joined_or_invited_or_knocked_users = (
+                    await self._users_for_member_identities(
+                        sync_result_builder, newly_joined_or_invited_or_knocked_users
+                    )
+                )
+                newly_left_users = await self._users_for_member_identities(
+                    sync_result_builder, newly_left_users
+                )
 
                 if include_presence_data:
                     logger.debug("Fetching presence data")
@@ -2217,8 +2240,15 @@ class SyncHandler:
                 # User joined a room - we have to then check the room state to ensure we
                 # respect any bans if there's a race between the join and ban events.
                 if event.membership == Membership.JOIN:
-                    user_ids_in_room = await self.store.get_users_in_room(room_id)
-                    if user_id in user_ids_in_room:
+                    # (Keyed by user ID, which in an MSC4430 member key room
+                    # the room's own state is not.)
+                    (
+                        current_membership,
+                        _,
+                    ) = await self.store.get_local_current_membership_for_user_in_room(
+                        user_id, room_id
+                    )
+                    if current_membership == Membership.JOIN:
                         mutable_joined_room_ids.add(room_id)
                 # The user left the room, or left and was re-invited but not joined yet
                 else:
@@ -2736,7 +2766,14 @@ class SyncHandler:
         extra_users_ids = set(newly_joined_or_invited_users)
         for room_id in newly_joined_rooms:
             users = await self.store.get_users_in_room(room_id)
-            extra_users_ids.update(users)
+            # (Member keys in an MSC4430 room; presence is about users.)
+            extra_users_ids.update(
+                u
+                for u in (
+                    await self.store.get_user_ids_for_identities(room_id, users)
+                ).values()
+                if u is not None
+            )
         extra_users_ids.discard(user.to_string())
 
         if extra_users_ids:
@@ -2966,6 +3003,67 @@ class SyncHandler:
                 return True
         return False
 
+    async def _users_for_member_identities(
+        self,
+        sync_result_builder: "SyncResultBuilder",
+        identities: AbstractSet[str],
+    ) -> AbstractSet[str]:
+        """Map membership state keys found in the joined rooms' timelines (see
+        `SyncResultBuilder.calculate_user_changes`) to users: user IDs are kept,
+        and in MSC4430 member key rooms the keys become the users behind them
+        (dropped if that is nobody)."""
+        if not identities or not self._msc4430_enabled:
+            return identities
+        # Which room each key was seen in, so it can be looked up there.
+        identity_rooms: dict[str, str] = {}
+        for joined_sync in sync_result_builder.joined:
+            for ev in joined_sync.timeline.events:
+                if (
+                    ev.event.type == EventTypes.Member
+                    and ev.event.state_key in identities
+                ):
+                    identity_rooms[ev.event.state_key] = joined_sync.room_id
+        users: set[str] = set()
+        by_room: dict[str, set[str]] = {}
+        for identity in identities:
+            room_id = identity_rooms.get(identity)
+            if room_id is None:
+                users.add(identity)
+            else:
+                by_room.setdefault(room_id, set()).add(identity)
+        for room_id, room_identities in by_room.items():
+            mapped = await self.store.get_user_ids_for_identities(
+                room_id, room_identities
+            )
+            users.update(u for u in mapped.values() if u is not None)
+        return users
+
+    async def _invite_is_ignored(
+        self,
+        room_id: str,
+        inviter: str,
+        ignored_users: AbstractSet[str],
+        invite_config: InviteRulesConfig,
+    ) -> bool:
+        """Whether an invite is hidden from the user because of who sent it: the
+        inviter is on their ignore list, or their invite rules ignore them. In
+        an MSC4430 member key room the inviter is a member key, which the user
+        may have ignored as such, and which maps to a user (if it maps to any)
+        for the user-keyed checks; a key that maps to nobody cannot be matched
+        by the rules and is not ignored by them.
+        """
+        if inviter in ignored_users:
+            return True
+        inviter_user_id = await self._member_key_handler.user_id_for_identity(
+            room_id, inviter
+        )
+        if inviter_user_id is None:
+            return False
+        return (
+            inviter_user_id in ignored_users
+            or invite_config.get_invite_rule(inviter_user_id) == InviteRule.IGNORE
+        )
+
     async def _get_room_changes_for_incremental_sync(
         self,
         sync_result_builder: "SyncResultBuilder",
@@ -2994,6 +3092,13 @@ class SyncHandler:
         """
         user_id = sync_result_builder.sync_config.user.to_string()
         since_token = sync_result_builder.since_token
+        # The user's identities in MSC4430 member key rooms (their keys), under
+        # which their membership is held in those rooms' state.
+        member_keys_by_room: Mapping[str, str] = (
+            await self.store.get_local_member_keys_for_user(user_id)
+            if self._msc4430_enabled
+            else {}
+        )
         now_token = sync_result_builder.now_token
         sync_config = sync_result_builder.sync_config
         membership_change_events = sync_result_builder.membership_change_events
@@ -3040,13 +3145,18 @@ class SyncHandler:
                 # User is in the room so we don't need to do the invite/leave checks
                 continue
 
+            room_user_id = member_keys_by_room.get(room_id, user_id)
             if room_id in sync_result_builder.joined_room_ids or has_join:
                 old_state_ids = await self._state_storage_controller.get_state_ids_at(
                     room_id,
                     since_token,
-                    state_filter=StateFilter.from_types([(EventTypes.Member, user_id)]),
+                    state_filter=StateFilter.from_types(
+                        [(EventTypes.Member, room_user_id)]
+                    ),
+                )
+                old_mem_ev_id = old_state_ids.get(
+                    (EventTypes.Member, room_user_id), None
                 )
-                old_mem_ev_id = old_state_ids.get((EventTypes.Member, user_id), None)
                 old_mem_ev = None
                 if old_mem_ev_id:
                     old_mem_ev = await self.store.get_event(
@@ -3076,12 +3186,12 @@ class SyncHandler:
                                 room_id,
                                 since_token,
                                 state_filter=StateFilter.from_types(
-                                    [(EventTypes.Member, user_id)]
+                                    [(EventTypes.Member, room_user_id)]
                                 ),
                             )
                         )
                         old_mem_ev_id = old_state_ids.get(
-                            (EventTypes.Member, user_id), None
+                            (EventTypes.Member, room_user_id), None
                         )
                         old_mem_ev = None
                         if old_mem_ev_id:
@@ -3094,10 +3204,8 @@ class SyncHandler:
             # Only bother if we're still currently invited
             should_invite = last_non_join.membership == Membership.INVITE
             if should_invite:
-                if (
-                    last_non_join.sender not in ignored_users
-                    and invite_config.get_invite_rule(last_non_join.sender)
-                    != InviteRule.IGNORE
+                if not await self._invite_is_ignored(
+                    room_id, last_non_join.sender, ignored_users, invite_config
                 ):
                     invite_room_sync = InvitedSyncResult(room_id, invite=last_non_join)
                     if invite_room_sync:
@@ -3255,6 +3363,14 @@ class SyncHandler:
         )
         invite_config = await self.store.get_invite_config_for_user(user_id)
 
+        # The user's identities in MSC4430 member key rooms (their keys), to tell
+        # their own leaves from kicks below.
+        member_keys_by_room: Mapping[str, str] = (
+            await self.store.get_local_member_keys_for_user(user_id)
+            if self._msc4430_enabled
+            else {}
+        )
+
         room_entries = []
         invited = []
         knocked = []
@@ -3277,9 +3393,9 @@ class SyncHandler:
                     )
                 )
             elif event.membership == Membership.INVITE:
-                if event.sender in ignored_users:
-                    continue
-                if invite_config.get_invite_rule(event.sender) == InviteRule.IGNORE:
+                if await self._invite_is_ignored(
+                    event.room_id, event.sender, ignored_users, invite_config
+                ):
                     continue
                 invite = await self.store.get_event(event.event_id)
                 invited.append(InvitedSyncResult(room_id=event.room_id, invite=invite))
@@ -3290,7 +3406,10 @@ class SyncHandler:
                 # Always send down rooms we were banned from or kicked from.
                 if not sync_config.filter_collection.include_leave:
                     if event.membership == Membership.LEAVE:
-                        if user_id == event.sender:
+                        if event.sender in (
+                            user_id,
+                            member_keys_by_room.get(event.room_id),
+                        ):
                             continue
 
                 leave_token = now_token.copy_and_replace(
diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index 03dd341744..73e8e2da20 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -43,6 +43,7 @@ from synapse.api.constants import (
 from synapse.api.room_versions import PushRuleRoomFlag
 from synapse.event_auth import auth_types_for_event, get_user_power_level
 from synapse.events import EventBase, relation_from_event
+from synapse.events.py_protocol import supports_msc4430_member_keys
 from synapse.events.snapshot import EventContext, EventPersistencePair
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.metrics import SERVER_NAME_LABEL
@@ -130,6 +131,7 @@ class BulkPushRuleEvaluator:
 
     def __init__(self, hs: "HomeServer"):
         self.hs = hs
+        self._member_key_handler = hs.get_member_key_handler()
         self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self.server_name = hs.hostname  # nb must be called this for @measure_func
@@ -169,14 +171,20 @@ class BulkPushRuleEvaluator:
             # We never notify a user about their own actions. This is enforced in
             # `_action_for_event_by_user` in the loop over `rules_by_user`, but we
             # do the same check here to avoid unnecessary DB queries.
-            if event.sender != event.state_key and self.hs.is_mine_id(event.state_key):
+            #
+            # (The target as a user: in an MSC4430 member key room the state
+            # key is their member key.)
+            target_user_id = await self._member_key_handler.local_user_id_for_identity(
+                event.room_id, event.state_key
+            )
+            if event.sender != event.state_key and target_user_id is not None:
                 # Check the target is in the room, to avoid notifying them of
                 # e.g. a pre-emptive ban.
                 target_already_in_room = await self.store.check_local_user_in_room(
-                    event.state_key, event.room_id
+                    target_user_id, event.room_id
                 )
                 if target_already_in_room:
-                    local_users = [event.state_key]
+                    local_users = [target_user_id]
         else:
             # We get the users who may need to be notified by first fetching the
             # local users currently in the room, finding those that have push rules,
@@ -202,13 +210,25 @@ class BulkPushRuleEvaluator:
             and event.type == EventTypes.Member
             and event.membership == Membership.INVITE
         ):
-            invited = event.state_key
-            invite_config = await self.store.get_invite_config_for_user(invited)
-            if invite_config.get_invite_rule(event.sender) != InviteRule.ALLOW:
-                # Invite was blocked or ignored, never notify.
-                return {}
-            if self.hs.is_mine_id(invited) and invited not in local_users:
-                local_users.append(invited)
+            # The invitee and the inviter as users (in an MSC4430 member key
+            # room, member keys; an inviter that maps to nobody cannot be
+            # matched by the invitee's rules).
+            invited = await self._member_key_handler.local_user_id_for_identity(
+                event.room_id, event.state_key
+            )
+            if invited is not None:
+                invite_config = await self.store.get_invite_config_for_user(invited)
+                inviter = await self._member_key_handler.user_id_for_identity(
+                    event.room_id, event.sender
+                )
+                if (
+                    inviter is not None
+                    and invite_config.get_invite_rule(inviter) != InviteRule.ALLOW
+                ):
+                    # Invite was blocked or ignored, never notify.
+                    return {}
+                if invited not in local_users:
+                    local_users.append(invited)
 
         if not local_users:
             return {}
@@ -484,8 +504,18 @@ class BulkPushRuleEvaluator:
                 event.room_id, thread_id
             )
 
+        # The sender as a user (their member key maps to them in an MSC4430
+        # member key room): we never notify a user about their own events.
+        sender_user_id = (
+            await self._member_key_handler.local_user_id_for_identity(
+                event.room_id, event.sender
+            )
+            if supports_msc4430_member_keys(event)
+            else event.sender
+        )
+
         for uid, rules in rules_by_user.items():
-            if event.sender == uid:
+            if uid in (event.sender, sender_user_id):
                 continue
 
             display_name = None
diff --git a/synapse/storage/databases/main/member_keys.py b/synapse/storage/databases/main/member_keys.py
index 3db59da27e..c08e68c64d 100644
--- a/synapse/storage/databases/main/member_keys.py
+++ b/synapse/storage/databases/main/member_keys.py
@@ -234,6 +234,44 @@ class MemberKeysWorkerStore(CacheInvalidationWorkerStore):
                     local_users[member_key] = local_key.user_id
         return local_users
 
+    async def get_user_ids_for_identities(
+        self, room_id: str, identities: Iterable[str]
+    ) -> dict[str, str | None]:
+        """The user behind each room member identity: a user ID is its own; a
+        member key we hold in the room maps to our user; any other member key
+        maps to the user its link claims if that link is *verified*, else to
+        `None` (an unverified claim is anyone's to make, so such a key is
+        nobody's). Member keys map to `None` when MSC4430 is off.
+
+        Rooms without member keys query nothing.
+        """
+        users: dict[str, str | None] = {}
+        member_keys: list[str] = []
+        for identity in identities:
+            if is_member_key(identity):
+                users[identity] = None
+                if self._msc4430_enabled:
+                    member_keys.append(identity)
+            else:
+                users[identity] = identity
+        if not member_keys:
+            return users
+
+        local_keys = await self.get_local_member_keys_by_key(room_id, member_keys)
+        remote_keys = []
+        for member_key in member_keys:
+            local_key = local_keys.get(member_key)
+            if local_key is not None:
+                users[member_key] = local_key.user_id
+            else:
+                remote_keys.append(member_key)
+        if remote_keys:
+            links = await self.get_member_links(room_id, remote_keys)
+            for member_key, link in links.items():
+                if link is not None:
+                    users[member_key] = link.verified_user_id
+        return users
+
     async def get_hosts_for_identities(
         self, room_id: str, identities: Iterable[str]
     ) -> dict[str, str]:
diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py
index 3a1a522dfd..0047461a18 100644
--- a/synapse/storage/databases/main/stream.py
+++ b/synapse/storage/databases/main/stream.py
@@ -942,6 +942,36 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         return ret, next_key, limited
 
     @trace
+    async def _membership_state_key_clause(
+        self, table_alias: str, user_id: str, column: str = "state_key"
+    ) -> tuple[str, list[str]]:
+        """An SQL clause matching the given user's membership rows in a table
+        whose `column` holds the membership state key as the room knows it.
+
+        In rooms without member keys that is the user ID. In an MSC4430 member
+        key room it is the user's member key for that room, so with MSC4430 on
+        the clause also matches the keys we hold for the user, each scoped to
+        its room (keys are public strings anyone can name as a state key: a key
+        of ours from another room is not this user). The keys come from the
+        cached per-user lookup and are spelled out, so every arm of the clause
+        can use the table's `(room_id, column)` or `column` index; a user with
+        no keys, and the feature being off, give the legacy clause byte for
+        byte.
+
+        The table must have a `room_id` column.
+        """
+        if not self._msc4430_enabled:
+            return f"{table_alias}.{column} = ?", [user_id]
+        member_keys_by_room = await self.get_local_member_keys_for_user(user_id)  # type: ignore[attr-defined]
+        if not member_keys_by_room:
+            return f"{table_alias}.{column} = ?", [user_id]
+        arms = [f"{table_alias}.{column} = ?"]
+        args = [user_id]
+        for room_id, member_key in member_keys_by_room.items():
+            arms.append(f"({table_alias}.room_id = ? AND {table_alias}.{column} = ?)")
+            args.extend((room_id, member_key))
+        return "(" + " OR ".join(arms) + ")", args
+
     async def get_current_state_delta_membership_changes_for_user(
         self,
         user_id: str,
@@ -1002,13 +1032,20 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             if not has_changed:
                 return []
 
+        # In an MSC4430 member key room the user's membership is under their
+        # member key, see `_membership_state_key_clause`.
+        state_key_clause, state_key_args = await self._membership_state_key_clause(
+            "s", user_id
+        )
+
         def f(txn: LoggingTransaction) -> list[CurrentStateDeltaMembership]:
             # To handle tokens with a non-empty instance_map we fetch more
             # results than necessary and then filter down
             min_from_id = from_key.stream
             max_to_id = to_key.get_max_stream_pos()
 
-            args: list[Any] = [min_from_id, max_to_id, EventTypes.Member, user_id]
+            args: list[Any] = [min_from_id, max_to_id, EventTypes.Member]
+            args.extend(state_key_args)
 
             # TODO: It would be good to assert that the `from_token`/`to_token` is >=
             # the first row in `current_state_delta_stream` for the rooms we're
@@ -1045,9 +1082,9 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
                     LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id
                 WHERE s.stream_id > ? AND s.stream_id <= ?
                     AND s.type = ?
-                    AND s.state_key = ?
+                    AND %s
                 ORDER BY s.stream_id ASC
-            """
+            """ % (state_key_clause,)
 
             txn.execute(sql, args)
 
@@ -1192,6 +1229,10 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         if excluded_room_ids is not None:
             room_ids_to_exclude = excluded_room_ids
 
+        state_key_clause, state_key_args = await self._membership_state_key_clause(
+            "s", user_id
+        )
+
         def f(txn: LoggingTransaction) -> dict[str, RoomsForUserStateReset]:
             # To handle tokens with a non-empty instance_map we fetch more
             # results than necessary and then filter down
@@ -1201,7 +1242,10 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             # This query looks at membership changes in
             # `sliding_sync_membership_snapshots` which will not include users
             # that were state reset out of rooms; so we need to look for that
-            # case in `current_state_delta_stream`.
+            # case in `current_state_delta_stream`. (The snapshots are keyed by
+            # user ID in every room; the state deltas by the room's state key,
+            # which in an MSC4430 member key room is the user's member key, see
+            # `_membership_state_key_clause`.)
             sql = """
                 SELECT
                     room_id,
@@ -1243,16 +1287,16 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
                         LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id
                     WHERE
                         s.type = ?
-                        AND s.state_key = ?
+                        AND %s
                 ) AS c
                 INNER JOIN rooms USING (room_id)
                 WHERE event_stream_ordering > ? AND event_stream_ordering <= ?
                 ORDER BY event_stream_ordering ASC
-            """
+            """ % (state_key_clause,)
 
             txn.execute(
                 sql,
-                (user_id, EventTypes.Member, user_id, min_from_id, max_to_id),
+                (user_id, EventTypes.Member, *state_key_args, min_from_id, max_to_id),
             )
 
             membership_changes: dict[str, RoomsForUserStateReset] = {}
@@ -1358,30 +1402,20 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             if not has_changed:
                 return []
 
+        # In an MSC4430 member key room, `room_memberships.user_id` is the
+        # member's state key, i.e. the user's member key for that room, see
+        # `_membership_state_key_clause`.
+        user_clause, user_args = await self._membership_state_key_clause(
+            "m", user_id, column="user_id"
+        )
+
         def f(txn: LoggingTransaction) -> list[_EventDictReturn]:
             # To handle tokens with a non-empty instance_map we fetch more
             # results than necessary and then filter down
             min_from_id = from_key.stream
             max_to_id = to_key.get_max_stream_pos()
 
-            if self._msc4430_enabled:
-                # In an MSC4430 member key room, `room_memberships.user_id` is the
-                # member's state key, i.e. the user's member key for that room,
-                # so the user's membership events there are the ones under the
-                # keys we hold for them. Scoped to the event's room: a key of
-                # ours from another room appearing as a state key here is not
-                # this user (keys are public strings anyone can name).
-                user_clause = """(
-                        m.user_id = ?
-                        OR m.user_id IN (
-                            SELECT k.member_key FROM msc4430_member_keys AS k
-                            WHERE k.room_id = m.room_id AND k.user_id = ?
-                        )
-                    )"""
-                args: list[Any] = [user_id, user_id, min_from_id, max_to_id]
-            else:
-                user_clause = "m.user_id = ?"
-                args = [user_id, min_from_id, max_to_id]
+            args: list[Any] = [*user_args, min_from_id, max_to_id]
 
             ignore_room_clause = ""
             if excluded_rooms is not None and len(excluded_rooms) > 0:
diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi
index 2633adf6f7..44412d86ee 100644
--- a/synapse/synapse_rust/events.pyi
+++ b/synapse/synapse_rust/events.pyi
@@ -372,6 +372,7 @@ class SerializeEventConfig:
         include_admin_metadata: bool,
         msc4354_enabled: bool,
         msc4428_enabled: bool,
+        requester_member_keys: list[str] | None = None,
     ) -> None: ...
     @property
     def as_client_event(self) -> bool:
@@ -418,6 +419,13 @@ class SerializeEventConfig:
         When ``True``, the sender is repeated in ``unsigned`` as
         ``org.matrix.msc4428.stable_id``."""
 
+    @property
+    def requester_member_keys(self) -> set[str] | None:
+        """MSC4430: the requester's member keys. In a member key room the
+        ``sender`` of the requester's own events is their key in that room, so
+        sender-only fields (``transaction_id``, ``delay_id``) are gated on these
+        as well as on ``requester.user_id``."""
+
 def serialize_events(
     events: list[tuple[Event, str | None]],
     time_now_ms: int,
diff --git a/synapse/visibility.py b/synapse/visibility.py
index 4eeba14a6c..663eb745c6 100644
--- a/synapse/visibility.py
+++ b/synapse/visibility.py
@@ -24,6 +24,7 @@ from enum import Enum, auto
 from typing import (
     Collection,
     Final,
+    Mapping,
     Sequence,
 )
 
@@ -36,6 +37,7 @@ from synapse.api.constants import (
     Membership,
 )
 from synapse.events import EventBase
+from synapse.events.py_protocol import supports_msc4430_member_keys
 from synapse.events.snapshot import EventContext
 from synapse.events.utils import FilteredEvent, prune_event
 from synapse.logging.opentracing import trace
@@ -151,6 +153,31 @@ async def filter_and_transform_events_for_client(
             )
 
     types = [_HISTORY_VIS_KEY, (EventTypes.Member, user_id)]
+
+    # MSC4430: in a member key room the user's membership is under their member
+    # key in that room, and the senders are keys which the checks below (ignored
+    # senders, erased senders) must see as the users behind them.
+    member_key_room_ids = {e.room_id for e in events if supports_msc4430_member_keys(e)}
+    user_room_identities: Mapping[str, str] = {}
+    # (room ID, sender) -> the user behind the sender: a key is only a user in
+    # the room it was looked up in.
+    sender_user_ids: dict[tuple[str, str], str | None] = {}
+    if member_key_room_ids:
+        user_room_identities = await storage.main.get_local_member_keys_for_user(
+            user_id
+        )
+        types.extend(
+            (EventTypes.Member, member_key)
+            for room_id, member_key in user_room_identities.items()
+            if room_id in member_key_room_ids
+        )
+        for room_id in member_key_room_ids:
+            mapped = await storage.main.get_user_ids_for_identities(
+                room_id, {e.sender for e in events if e.room_id == room_id}
+            )
+            for sender, sender_user in mapped.items():
+                sender_user_ids[(room_id, sender)] = sender_user
+
     if found_call_invite:
         # We need to fetch the room's join rules state to determine
         # whether to allow call invites in public rooms.
@@ -165,7 +192,16 @@ async def filter_and_transform_events_for_client(
     # Get the users who are ignored by the requesting user.
     ignore_list = await storage.main.ignored_users(user_id)
 
-    erased_senders = await storage.main.are_users_erased(e.sender for e in events)
+    def sender_user_id(event: EventBase) -> str | None:
+        """The user behind the event's sender, for the per-user checks: the
+        sender itself in rooms without member keys."""
+        if event.room_id in member_key_room_ids:
+            return sender_user_ids.get((event.room_id, event.sender))
+        return event.sender
+
+    erased_senders = await storage.main.are_users_erased(
+        {u for u in (sender_user_id(e) for e in events) if u is not None}
+    )
 
     if filter_send_to_client:
         room_ids = {e.room_id for e in events}
@@ -178,17 +214,27 @@ async def filter_and_transform_events_for_client(
 
     def allowed(event: EventBase) -> FilteredEvent | None:
         state_after_event = event_id_to_state.get(event.event_id)
+        # The user as this room knows them: their member key, in a member key
+        # room where they have one.
+        room_user_id = (
+            user_room_identities.get(event.room_id, user_id)
+            if event.room_id in member_key_room_ids
+            else user_id
+        )
+        sender = sender_user_id(event)
         filtered = _check_client_allowed_to_see_event(
-            user_id=user_id,
+            user_id=room_user_id,
             event=event,
             clock=storage.main.clock,
             filter_send_to_client=filter_send_to_client,
-            sender_ignored=event.sender in ignore_list,
+            # A member key may be ignored as such (it may be the only handle a
+            # client has on it) or as the user behind it.
+            sender_ignored=event.sender in ignore_list or sender in ignore_list,
             always_include_ids=always_include_ids,
             retention_policy=retention_policies[event.room_id],
             state=state_after_event,
             is_peeking=is_peeking,
-            sender_erased=erased_senders.get(event.sender, False),
+            sender_erased=erased_senders.get(sender, False) if sender else False,
         )
         if filtered is None:
             return None
@@ -219,10 +265,12 @@ async def filter_and_transform_events_for_client(
         # user's own membership event, so we can just inspect that.
 
         user_membership_event: EventBase | None
-        if event.type == EventTypes.Member and event.state_key == user_id:
+        if event.type == EventTypes.Member and event.state_key == room_user_id:
             user_membership_event = event
         elif state_after_event is not None:
-            user_membership_event = state_after_event.get((EventTypes.Member, user_id))
+            user_membership_event = state_after_event.get(
+                (EventTypes.Member, room_user_id)
+            )
         else:
             # unreachable!
             raise Exception("Missing state for event that is not user's own membership")
@@ -286,7 +334,14 @@ async def filter_event_for_clients_with_state(
     # see events in the room at that point in the DAG, and that shouldn't be decided
     # on those checks.
     if filter_send_to_client:
-        ignored_by = await store.ignored_by(event.sender)
+        ignored_by = set(await store.ignored_by(event.sender))
+        if supports_msc4430_member_keys(event):
+            # A member key may be ignored as such, or as the user behind it.
+            sender_user_id = (
+                await store.get_user_ids_for_identities(event.room_id, [event.sender])
+            )[event.sender]
+            if sender_user_id is not None:
+                ignored_by.update(await store.ignored_by(sender_user_id))
         retention_policy = await store.get_retention_policy_for_room(event.room_id)
 
         for user_id in user_ids:
@@ -301,13 +356,28 @@ async def filter_event_for_clients_with_state(
             ):
                 allowed_user_ids.discard(user_id)
 
+    # MSC4430: in a member key room each user's membership is under their member
+    # key in that room (a user with no key there is under their user ID, which
+    # the state then does not hold).
+    room_identities: dict[str, str] = {u: u for u in allowed_user_ids}
+    if supports_msc4430_member_keys(event):
+        local_member_keys = await store.get_local_member_keys_in_room(event.room_id)
+        member_key_by_user_id = {
+            local_user_id: member_key
+            for member_key, local_user_id in local_member_keys.items()
+        }
+        for user_id in allowed_user_ids:
+            room_identities[user_id] = member_key_by_user_id.get(user_id, user_id)
+
     if event.internal_metadata.outlier:
         # Normally these can't be seen by clients, but we make an exception for
         # for out-of-band membership events (eg, incoming invites, or rejections of
         # said invite) for the user themselves.
-        if event.type == EventTypes.Member and event.state_key in allowed_user_ids:
-            logger.debug("Returning out-of-band-membership event %s", event)
-            return {event.state_key}
+        if event.type == EventTypes.Member:
+            for user_id in allowed_user_ids:
+                if room_identities[user_id] == event.state_key:
+                    logger.debug("Returning out-of-band-membership event %s", event)
+                    return {user_id}
 
         return set()
 
@@ -329,7 +399,7 @@ async def filter_event_for_clients_with_state(
 
     filter_list = []
     for user_id in allowed_user_ids:
-        filter_list.append((EventTypes.Member, user_id))
+        filter_list.append((EventTypes.Member, room_identities[user_id]))
     filter_list.append((EventTypes.RoomHistoryVisibility, ""))
 
     state_filter = StateFilter.from_types(filter_list)
@@ -339,7 +409,9 @@ async def filter_event_for_clients_with_state(
     return {
         user_id
         for user_id in allowed_user_ids
-        if _check_membership(user_id, event, visibility, state_map, is_peeking).allowed
+        if _check_membership(
+            room_identities[user_id], event, visibility, state_map, is_peeking
+        ).allowed
     }
 
 
diff --git a/tests/rest/client/test_member_key_rooms.py b/tests/rest/client/test_member_key_rooms.py
index 80d23b50a1..37ddf44466 100644
--- a/tests/rest/client/test_member_key_rooms.py
+++ b/tests/rest/client/test_member_key_rooms.py
@@ -23,22 +23,38 @@ client is served (`/sync`, `/event`, `/members`) is a later change, so the room
 inspected through `/state` and the store.
 """
 
+import json
+import urllib.parse
 from http import HTTPStatus
 from unittest.mock import Mock
 
 from twisted.internet.testing import MemoryReactor
 
-from synapse.api.constants import EventContentFields, EventTypes, MemberKeys, Membership
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    MemberInfo,
+    MemberKeys,
+    Membership,
+)
 from synapse.api.room_versions import RoomVersions
 from synapse.events import EventBase
 from synapse.rest import admin
-from synapse.rest.client import login, room, room_upgrade_rest_servlet
+from synapse.rest.client import (
+    account_data,
+    login,
+    room,
+    room_upgrade_rest_servlet,
+    sync,
+)
 from synapse.server import HomeServer
+from synapse.storage.member_keys import MemberLink
 from synapse.types import JsonDict, is_member_key
 from synapse.util.clock import Clock
 
 from tests import unittest
-from tests.test_utils.member_key_rooms import Member
+from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
+from tests.test_utils.member_key_rooms import Member, MemberKeyRoom
 from tests.unittest import override_config
 
 ROOM_VERSION = RoomVersions.MSC4430v12
@@ -422,3 +438,460 @@ class MemberKeyRoomsTestCase(unittest.HomeserverTestCase):
             room_version=ROOM_VERSION.identifier,
             expect_code=HTTPStatus.BAD_REQUEST,
         )
+
+
+class MemberKeyClientViewTestCase(unittest.HomeserverTestCase):
+    """What a client is served from a member key room: events whose senders are
+    keys with `stable_id` set, a `member_info` map resolving the keys, the
+    requester's own events recognised for the transaction ID echo, visibility
+    decided by the requester's key, and `/members` and `/joined_members` keyed
+    by key. These are the local halves of the Complement tests for the stable
+    ID, `member_info` in `/sync`, the member list endpoints, the transaction ID
+    echo and the incremental sync after a join."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+        account_data.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config.setdefault("experimental_features", {}).setdefault(
+            "msc4430_enabled", True
+        )
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = hs.get_datastores().main
+        self.alice = self.register_user("alice", "pass")
+        self.alice_tok = self.login("alice", "pass")
+        self.bob = self.register_user("bob", "pass")
+        self.bob_tok = self.login("bob", "pass")
+        self.room_id = self.helper.create_room_as(
+            self.alice,
+            tok=self.alice_tok,
+            room_version=ROOM_VERSION.identifier,
+            extra_content={"preset": "public_chat"},
+        )
+        self.alice_key = self._key(self.alice)
+
+    def _key(self, user_id: str) -> str:
+        local_key = self.get_success(
+            self.store.get_local_member_key(self.room_id, user_id)
+        )
+        assert local_key is not None
+        return local_key.member_key
+
+    def _sync(self, tok: str, since: str | None = None, **params: str) -> JsonDict:
+        query = {**params}
+        if since is not None:
+            query["since"] = since
+        qs = "&".join(f"{k}={v}" for k, v in query.items())
+        channel = self.make_request(
+            "GET", f"/sync?{qs}" if qs else "/sync", access_token=tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        return channel.json_body
+
+    def _joined_room(self, body: JsonDict) -> JsonDict:
+        return body["rooms"]["join"][self.room_id]
+
+    def _stranger_join(self, stranger: Member) -> EventBase:
+        """A join by a key this server does not hold, as federation would bring."""
+        builder = self.get_success(
+            MemberKeyRoom.for_existing_room(self.hs, self.room_id)
+        )
+        return self.get_success(builder.send_member(stranger, "join"))
+
+    def test_initial_sync(self) -> None:
+        """Every event names a key as sender and stable ID; the map resolves
+        every member's key to their user ID."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        self.helper.send(self.room_id, body="hello", tok=self.bob_tok)
+
+        room_body = self._joined_room(self._sync(self.alice_tok))
+        events = room_body["timeline"]["events"] + room_body["state"]["events"]
+        self.assertTrue(events)
+        for event in events:
+            self.assertTrue(is_member_key(event["sender"]), event)
+            self.assertEqual(
+                event["unsigned"]["org.matrix.msc4428.stable_id"], event["sender"]
+            )
+            if event["type"] == EventTypes.Member:
+                self.assertTrue(is_member_key(event["state_key"]), event)
+        self.assertEqual(
+            room_body[MemberInfo.FIELD],
+            {
+                self.alice_key: {MemberInfo.USER_ID: self.alice},
+                bob_key: {MemberInfo.USER_ID: self.bob},
+            },
+        )
+        # The user's own membership is served as state under their key.
+        memberships = {
+            e["state_key"]: e["content"]["membership"]
+            for e in events
+            if e["type"] == EventTypes.Member
+        }
+        self.assertEqual(memberships, {self.alice_key: "join", bob_key: "join"})
+
+    def test_incremental_sync_after_join_and_leave(self) -> None:
+        """A room joined after the last sync comes down with its create event,
+        and only the new member's entry; leaving puts it under `leave`."""
+        since = self._sync(self.bob_tok)["next_batch"]
+        self.assertNotIn(
+            self.room_id, self._sync(self.bob_tok).get("rooms", {}).get("join", {})
+        )
+
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        body = self._sync(self.bob_tok, since=since)
+        room_body = self._joined_room(body)
+        types = {e["type"] for e in room_body["timeline"]["events"]} | {
+            e["type"] for e in room_body["state"]["events"]
+        }
+        self.assertIn(EventTypes.Create, types)
+        self.assertEqual(
+            room_body[MemberInfo.FIELD][bob_key], {MemberInfo.USER_ID: self.bob}
+        )
+
+        # Alice, already synced, learns only of Bob.
+        alice_since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.send(self.room_id, body="hi", tok=self.bob_tok)
+        room_body = self._joined_room(self._sync(self.alice_tok, since=alice_since))
+        self.assertEqual(
+            room_body[MemberInfo.FIELD], {bob_key: {MemberInfo.USER_ID: self.bob}}
+        )
+
+        since = body["next_batch"]
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        body = self._sync(self.bob_tok, since=since)
+        self.assertIn(self.room_id, body["rooms"]["leave"])
+        self.assertNotIn(self.room_id, body["rooms"].get("join", {}))
+
+    def test_transaction_id_echoed_to_the_sending_device_only(self) -> None:
+        other_device_tok = self.login("alice", "pass")
+        event_id = self.helper.send(
+            self.room_id, body="hello", tok=self.alice_tok, txn_id="txn1"
+        )["event_id"]
+
+        def find(body: JsonDict) -> JsonDict:
+            for event in self._joined_room(body)["timeline"]["events"]:
+                if event["event_id"] == event_id:
+                    return event
+            raise AssertionError("event not in timeline")
+
+        self.assertEqual(
+            find(self._sync(self.alice_tok))["unsigned"]["transaction_id"], "txn1"
+        )
+        self.assertNotIn(
+            "transaction_id", find(self._sync(other_device_tok))["unsigned"]
+        )
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        self.assertNotIn("transaction_id", find(self._sync(self.bob_tok))["unsigned"])
+
+    def test_members_and_joined_members(self) -> None:
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+
+        channel = self.make_request(
+            "GET", f"/rooms/{self.room_id}/members", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.assertEqual(
+            {e["state_key"] for e in channel.json_body["chunk"]},
+            {self.alice_key, bob_key},
+        )
+        self.assertEqual(
+            channel.json_body[MemberInfo.FIELD],
+            {
+                self.alice_key: {MemberInfo.USER_ID: self.alice},
+                bob_key: {MemberInfo.USER_ID: self.bob},
+            },
+        )
+
+        channel = self.make_request(
+            "GET", f"/rooms/{self.room_id}/joined_members", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        joined = channel.json_body["joined"]
+        self.assertEqual(set(joined), {self.alice_key, bob_key})
+        self.assertEqual(joined[bob_key][MemberInfo.USER_ID], self.bob)
+        self.assertEqual(joined[bob_key]["display_name"], "bob")
+
+    def test_event_and_messages_are_visible_to_members_only(self) -> None:
+        """Visibility is decided by the requester's key: with `joined` history,
+        a later joiner sees nothing from before their join."""
+        self.helper.send_state(
+            self.room_id,
+            EventTypes.RoomHistoryVisibility,
+            {"history_visibility": "joined"},
+            tok=self.alice_tok,
+        )
+        before = self.helper.send(self.room_id, body="before", tok=self.alice_tok)[
+            "event_id"
+        ]
+        self.helper.get_event(self.room_id, before, tok=self.alice_tok)
+        self.helper.get_event(
+            self.room_id, before, tok=self.bob_tok, expect_code=HTTPStatus.NOT_FOUND
+        )
+
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        after = self.helper.send(self.room_id, body="after", tok=self.alice_tok)[
+            "event_id"
+        ]
+        self.helper.get_event(
+            self.room_id, before, tok=self.bob_tok, expect_code=HTTPStatus.NOT_FOUND
+        )
+        self.helper.get_event(self.room_id, after, tok=self.bob_tok)
+
+        channel = self.make_request(
+            "GET",
+            f"/rooms/{self.room_id}/messages?dir=b",
+            access_token=self.bob_tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        bodies = [
+            e["content"]["body"]
+            for e in channel.json_body["chunk"]
+            if e["type"] == EventTypes.Message
+        ]
+        self.assertEqual(bodies, ["after"])
+
+    def test_member_info_for_erased_and_unknown_members(self) -> None:
+        """An erased member maps to null, as does a key nobody vouches for; a
+        verified link names the user it claims."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        self.get_success(self.store.mark_user_erased(self.bob))
+
+        stranger, linked = Member(), Member()
+        self._stranger_join(stranger)
+        self._stranger_join(linked)
+        self.get_success(
+            self.store.upsert_member_link(
+                MemberLink(
+                    room_id=self.room_id,
+                    member_key=linked.key,
+                    domain="other.example",
+                    localpart="carol",
+                    verified=True,
+                    event_id="$verified",
+                    key_id="ed25519:x",
+                )
+            )
+        )
+
+        channel = self.make_request(
+            "GET", f"/rooms/{self.room_id}/members", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.assertEqual(
+            channel.json_body[MemberInfo.FIELD],
+            {
+                self.alice_key: {MemberInfo.USER_ID: self.alice},
+                bob_key: {MemberInfo.USER_ID: None},
+                stranger.key: {MemberInfo.USER_ID: None},
+                linked.key: {MemberInfo.USER_ID: "@carol:other.example"},
+            },
+        )
+
+    def _ignore(self, tok: str, user_id: str, *ignored: str) -> None:
+        channel = self.make_request(
+            "PUT",
+            f"/user/{user_id}/account_data/m.ignored_user_list",
+            {"ignored_users": {i: {} for i in ignored}},
+            access_token=tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+    def test_profile_change_is_not_a_rejoin(self) -> None:
+        """Changing one's own membership (a display name) is an ordinary
+        timeline event on the next sync, not the room sent again as newly
+        joined; and the room's own leave is a leave, not a kick."""
+        since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.send_state(
+            self.room_id,
+            EventTypes.Member,
+            {"membership": Membership.JOIN, "displayname": "Alice!"},
+            tok=self.alice_tok,
+            state_key=self.alice,
+        )
+        room_body = self._joined_room(self._sync(self.alice_tok, since=since))
+        self.assertEqual(
+            [e["type"] for e in room_body["timeline"]["events"]], [EventTypes.Member]
+        )
+        self.assertFalse(room_body["timeline"]["limited"])
+        self.assertEqual(room_body["state"]["events"], [])
+
+    def test_device_lists_and_presence_name_users(self) -> None:
+        """Device list changes and presence are about users, not keys."""
+        since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        body = self._sync(self.alice_tok, since=since)
+        self.assertEqual(body["device_lists"]["changed"], [self.bob])
+        self.assertEqual({e["sender"] for e in body["presence"]["events"]}, {self.bob})
+
+        since = body["next_batch"]
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        body = self._sync(self.alice_tok, since=since)
+        self.assertEqual(body["device_lists"]["left"], [self.bob])
+
+    def test_erasure_changes_the_keys_entry(self) -> None:
+        """Erasure is delivered against the member's key, the stable ID in
+        these rooms, not against their user ID."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        since = self._sync(self.alice_tok)["next_batch"]
+
+        self.get_success(self.store.mark_user_erased(self.bob))
+        self.get_success(self.hs.get_member_info_handler().on_user_erased(self.bob))
+        room_body = self._joined_room(self._sync(self.alice_tok, since=since))
+        self.assertEqual(
+            room_body[MemberInfo.FIELD], {bob_key: {MemberInfo.USER_ID: None}}
+        )
+
+    def test_ignored_senders_by_user_id_and_by_key(self) -> None:
+        """A member may be ignored as the user behind their key or as the key
+        itself; either way their events and notifications stop."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        alice_key = self.alice_key
+
+        for ignored in (self.alice, alice_key):
+            self._ignore(self.bob_tok, self.bob, ignored)
+            since = self._sync(self.bob_tok)["next_batch"]
+            self.helper.send(self.room_id, body="ignored?", tok=self.alice_tok)
+            body = self._sync(self.bob_tok, since=since)
+            room_body = body.get("rooms", {}).get("join", {}).get(self.room_id, {})
+            self.assertEqual(room_body.get("timeline", {}).get("events", []), [])
+            self.assertEqual(
+                room_body.get("unread_notifications", {}).get("notification_count", 0),
+                0,
+            )
+            self._ignore(self.bob_tok, self.bob)
+
+    def test_invite_from_ignored_user_is_hidden(self) -> None:
+        self._ignore(self.bob_tok, self.bob, self.alice)
+        self.helper.invite(
+            self.room_id, src=self.alice, targ=self.bob, tok=self.alice_tok
+        )
+        body = self._sync(self.bob_tok)
+        self.assertNotIn(self.room_id, body.get("rooms", {}).get("invite", {}))
+
+    def test_own_events_are_not_notified(self) -> None:
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.send(self.room_id, body="mine", tok=self.alice_tok)
+        room_body = self._joined_room(self._sync(self.alice_tok, since=since))
+        self.assertEqual(room_body["unread_notifications"]["notification_count"], 0)
+
+    def test_heroes_exclude_the_requester(self) -> None:
+        """A nameless room is summarised by the other members' keys (the
+        summary is sent to clients that lazy-load members)."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        lazy = urllib.parse.quote(
+            json.dumps({"room": {"state": {"lazy_load_members": True}}})
+        )
+        summary = self._joined_room(self._sync(self.alice_tok, filter=lazy))["summary"]
+        self.assertEqual(summary["m.heroes"], [bob_key])
+        summary = self._joined_room(self._sync(self.bob_tok, filter=lazy))["summary"]
+        self.assertEqual(summary["m.heroes"], [self.alice_key])
+
+    def test_notification_count(self) -> None:
+        """Push evaluation decides who can see the event by their key."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        since = self._sync(self.bob_tok)["next_batch"]
+        self.helper.send(self.room_id, body="ping", tok=self.alice_tok)
+        room_body = self._joined_room(self._sync(self.bob_tok, since=since))
+        self.assertEqual(room_body["unread_notifications"]["notification_count"], 1)
+
+
+class MemberKeySlidingSyncTestCase(SlidingSyncBase):
+    """Sliding sync in a member key room: `$ME` is the requester's key, the room
+    list follows joins and leaves, and `member_info` resolves the keys."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+        account_data.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config["experimental_features"]["msc4430_enabled"] = True
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self.store = hs.get_datastores().main
+        self.alice = self.register_user("alice", "pass")
+        self.alice_tok = self.login("alice", "pass")
+        self.bob = self.register_user("bob", "pass")
+        self.bob_tok = self.login("bob", "pass")
+        self.room_id = self.helper.create_room_as(
+            self.alice,
+            tok=self.alice_tok,
+            room_version=ROOM_VERSION.identifier,
+            extra_content={"preset": "public_chat"},
+        )
+        self.sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [[EventTypes.Member, "$ME"]],
+                    "timeline_limit": 2,
+                }
+            }
+        }
+
+    def _key(self, user_id: str) -> str:
+        local_key = self.get_success(
+            self.store.get_local_member_key(self.room_id, user_id)
+        )
+        assert local_key is not None
+        return local_key.member_key
+
+    def test_invite_from_ignored_user_is_hidden(self) -> None:
+        channel = self.make_request(
+            "PUT",
+            f"/user/{self.bob}/account_data/m.ignored_user_list",
+            {"ignored_users": {self.alice: {}}},
+            access_token=self.bob_tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.helper.invite(
+            self.room_id, src=self.alice, targ=self.bob, tok=self.alice_tok
+        )
+        body, _ = self.do_sync(self.sync_body, tok=self.bob_tok)
+        self.assertNotIn(self.room_id, body["rooms"])
+
+    def test_me_join_and_leave(self) -> None:
+        body, pos = self.do_sync(self.sync_body, tok=self.bob_tok)
+        self.assertNotIn(self.room_id, body["rooms"])
+
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.bob_tok)
+        room_body = body["rooms"][self.room_id]
+        self.assertTrue(room_body["initial"])
+        # `$ME` is Bob's key: his own membership, and nobody else's.
+        self.assertEqual(
+            [(e["type"], e["state_key"]) for e in room_body["required_state"]],
+            [(EventTypes.Member, bob_key)],
+        )
+        self.assertEqual(
+            room_body[MemberInfo.FIELD][bob_key], {MemberInfo.USER_ID: self.bob}
+        )
+
+        # Leaving on one's own removes the room from the list (it is not a kick).
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.bob_tok)
+        self.assertIn(self.room_id, body["rooms"])  # newly left: delivered once
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.bob_tok)
+        self.assertNotIn(self.room_id, body["rooms"])
diff --git a/tests/storage/test_member_key_persistence.py b/tests/storage/test_member_key_persistence.py
index 922cc51f59..65364b75b1 100644
--- a/tests/storage/test_member_key_persistence.py
+++ b/tests/storage/test_member_key_persistence.py
@@ -268,6 +268,54 @@ class MemberKeyPersistenceTestCase(unittest.HomeserverTestCase):
             self.get_success(self.store.get_rooms_for_user(self.bob)), frozenset()
         )
 
+    def test_state_delta_membership_changes_for_user(self) -> None:
+        """The two queries over `current_state_delta_stream` find the user's
+        changes under their key: a join, a kick by another key, and nothing
+        for a key of ours from another room."""
+        bob_member = self._local_key(self.bob)
+        stranger = Member()
+        self.get_success(self.room.send_member(self.alice_member, "join"))
+        before = self.store.get_room_max_token()
+        self.get_success(self.room.send_member(bob_member, "join"))
+        kick = self.get_success(
+            self.room.send_member(bob_member, "leave", sender=self.alice_member)
+        )
+        self.get_success(
+            self.store.create_local_member_key(
+                OTHER_ROOM, "@carol:test", stranger.signing_key
+            )
+        )
+        self.get_success(self.room.send_member(stranger, "join"))
+        after = self.store.get_room_max_token()
+
+        changes = self.get_success(
+            self.store.get_current_state_delta_membership_changes_for_user(
+                self.bob, before, after
+            )
+        )
+        self.assertEqual(
+            [(c.room_id, c.membership, c.sender) for c in changes],
+            [
+                (self.room_id, Membership.JOIN, bob_member.key),
+                (self.room_id, Membership.LEAVE, self.alice_member.key),
+            ],
+        )
+        sliding = self.get_success(
+            self.store.get_sliding_sync_membership_changes(self.bob, before, after)
+        )
+        self.assertEqual(
+            {room_id: (c.membership, c.event_id) for room_id, c in sliding.items()},
+            {self.room_id: (Membership.LEAVE, kick.event_id)},
+        )
+        self.assertEqual(
+            self.get_success(
+                self.store.get_current_state_delta_membership_changes_for_user(
+                    "@carol:test", before, after
+                )
+            ),
+            [],
+        )
+
     def test_leave_removes_membership(self) -> None:
         """A leave updates the user's row, and when the leaver was our last
         member the server leaves the room (which depends on recognising the
diff --git a/tests/test_utils/member_key_rooms.py b/tests/test_utils/member_key_rooms.py
index fb0b2b5529..f4db0e47dc 100644
--- a/tests/test_utils/member_key_rooms.py
+++ b/tests/test_utils/member_key_rooms.py
@@ -202,6 +202,25 @@ class MemberKeyRoom:
             self._depth = max(self._depth, event.depth)
         return persisted
 
+    @classmethod
+    async def for_existing_room(
+        cls, hs: HomeServer, room_id: str, creator: Member | None = None
+    ) -> "MemberKeyRoom":
+        """Attach to a member key room that already exists (one created through
+        the client API, say), chaining new events onto its current extremities.
+        `creator` is only a label: the events sent through the room are signed
+        by whichever `Member` they name."""
+        room = cls(hs, creator or Member())
+        room.room_id = room_id
+        store = hs.get_datastores().main
+        prev_events = await store.get_prev_events_for_room(room_id)
+        room._latest_event_id = prev_events[0]
+        state_extremities = list(await store.get_state_dag_extremities(room_id))
+        room._latest_state_event_id = state_extremities[0]
+        _, depth = await store.get_max_depth_of(prev_events)
+        room._depth = depth
+        return room
+
     async def create(self, content: JsonDict | None = None) -> EventBase:
         """Store the `rooms` row and persist the create event, sent by the
         creator's member key. The room ID is the create event's ID."""