diff --git a/changelog.d/19487.feature b/changelog.d/19487.feature
new file mode 100644
index 00000000000..4eb1d8f2617
--- /dev/null
+++ b/changelog.d/19487.feature
@@ -0,0 +1 @@
+Expose [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) over the legacy (v3) /sync API.
\ No newline at end of file
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index a32748c2a9c..c8ef5e2aa6c 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -37,6 +37,7 @@
     EventContentFields,
     EventTypes,
     Membership,
+    StickyEvent,
 )
 from synapse.api.filtering import FilterCollection
 from synapse.api.presence import UserPresenceState
@@ -147,6 +148,7 @@ class JoinedSyncResult:
     state: StateMap[EventBase]
     ephemeral: list[JsonDict]
     account_data: list[JsonDict]
+    sticky: list[EventBase]
     unread_notifications: JsonDict
     unread_thread_notifications: JsonDict
     summary: JsonDict | None
@@ -157,7 +159,11 @@ def __bool__(self) -> bool:
         to tell if room needs to be part of the sync result.
         """
         return bool(
-            self.timeline or self.state or self.ephemeral or self.account_data
+            self.timeline
+            or self.state
+            or self.ephemeral
+            or self.account_data
+            or self.sticky
             # nb the notification count does not, er, count: if there's nothing
             # else in the result, we don't need to send it.
         )
@@ -601,6 +607,41 @@ async def ephemeral_by_room(
 
         return now_token, ephemeral_by_room
 
+    async def sticky_events_by_room(
+        self,
+        sync_result_builder: "SyncResultBuilder",
+        now_token: StreamToken,
+        since_token: StreamToken | None = None,
+    ) -> tuple[StreamToken, dict[str, list[str]]]:
+        """Get the sticky events for each room the user is in
+        Args:
+            sync_result_builder
+            now_token: Where the server is currently up to.
+            since_token: Where the server was when the client last synced.
+        Returns:
+            A tuple of the now StreamToken, updated to reflect the which sticky
+            events are included, and a dict mapping from room_id to a list
+            of sticky event IDs for that room (in sticky event stream order).
+        """
+        now = self.clock.time_msec()
+        with Measure(
+            self.clock, name="sticky_events_by_room", server_name=self.server_name
+        ):
+            from_id = since_token.sticky_events_key if since_token else 0
+
+            room_ids = sync_result_builder.joined_room_ids
+
+            to_id, sticky_by_room = await self.store.get_sticky_events_in_rooms(
+                room_ids,
+                from_id=from_id,
+                to_id=now_token.sticky_events_key,
+                now=now,
+                limit=StickyEvent.MAX_EVENTS_IN_SYNC,
+            )
+            now_token = now_token.copy_and_replace(StreamKeyType.STICKY_EVENTS, to_id)
+
+        return now_token, sticky_by_room
+
     async def _load_filtered_recents(
         self,
         room_id: str,
@@ -2177,11 +2218,43 @@ async def _generate_sync_entry_for_rooms(
             )
             sync_result_builder.now_token = now_token
 
+        sticky_by_room: dict[str, list[str]] = {}
+        if self.hs_config.experimental.msc4354_enabled:
+            now_token, sticky_by_room = await self.sticky_events_by_room(
+                sync_result_builder, now_token, since_token
+            )
+            sync_result_builder.now_token = now_token
+
         # 2. We check up front if anything has changed, if it hasn't then there is
         # no point in going further.
+        #
+        # If this is an initial sync (no since_token), then of course we can't skip
+        # the sync entry, as we have no base to use as a comparison for the question
+        # 'has anything changed' (this is the client's first time 'seeing' anything).
+        #
+        # Otherwise, for incremental syncs, we consider skipping the sync entry,
+        # doing cheap checks first:
+        #
+        # - are there any per-room EDUs;
+        # - is there any Room Account Data; or
+        # - are there any sticky events in the rooms; or
+        # - might the rooms have changed
+        #   (using in-memory event stream change caches, which can
+        #   only answer either 'Not changed' or 'Possibly changed')
+        #
+        # If none of those cheap checks give us a reason to continue generating the sync entry,
+        # we finally query the database to check for changed room tags.
+        # If there are also no changed tags, we can short-circuit return an empty sync entry.
         if not sync_result_builder.full_state:
-            if since_token and not ephemeral_by_room and not account_data_by_room:
-                have_changed = await self._have_rooms_changed(sync_result_builder)
+            # Cheap checks first
+            if (
+                since_token
+                and not ephemeral_by_room
+                and not account_data_by_room
+                and not sticky_by_room
+            ):
+                # This is also a cheap check, but we log the answer
+                have_changed = self._may_have_rooms_changed(sync_result_builder)
                 log_kv({"rooms_have_changed": have_changed})
                 if not have_changed:
                     tags_by_room = await self.store.get_updated_tags(
@@ -2225,6 +2298,7 @@ async def handle_room_entries(room_entry: "RoomSyncResultBuilder") -> None:
                 ephemeral=ephemeral_by_room.get(room_entry.room_id, []),
                 tags=tags_by_room.get(room_entry.room_id),
                 account_data=account_data_by_room.get(room_entry.room_id, {}),
+                sticky_event_ids=sticky_by_room.get(room_entry.room_id, []),
                 always_include=sync_result_builder.full_state,
             )
             logger.debug("Generated room entry for %s", room_entry.room_id)
@@ -2237,11 +2311,9 @@ async def handle_room_entries(room_entry: "RoomSyncResultBuilder") -> None:
 
         return set(newly_joined_rooms), set(newly_left_rooms)
 
-    async def _have_rooms_changed(
-        self, sync_result_builder: "SyncResultBuilder"
-    ) -> bool:
+    def _may_have_rooms_changed(self, sync_result_builder: "SyncResultBuilder") -> bool:
         """Returns whether there may be any new events that should be sent down
-        the sync. Returns True if there are.
+        the sync. Returns True if there **may** be.
 
         Does not modify the `sync_result_builder`.
         """
@@ -2611,6 +2683,7 @@ async def _generate_room_entry(
         ephemeral: list[JsonDict],
         tags: Mapping[str, JsonMapping] | None,
         account_data: Mapping[str, JsonMapping],
+        sticky_event_ids: list[str],
         always_include: bool = False,
     ) -> None:
         """Populates the `joined` and `archived` section of `sync_result_builder`
@@ -2640,6 +2713,8 @@ async def _generate_room_entry(
             tags: List of *all* tags for room, or None if there has been
                 no change.
             account_data: List of new account data for room
+            sticky_event_ids: MSC4354 sticky events in the room, if any.
+                In sticky event stream order.
             always_include: Always include this room in the sync response,
                 even if empty.
         """
@@ -2650,7 +2725,13 @@ async def _generate_room_entry(
         events = room_builder.events
 
         # We want to shortcut out as early as possible.
-        if not (always_include or account_data or ephemeral or full_state):
+        if not (
+            always_include
+            or account_data
+            or ephemeral
+            or full_state
+            or sticky_event_ids
+        ):
             if events == [] and tags is None:
                 return
 
@@ -2742,6 +2823,7 @@ async def _generate_room_entry(
                 or account_data_events
                 or ephemeral
                 or full_state
+                or sticky_event_ids
             ):
                 return
 
@@ -2788,6 +2870,32 @@ async def _generate_room_entry(
 
             if room_builder.rtype == "joined":
                 unread_notifications: dict[str, int] = {}
+                sticky_events: list[EventBase] = []
+                if sticky_event_ids:
+                    # As per MSC4354:
+                    # Remove sticky events that are already in the timeline, else we will needlessly duplicate
+                    # events.
+                    # There is no purpose in including sticky events in the sticky section if they're already in
+                    # the timeline, as either way the client becomes aware of them.
+                    # This is particularly important given the risk of sticky events spam since
+                    # anyone can send sticky events, so halving the bandwidth on average for each sticky
+                    # event is helpful.
+                    timeline_event_id_set = {ev.event_id for ev in batch.events}
+                    # Must preserve sticky event stream order
+                    sticky_event_ids = [
+                        e for e in sticky_event_ids if e not in timeline_event_id_set
+                    ]
+                    if sticky_event_ids:
+                        # Fetch and filter the sticky events
+                        sticky_events = await filter_and_transform_events_for_client(
+                            self._storage_controllers,
+                            sync_result_builder.sync_config.user.to_string(),
+                            await self.store.get_events_as_list(sticky_event_ids),
+                            # As per MSC4354:
+                            # > History visibility checks MUST NOT be applied to sticky events.
+                            # > Any joined user is authorised to see sticky events for the duration they remain sticky.
+                            always_include_ids=frozenset(sticky_event_ids),
+                        )
                 room_sync = JoinedSyncResult(
                     room_id=room_id,
                     timeline=batch,
@@ -2798,6 +2906,7 @@ async def _generate_room_entry(
                     unread_thread_notifications={},
                     summary=summary,
                     unread_count=0,
+                    sticky=sticky_events,
                 )
 
                 if room_sync or always_include:
diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py
index 91f2f16771a..710d097eab0 100644
--- a/synapse/rest/client/sync.py
+++ b/synapse/rest/client/sync.py
@@ -619,6 +619,14 @@ async def encode_room(
             ephemeral_events = room.ephemeral
             result["ephemeral"] = {"events": ephemeral_events}
             result["unread_notifications"] = room.unread_notifications
+            if room.sticky:
+                # The sticky events have already been deduplicated so that events
+                # appearing in the timeline won't appear again here
+                result["msc4354_sticky"] = {
+                    "events": await self._event_serializer.serialize_events(
+                        room.sticky, time_now, config=serialize_options
+                    )
+                }
             if room.unread_thread_notifications:
                 result["unread_thread_notifications"] = room.unread_thread_notifications
                 if self._msc3773_enabled:
diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py
index 101306296e4..38b84443df6 100644
--- a/synapse/storage/databases/main/sticky_events.py
+++ b/synapse/storage/databases/main/sticky_events.py
@@ -13,9 +13,7 @@
 import logging
 import random
 from dataclasses import dataclass
-from typing import (
-    TYPE_CHECKING,
-)
+from typing import TYPE_CHECKING, Collection, cast
 
 from twisted.internet.defer import Deferred
 
@@ -25,6 +23,7 @@
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
+    make_in_list_sql_clause,
 )
 from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
@@ -138,6 +137,98 @@ def get_max_sticky_events_stream_id(self) -> int:
     def get_sticky_events_stream_id_generator(self) -> MultiWriterIdGenerator:
         return self._sticky_events_id_gen
 
+    async def get_sticky_events_in_rooms(
+        self,
+        room_ids: Collection[str],
+        *,
+        from_id: int,
+        to_id: int,
+        now: int,
+        limit: int | None,
+    ) -> tuple[int, dict[str, list[str]]]:
+        """
+        Fetch all the sticky events' IDs in the given rooms, with sticky stream IDs satisfying
+        from_id < sticky stream ID <= to_id.
+
+        The events are returned ordered by the sticky events stream.
+
+        Args:
+            room_ids: The room IDs to return sticky events in.
+            from_id: The sticky stream ID that sticky events should be returned from (exclusive).
+            to_id: The sticky stream ID that sticky events should end at (inclusive).
+            now: The current time in unix millis, used for skipping expired events.
+            limit: Max sticky events to return, or None to apply no limit.
+        Returns:
+            to_id, dict[room_id, list[event_ids]]
+        """
+        sticky_events_rows = await self.db_pool.runInteraction(
+            "get_sticky_events_in_rooms",
+            self._get_sticky_events_in_rooms_txn,
+            room_ids,
+            from_id=from_id,
+            to_id=to_id,
+            now=now,
+            limit=limit,
+        )
+
+        if not sticky_events_rows:
+            return to_id, {}
+
+        # Get stream_id of the last row, which is the highest
+        new_to_id, _, _ = sticky_events_rows[-1]
+
+        # room ID -> event IDs
+        room_id_to_event_ids: dict[str, list[str]] = {}
+        for _, room_id, event_id in sticky_events_rows:
+            events = room_id_to_event_ids.setdefault(room_id, [])
+            events.append(event_id)
+
+        return (new_to_id, room_id_to_event_ids)
+
+    def _get_sticky_events_in_rooms_txn(
+        self,
+        txn: LoggingTransaction,
+        room_ids: Collection[str],
+        *,
+        from_id: int,
+        to_id: int,
+        now: int,
+        limit: int | None,
+    ) -> list[tuple[int, str, str]]:
+        if len(room_ids) == 0:
+            return []
+        room_id_in_list_clause, room_id_in_list_values = make_in_list_sql_clause(
+            txn.database_engine, "se.room_id", room_ids
+        )
+        limit_clause = ""
+        limit_params: tuple[int, ...] = ()
+        if limit is not None:
+            limit_clause = "LIMIT ?"
+            limit_params = (limit,)
+
+        if isinstance(self.database_engine, PostgresEngine):
+            expr_soft_failed = "COALESCE(((ej.internal_metadata::jsonb)->>'soft_failed')::boolean, FALSE)"
+        else:
+            expr_soft_failed = "COALESCE(ej.internal_metadata->>'soft_failed', FALSE)"
+
+        txn.execute(
+            f"""
+            SELECT se.stream_id, se.room_id, event_id
+            FROM sticky_events se
+            INNER JOIN event_json ej USING (event_id)
+            WHERE
+                NOT {expr_soft_failed}
+                AND ? < expires_at
+                AND ? < stream_id
+                AND stream_id <= ?
+                AND {room_id_in_list_clause}
+            ORDER BY stream_id ASC
+            {limit_clause}
+            """,
+            (now, from_id, to_id, *room_id_in_list_values, *limit_params),
+        )
+        return cast(list[tuple[int, str, str]], txn.fetchall())
+
     async def get_updated_sticky_events(
         self, *, from_id: int, to_id: int, limit: int
     ) -> list[StickyEventUpdate]:
diff --git a/tests/rest/client/test_sync_sticky_events.py b/tests/rest/client/test_sync_sticky_events.py
new file mode 100644
index 00000000000..7a38debdb95
--- /dev/null
+++ b/tests/rest/client/test_sync_sticky_events.py
@@ -0,0 +1,529 @@
+#
+# 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>.
+import json
+import sqlite3
+from dataclasses import dataclass
+from unittest.mock import patch
+from urllib.parse import quote
+
+from twisted.internet.testing import MemoryReactor
+
+from synapse.api.constants import EventTypes, EventUnsignedContentFields, StickyEvent
+from synapse.rest import admin
+from synapse.rest.client import account_data, login, register, room, sync
+from synapse.server import HomeServer
+from synapse.types import JsonDict
+from synapse.util.clock import Clock
+from synapse.util.duration import Duration
+
+from tests import unittest
+from tests.utils import USE_POSTGRES_FOR_TESTS
+
+
+class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
+    """
+    Tests for oldschool (v3) /sync with sticky events (MSC4354)
+    """
+
+    if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0):
+        # We need the JSON functionality in SQLite
+        skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)"
+
+    servlets = [
+        room.register_servlets,
+        login.register_servlets,
+        register.register_servlets,
+        admin.register_servlets,
+        sync.register_servlets,
+        account_data.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config["experimental_features"] = {"msc4354_enabled": True}
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        # Register an account
+        self.user_id = self.register_user("user1", "pass")
+        self.token = self.login(self.user_id, "pass")
+
+        # Create a room
+        self.room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+    def test_single_sticky_event_appears_in_initial_sync(self) -> None:
+        """
+        Test sending a single sticky event and then doing an initial /sync.
+        """
+
+        # Send a sticky event
+        sticky_event_response = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "sticky message", "msgtype": "m.text"},
+            tok=self.token,
+        )
+        sticky_event_id = sticky_event_response["event_id"]
+
+        # Perform initial sync
+        channel = self.make_request(
+            "GET",
+            "/sync",
+            access_token=self.token,
+        )
+
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # Get timeline events from the sync response
+        timeline_events = channel.json_body["rooms"]["join"][self.room_id]["timeline"][
+            "events"
+        ]
+
+        # Verify the sticky event is present and has the sticky TTL field
+        self.assertEqual(
+            timeline_events[-1]["event_id"],
+            sticky_event_id,
+            f"Sticky event {sticky_event_id} not found in sync timeline",
+        )
+        self.assertEqual(
+            timeline_events[-1]["unsigned"][EventUnsignedContentFields.STICKY_TTL],
+            # The other 100 ms is advanced in FakeChannel.await_result.
+            59_900,
+        )
+
+        self.assertNotIn(
+            "sticky",
+            channel.json_body["rooms"]["join"][self.room_id],
+            "Unexpected sticky section of sync response (sticky event should be deduplicated)",
+        )
+
+    def test_sticky_event_beyond_timeline_in_initial_sync(self) -> None:
+        """
+        Test that when sending a sticky event which is subsequently
+        pushed out of the timeline window by other messages,
+        the sticky event comes down the dedicated sticky section of the sync response.
+        """
+        # Send the first sticky event
+        first_sticky_response = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "first sticky", "msgtype": "m.text"},
+            tok=self.token,
+        )
+        first_sticky_event_id = first_sticky_response["event_id"]
+
+        # Send 10 regular timeline events,
+        # in order to push the sticky event out of the timeline window
+        # that the /sync will get.
+        regular_event_ids = []
+        for i in range(10):
+            # (Note: each one advances time by 100ms)
+            response = self.helper.send(
+                room_id=self.room_id,
+                body=f"regular message {i}",
+                tok=self.token,
+            )
+            regular_event_ids.append(response["event_id"])
+
+        # Send another sticky event
+        # (Note: this advances time by 100ms)
+        second_sticky_response = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "second sticky", "msgtype": "m.text"},
+            tok=self.token,
+        )
+        second_sticky_event_id = second_sticky_response["event_id"]
+
+        # Perform initial sync
+        channel = self.make_request(
+            "GET",
+            "/sync",
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # Get timeline events from the sync response
+        timeline_events = channel.json_body["rooms"]["join"][self.room_id]["timeline"][
+            "events"
+        ]
+        timeline_event_ids = [event["event_id"] for event in timeline_events]
+        sticky_events = channel.json_body["rooms"]["join"][self.room_id][
+            "msc4354_sticky"
+        ]["events"]
+
+        # This is canary to check the test setup is valid and that we're actually excluding the sticky event
+        # because it's outside the timeline window, not for some other potential reason.
+        self.assertNotIn(
+            regular_event_ids[0],
+            timeline_event_ids,
+            f"First regular event {regular_event_ids[0]} unexpectedly found in sync timeline (this means our test is invalid)",
+        )
+
+        # Assertions for the first sticky event: should be only in sticky section
+        self.assertNotIn(
+            first_sticky_event_id,
+            timeline_event_ids,
+            f"First sticky event {first_sticky_event_id} unexpectedly found in sync timeline (expected it to be outside the timeline window)",
+        )
+        self.assertEqual(
+            len(sticky_events),
+            1,
+            f"Expected exactly 1 item in sticky events section, got {sticky_events}",
+        )
+        self.assertEqual(sticky_events[0]["event_id"], first_sticky_event_id)
+        self.assertEqual(
+            # The 'missing' 1100 ms were elapsed when sending events
+            sticky_events[0]["unsigned"][EventUnsignedContentFields.STICKY_TTL],
+            58_800,
+        )
+
+        # Assertions for the second sticky event: should be only in timeline section
+        self.assertEqual(
+            timeline_events[-1]["event_id"],
+            second_sticky_event_id,
+            f"Second sticky event {second_sticky_event_id} not found in sync timeline",
+        )
+        self.assertEqual(
+            timeline_events[-1]["unsigned"][EventUnsignedContentFields.STICKY_TTL],
+            # The other 100 ms is advanced in FakeChannel.await_result.
+            59_900,
+        )
+        # (sticky section: we already checked it only has 1 item and
+        # that item was the first above)
+
+    def test_sticky_event_filtered_from_timeline_appears_in_sticky_section(
+        self,
+    ) -> None:
+        """
+        Test that a sticky event which is excluded from the timeline by a timeline filter
+        still appears in the sticky section of the sync response.
+
+        > Interaction with RoomFilter: The RoomFilter does not apply to the sticky.events section,
+        > as it is neither timeline nor state. However, the timeline filter MUST be applied before
+        > applying the deduplication logic above. In other words, if a sticky event would normally
+        > appear in both the timeline.events section and the sticky.events section, but is filtered
+        > out by the timeline filter, the sticky event MUST appear in sticky.events.
+        > — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#sync-api-changes
+        """
+        # A filter that excludes message events
+        filter_json = json.dumps(
+            {
+                "room": {
+                    # We only want these io.element.example events
+                    "timeline": {"types": ["io.element.example"]},
+                }
+            }
+        )
+
+        # Send a sticky message event (will be filtered by our filter)
+        sticky_event_id = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "sticky message", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+
+        # Send a non-message event (will pass the filter)
+        nonmessage_event_id = self.helper.send_event(
+            room_id=self.room_id,
+            type="io.element.example",
+            content={"membership": "join"},
+            tok=self.token,
+        )["event_id"]
+
+        # Perform initial sync with our filter
+        channel = self.make_request(
+            "GET",
+            f"/sync?filter={quote(filter_json)}",
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # Get timeline and sticky events from the sync response
+        timeline_events = channel.json_body["rooms"]["join"][self.room_id]["timeline"][
+            "events"
+        ]
+        sticky_events = channel.json_body["rooms"]["join"][self.room_id][
+            "msc4354_sticky"
+        ]["events"]
+
+        # Extract event IDs from the timeline
+        timeline_event_ids = [event["event_id"] for event in timeline_events]
+
+        # The sticky message event should NOT be in the timeline (filtered out)
+        self.assertNotIn(
+            sticky_event_id,
+            timeline_event_ids,
+            f"Sticky message event {sticky_event_id} should be filtered from timeline",
+        )
+
+        # The member event should be in the timeline
+        self.assertIn(
+            nonmessage_event_id,
+            timeline_event_ids,
+            f"Non-message event {nonmessage_event_id} should be in timeline",
+        )
+
+        # The sticky message event MUST appear in the sticky section
+        # because it was filtered from the timeline
+        received_sticky_event_ids = [e["event_id"] for e in sticky_events]
+        self.assertEqual(
+            received_sticky_event_ids,
+            [sticky_event_id],
+        )
+
+    def test_ignored_users_sticky_events(self) -> None:
+        """
+        Test that sticky events from ignored users are not delivered to clients.
+
+        > As with normal events, sticky events sent by ignored users MUST NOT be
+        > delivered to clients.
+        > — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#sync-api-changes
+        """
+        # Register a second user who will be ignored
+        user2_id = self.register_user("user2", "pass")
+        user2_token = self.login(user2_id, "pass")
+
+        # Join user2 to the room
+        self.helper.join(self.room_id, user2_id, tok=user2_token)
+
+        # User1 ignores user2
+        channel = self.make_request(
+            "PUT",
+            f"/_matrix/client/v3/user/{self.user_id}/account_data/m.ignored_user_list",
+            {"ignored_users": {user2_id: {}}},
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # User2 sends a sticky event
+        sticky_response = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "sticky from ignored user", "msgtype": "m.text"},
+            tok=user2_token,
+        )
+        sticky_event_id = sticky_response["event_id"]
+
+        # User1 syncs
+        channel = self.make_request(
+            "GET",
+            "/sync",
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # Check timeline events - should not include the sticky event from ignored user
+        timeline_events = channel.json_body["rooms"]["join"][self.room_id]["timeline"][
+            "events"
+        ]
+        timeline_event_ids = [event["event_id"] for event in timeline_events]
+
+        self.assertNotIn(
+            sticky_event_id,
+            timeline_event_ids,
+            f"Sticky event from ignored user {sticky_event_id} should not be in timeline",
+        )
+
+        # Check sticky events section - should also not include the event
+        sticky_events = (
+            channel.json_body["rooms"]["join"][self.room_id]
+            .get("msc4354_sticky", {})
+            .get("events", [])
+        )
+
+        sticky_event_ids = [event["event_id"] for event in sticky_events]
+
+        self.assertNotIn(
+            sticky_event_id,
+            sticky_event_ids,
+            f"Sticky event from ignored user {sticky_event_id} should not be in sticky section",
+        )
+
+    def test_history_visibility_bypass_for_sticky_events(self) -> None:
+        """
+        Test that joined users can see sticky events even when history visibility
+        is set to "joined" and they joined after the event was sent.
+        This is required by MSC4354: "History visibility checks MUST NOT
+        be applied to sticky events. Any joined user is authorised to see
+        sticky events for the duration they remain sticky."
+        """
+        # Create a new room with restricted history visibility
+        room_id = self.helper.create_room_as(
+            self.user_id,
+            tok=self.token,
+            extra_content={
+                # Anyone can join
+                "preset": "public_chat",
+                # But you can't see history before you joined
+                "initial_state": [
+                    {
+                        "type": EventTypes.RoomHistoryVisibility,
+                        "state_key": "",
+                        "content": {"history_visibility": "joined"},
+                    }
+                ],
+            },
+            is_public=False,
+        )
+
+        # User1 sends a sticky event
+        sticky_event_id = self.helper.send_sticky_event(
+            room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=5),
+            content={"body": "sticky message", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+
+        # User1 also sends a regular event, to verify our test setup
+        regular_event_id = self.helper.send(
+            room_id=room_id,
+            body="regular message",
+            tok=self.token,
+        )["event_id"]
+
+        # Register and join a second user after the sticky event was sent
+        user2_id = self.register_user("user2", "pass")
+        user2_token = self.login(user2_id, "pass")
+        self.helper.join(room_id, user2_id, tok=user2_token)
+
+        # User2 syncs - they should see the sticky event even though
+        # history visibility is "joined" and they joined after it was sent
+        channel = self.make_request(
+            "GET",
+            "/sync",
+            access_token=user2_token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # Get sticky events from the sync response
+        sticky_events = channel.json_body["rooms"]["join"][room_id]["msc4354_sticky"][
+            "events"
+        ]
+        sticky_event_ids = [event["event_id"] for event in sticky_events]
+
+        # The sticky event should be visible to user2
+        self.assertEqual(
+            [sticky_event_id],
+            sticky_event_ids,
+            f"Sticky event {sticky_event_id} should be visible despite history visibility",
+        )
+
+        # Also check that the regular (non-sticky) event sent at the same time
+        # is NOT visible. This is to verify our test setup.
+        timeline_events = channel.json_body["rooms"]["join"][room_id]["timeline"][
+            "events"
+        ]
+        timeline_event_ids = [event["event_id"] for event in timeline_events]
+
+        # The regular message should NOT be visible (history visibility = joined)
+        self.assertNotIn(
+            regular_event_id,
+            timeline_event_ids,
+            f"Expecting to not see regular event ({regular_event_id}) before user1 joined.",
+        )
+
+    @patch.object(StickyEvent, "MAX_EVENTS_IN_SYNC", 3)
+    def test_pagination_with_many_sticky_events(self) -> None:
+        """
+        Test that pagination works correctly when there are more sticky events than
+        the intended limit.
+
+        The MSC doesn't define a limit or how to set one.
+        See thread: https://github.com/matrix-org/matrix-spec-proposals/pull/4354#discussion_r2885670008
+
+        But Synapse currently emits 100 at a time, controlled by `MAX_EVENTS_IN_SYNC`.
+        In this test we patch it to 3 (as sending 100 events is not very efficient).
+        """
+
+        # A filter that excludes message events
+        # This is needed so that the sticky events come down the sticky section
+        # and not the timeline section, which would hamper our test.
+        filter_json = json.dumps(
+            {
+                "room": {
+                    # We only want these io.element.example events
+                    "timeline": {"types": ["io.element.example"]},
+                }
+            }
+        )
+
+        # Send 8 sticky events: enough for 2 full pages and then a partial page with 2.
+        sent_sticky_event_ids = []
+        for i in range(8):
+            response = self.helper.send_sticky_event(
+                self.room_id,
+                EventTypes.Message,
+                duration=Duration(minutes=1),
+                content={"body": f"sticky message {i}", "msgtype": "m.text"},
+                tok=self.token,
+            )
+            sent_sticky_event_ids.append(response["event_id"])
+
+        @dataclass
+        class SyncHelperResponse:
+            sticky_event_ids: list[str]
+            """Event IDs of events returned from the sticky section, in-order."""
+
+            next_batch: str
+            """Sync token to pass to `?since=` in order to do an incremental sync."""
+
+        def _do_sync(*, since: str | None) -> SyncHelperResponse:
+            """Small helper to do a sync and get the sticky events out."""
+            and_since_param = "" if since is None else f"&since={since}"
+            channel = self.make_request(
+                "GET",
+                f"/sync?filter={quote(filter_json)}{and_since_param}",
+                access_token=self.token,
+            )
+            self.assertEqual(channel.code, 200, channel.result)
+
+            # Get sticky events from the sync response
+            sticky_events = []
+            # In this test, when no sticky events are in the response,
+            # we don't have a rooms section at all
+            if "rooms" in channel.json_body:
+                sticky_events = channel.json_body["rooms"]["join"][self.room_id][
+                    "msc4354_sticky"
+                ]["events"]
+
+            return SyncHelperResponse(
+                sticky_event_ids=[
+                    sticky_event["event_id"] for sticky_event in sticky_events
+                ],
+                next_batch=channel.json_body["next_batch"],
+            )
+
+        # Perform initial sync, we should get the first 3 sticky events,
+        # in order.
+        sync1 = _do_sync(since=None)
+        self.assertEqual(sync1.sticky_event_ids, sent_sticky_event_ids[0:3])
+
+        # Now do an incremental sync and expect the next page of 3
+        sync2 = _do_sync(since=sync1.next_batch)
+        self.assertEqual(sync2.sticky_event_ids, sent_sticky_event_ids[3:6])
+
+        # Now do another incremental sync and expect the last 2
+        sync3 = _do_sync(since=sync2.next_batch)
+        self.assertEqual(sync3.sticky_event_ids, sent_sticky_event_ids[6:8])
+
+        # Finally, expect an empty incremental sync at the end
+        sync4 = _do_sync(since=sync3.next_batch)
+        self.assertEqual(sync4.sticky_event_ids, [])