diff --git a/lms/djangoapps/ccx/modulestore.py b/lms/djangoapps/ccx/modulestore.py
index d47a8396084f..b2cb414960bf 100644
--- a/lms/djangoapps/ccx/modulestore.py
+++ b/lms/djangoapps/ccx/modulestore.py
@@ -304,3 +304,10 @@ def bulk_operations(self, course_id, emit_signals=True, ignore_case=False):
         course_id, _ = strip_ccx(course_id)
         with self._modulestore.bulk_operations(course_id, emit_signals=emit_signals, ignore_case=ignore_case):
             yield
+
+    @contextmanager
+    def fresh_block_reads(self, course_key):
+        """See the docs for xmodule.modulestore.mixed.MixedModuleStore"""
+        course_key, _ = strip_ccx(course_key)
+        with self._modulestore.fresh_block_reads(course_key):
+            yield
diff --git a/lms/djangoapps/ccx/tests/test_ccx_modulestore.py b/lms/djangoapps/ccx/tests/test_ccx_modulestore.py
index b3b36a96dc9a..ed94bac90296 100644
--- a/lms/djangoapps/ccx/tests/test_ccx_modulestore.py
+++ b/lms/djangoapps/ccx/tests/test_ccx_modulestore.py
@@ -126,6 +126,16 @@ def test_get_item(self):
             assert expected.display_name == actual.display_name
             assert expected.location == actual.location.to_block_locator()
 
+    def test_fresh_block_reads(self):
+        """verify that fresh block reads by ccx key ignore blocks changed in memory"""
+        expected = self.blocks[0]
+        block_key = self.ccx_locator.make_usage_key(expected.location.block_type, expected.location.block_id)
+        with self.store.bulk_operations(self.ccx_locator):
+            self.store.get_item(block_key).display_name = 'Changed in memory'
+            with self.store.fresh_block_reads(self.ccx_locator):
+                assert self.store.get_item(block_key).display_name == expected.display_name
+            assert self.store.get_item(block_key).display_name == 'Changed in memory'
+
     def test_publication_api(self):
         """verify that we can correctly discern a published item by ccx key"""
         with self.store.bulk_operations(self.ccx_locator):
diff --git a/lms/djangoapps/course_blocks/api.py b/lms/djangoapps/course_blocks/api.py
index de4e0443fb08..65b61b909a9f 100644
--- a/lms/djangoapps/course_blocks/api.py
+++ b/lms/djangoapps/course_blocks/api.py
@@ -6,6 +6,7 @@
 
 from django.conf import settings
 from edx_when import field_data
+from openedx_filters.learning.filters import CourseBlocksTransformCompleted
 
 from lms.djangoapps.course_api.blocks.transformers.block_completion import BlockCompletionTransformer
 from openedx.core.djangoapps.content.block_structure.api import get_block_structure_manager
@@ -104,9 +105,12 @@ def get_course_blocks(
         include_has_scheduled_content
     )
 
-    return get_block_structure_manager(starting_block_usage_key.course_key).get_transformed(
+    block_structure = get_block_structure_manager(starting_block_usage_key.course_key).get_transformed(
         transformers,
         starting_block_usage_key,
         collected_block_structure,
         user,
     )
+    return CourseBlocksTransformCompleted.run_filter(
+        block_structure=block_structure, user=user, starting_block_usage_key=starting_block_usage_key
+    )
diff --git a/lms/djangoapps/course_blocks/tests/test_api.py b/lms/djangoapps/course_blocks/tests/test_api.py
index 3643c520b5a4..53bbacf169f2 100644
--- a/lms/djangoapps/course_blocks/tests/test_api.py
+++ b/lms/djangoapps/course_blocks/tests/test_api.py
@@ -7,16 +7,33 @@
 
 import ddt
 from django.http.request import HttpRequest
+from django.test import override_settings
+from openedx_filters import PipelineStep
 
 from common.djangoapps.student.tests.factories import UserFactory
 from lms.djangoapps.course_blocks.api import get_course_blocks
+from lms.djangoapps.course_blocks.transformers.load_override_data import OverrideDataTransformer
 from lms.djangoapps.course_blocks.transformers.tests.helpers import CourseStructureTestCase
 from lms.djangoapps.course_blocks.transformers.tests.test_user_partitions import UserPartitionTestMixin
 from lms.djangoapps.courseware.block_render import make_track_function, prepare_runtime_for_user
+from openedx.core.djangoapps.content.block_structure.api import get_block_structure_manager
 from openedx.core.djangoapps.content.block_structure.transformers import BlockStructureTransformers
 from openedx.core.djangoapps.course_groups.cohorts import add_user_to_cohort
 from xmodule.modulestore.django import modulestore
 
+FILTERED_USERNAME = 'filtered_user'
+
+
+class TestOverrideDisplayNameStep(PipelineStep):
+    """
+    Pipeline step that overrides the course display name for a single user.
+    """
+
+    def run_filter(self, block_structure, user, starting_block_usage_key):  # pylint: disable=arguments-differ
+        if user.username == FILTERED_USERNAME:
+            block_structure.override_xblock_field(starting_block_usage_key, 'display_name', 'Changed')
+        return {'block_structure': block_structure}
+
 
 def get_block_side_effect(block_locator, user_known):
     """
@@ -147,3 +164,36 @@ def test_get_course_blocks(self, group_id, expected_blocks, user_known):
             set(block_structure.get_block_keys()),
             self.get_block_key_set(self.blocks, *expected_blocks)
         )
+
+    @override_settings(
+        OPEN_EDX_FILTERS_CONFIG={
+            'org.openedx.learning.course_blocks.transform.completed.v1': {
+                'pipeline': ['lms.djangoapps.course_blocks.tests.test_api.TestOverrideDisplayNameStep'],
+                'fail_silently': False,
+            },
+        },
+    )
+    # Register a transformer that collects display_name so the collected structure carries the authored value.
+    @patch(
+        'openedx.core.djangoapps.content.block_structure.transformer_registry.'
+        'TransformerRegistry.get_registered_transformers',
+        Mock(return_value={OverrideDataTransformer}),
+    )
+    def test_filter_overrides_do_not_leak_into_collected_structure(self):
+        """
+        Tests that changes made by the transform completed filter are returned on cold and warm caches
+        without modifying the collected block structure or the blocks returned to other users.
+        """
+        self.setup_partitions_and_course()
+        root_key = self.course.location
+        authored_name = self.course.display_name
+        filtered_user = UserFactory.create(username=FILTERED_USERNAME)
+
+        for _ in range(2):  # the first call collects the structure, the second reads it from the cache
+            block_structure = get_course_blocks(filtered_user, root_key, BlockStructureTransformers([]))
+            assert block_structure.get_xblock_field(root_key, 'display_name') == 'Changed'
+
+        collected = get_block_structure_manager(self.course.id).get_collected()
+        assert collected.get_xblock_field(root_key, 'display_name') == authored_name
+        block_structure = get_course_blocks(self.user, root_key, BlockStructureTransformers([]))
+        assert block_structure.get_xblock_field(root_key, 'display_name') == authored_name
diff --git a/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py b/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py
index 2d0ac7b41b7f..a723aaaed833 100644
--- a/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py
+++ b/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py
@@ -9,6 +9,7 @@
 from django.test.utils import override_settings
 from django.urls import reverse
 from edx_toggles.toggles.testutils import override_waffle_flag
+from openedx_filters import PipelineStep
 from openedx_filters.learning.filters import CoursewareAccessChecksRequested
 
 from common.djangoapps.course_modes.models import CourseMode
@@ -28,6 +29,19 @@
 )
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
 
+FILTERED_TABS = [{'tab_id': 'filtered', 'title': 'Filtered tab', 'url': 'https://example.com/filtered'}]
+filter_users = []
+
+
+class TestReplaceMetadataStep(PipelineStep):
+    """
+    Pipeline step that records the user it receives and replaces the title and tabs.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        filter_users.append(user)
+        return {'data': {**data, 'title': f'Filtered {course_key.course}', 'tabs': FILTERED_TABS}}
+
 
 @ddt.ddt
 @override_waffle_flag(COURSEWARE_MICROFRONTEND_PROGRESS_MILESTONES, active=True)
@@ -87,6 +101,29 @@ def test_get_masqueraded_user(self):
         self.update_masquerade(username=self.user.username)
         assert self.client.get(self.url).data['username'] == self.user.username
 
+    @override_settings(OPEN_EDX_FILTERS_CONFIG={
+        'org.openedx.learning.course_home_metadata.api.render.started.v1': {
+            'pipeline': [f'{__name__}.TestReplaceMetadataStep'],
+            'fail_silently': False,
+        },
+    })
+    def test_metadata_filter_replaces_data_for_masqueraded_user(self):
+        """
+        Test that the response uses the data returned by the pipeline, which receives the masqueraded user.
+        """
+        CourseEnrollment.enroll(self.user, self.course.id, CourseMode.VERIFIED)
+        self.client.logout()
+        self.client.login(username=self.staff_user.username, password='bar')
+        self.update_masquerade(username=self.user.username)
+        filter_users.clear()
+
+        response = self.client.get(self.url)
+
+        assert response.status_code == 200
+        assert response.data['title'] == f'Filtered {self.course.id.course}'
+        assert response.data['tabs'] == FILTERED_TABS
+        assert filter_users == [self.user]
+
     def test_get_unknown_course(self):
         self.client.logout()
         url = reverse('course-home:course-metadata', args=['course-v1:unknown+course+2T2020'])
diff --git a/lms/djangoapps/course_home_api/course_metadata/views.py b/lms/djangoapps/course_home_api/course_metadata/views.py
index f6f15203260e..2cdc67f69501 100644
--- a/lms/djangoapps/course_home_api/course_metadata/views.py
+++ b/lms/djangoapps/course_home_api/course_metadata/views.py
@@ -7,6 +7,7 @@
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys.edx.keys import CourseKey
+from openedx_filters.learning.filters import CourseHomeMetadataAPIRenderStarted
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.generics import RetrieveAPIView
 from rest_framework.response import Response
@@ -161,4 +162,7 @@ def get(self, request, *args, **kwargs):
         context['course_overview'] = course
         context['enrollment'] = enrollment
         serializer = self.get_serializer_class()(data, context=context)
-        return Response(serializer.data)
+        data = CourseHomeMetadataAPIRenderStarted.run_filter(
+            course_key=course_key, user=request.user, data=serializer.data
+        )
+        return Response(data)
diff --git a/lms/djangoapps/course_home_api/dates/tests/test_views.py b/lms/djangoapps/course_home_api/dates/tests/test_views.py
index 0e37fbcaafae..ac4864d6776e 100644
--- a/lms/djangoapps/course_home_api/dates/tests/test_views.py
+++ b/lms/djangoapps/course_home_api/dates/tests/test_views.py
@@ -5,13 +5,28 @@
 from datetime import datetime
 
 import ddt
+from django.test import override_settings
 from django.urls import reverse
+from openedx_filters import PipelineStep
 
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.models import CourseEnrollment
+from common.djangoapps.student.tests.factories import UserFactory
 from lms.djangoapps.course_home_api.tests.utils import BaseCourseHomeTests
 from openedx.features.content_type_gating.models import ContentTypeGatingConfig
 
+filter_users = []
+
+
+class TestReplaceDatesDataStep(PipelineStep):
+    """
+    Pipeline step that records the user and replaces the response data.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        filter_users.append(user)
+        return {'data': {'course_key': str(course_key)}}
+
 
 @ddt.ddt
 class DatesTabTestViews(BaseCourseHomeTests):
@@ -69,3 +84,25 @@ def test_masquerade(self):
 
         self.update_masquerade(role='student')
         assert not self.client.get(self.url).data.get('learner_is_full_access')
+
+    @override_settings(OPEN_EDX_FILTERS_CONFIG={
+        'org.openedx.learning.course_home_dates.api.render.started.v1': {
+            'pipeline': [f'{__name__}.TestReplaceDatesDataStep'],
+            'fail_silently': False,
+        },
+    })
+    def test_render_filter_receives_masqueraded_user(self):
+        """
+        Test that the render filter can replace the response and receives the masqueraded user.
+        """
+        learner = UserFactory()
+        CourseEnrollment.enroll(learner, self.course.id)
+        self.switch_to_staff()
+        self.update_masquerade(username=learner.username)
+        filter_users.clear()
+
+        response = self.client.get(self.url)
+
+        assert response.status_code == 200
+        assert response.data == {'course_key': str(self.course.id)}
+        assert filter_users == [learner]
diff --git a/lms/djangoapps/course_home_api/dates/views.py b/lms/djangoapps/course_home_api/dates/views.py
index 64f8fc34428e..05deaef68cca 100644
--- a/lms/djangoapps/course_home_api/dates/views.py
+++ b/lms/djangoapps/course_home_api/dates/views.py
@@ -6,6 +6,7 @@
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys.edx.keys import CourseKey
+from openedx_filters.learning.filters import CourseHomeDatesAPIRenderStarted
 from rest_framework.generics import RetrieveAPIView
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
@@ -130,5 +131,6 @@ def get(self, request, *args, **kwargs):
         context = self.get_serializer_context()
         context['learner_is_full_access'] = learner_is_full_access
         serializer = self.get_serializer_class()(data, context=context)
+        data = CourseHomeDatesAPIRenderStarted.run_filter(course_key=course_key, user=request.user, data=serializer.data)
 
-        return Response(serializer.data)
+        return Response(data)
diff --git a/lms/djangoapps/course_home_api/outline/tests/test_view.py b/lms/djangoapps/course_home_api/outline/tests/test_view.py
index b802a806eeca..79b9cbbaeadc 100644
--- a/lms/djangoapps/course_home_api/outline/tests/test_view.py
+++ b/lms/djangoapps/course_home_api/outline/tests/test_view.py
@@ -13,7 +13,9 @@
 from django.contrib.auth.models import AnonymousUser
 from django.test import override_settings
 from django.urls import reverse
+from django.utils.translation import get_language
 from edx_toggles.toggles.testutils import override_waffle_flag
+from openedx_filters import PipelineStep
 
 from cms.djangoapps.contentstore.outlines import update_outline_from_modulestore
 from common.djangoapps.course_modes.models import CourseMode
@@ -38,6 +40,46 @@
 from xmodule.course_block import COURSE_VISIBILITY_PUBLIC, COURSE_VISIBILITY_PUBLIC_OUTLINE
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
 
+block_filter_languages = []
+
+
+class TestLanguageBlockNamesStep(PipelineStep):
+    """
+    Pipeline step that appends the active language to every block display name.
+    """
+
+    def run_filter(self, block_structure, user, starting_block_usage_key):  # pylint: disable=arguments-differ
+        language = get_language()
+        block_filter_languages.append(language)
+        for block_key in block_structure:
+            display_name = block_structure.get_xblock_field(block_key, 'display_name')
+            block_structure.override_xblock_field(block_key, 'display_name', f'{display_name} [{language}]')
+        return {'block_structure': block_structure}
+
+
+filter_users = []
+
+
+class TestReplaceOutlineDataStep(PipelineStep):
+    """
+    Pipeline step that records the user and replaces the response data.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        filter_users.append(user)
+        return {'data': {'course_key': str(course_key)}}
+
+
+class TestRenameNavigationBlocksStep(PipelineStep):
+    """
+    Pipeline step that renames every block in place.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        for block in data['blocks'].values():
+            block['display_name'] = 'Changed'
+        return {'data': data}
+
 
 @ddt.ddt
 class OutlineTabTestViews(BaseCourseHomeTests):
@@ -478,6 +520,28 @@ def test_course_progress_analytics_disabled(self, mock_task):
         self.client.get(self.url)
         mock_task.assert_not_called()
 
+    @override_settings(OPEN_EDX_FILTERS_CONFIG={
+        'org.openedx.learning.course_home_outline.api.render.started.v1': {
+            'pipeline': [f'{__name__}.TestReplaceOutlineDataStep'],
+            'fail_silently': False,
+        },
+    })
+    def test_render_filter_receives_masqueraded_user(self):
+        """
+        Test that the render filter can replace the response and receives the masqueraded user.
+        """
+        learner = UserFactory()
+        CourseEnrollment.enroll(learner, self.course.id)
+        self.switch_to_staff()
+        self.update_masquerade(username=learner.username)
+        filter_users.clear()
+
+        response = self.client.get(self.url)
+
+        assert response.status_code == 200
+        assert response.data == {'course_key': str(self.course.id)}
+        assert filter_users == [learner]
+
 
 @ddt.ddt
 class SidebarBlocksTestViews(BaseCourseHomeTests):
@@ -958,3 +1022,71 @@ def test_navigation_does_not_cache_stale_data_after_publish(self):
         assert response.status_code == 200
         sequential_data = response.data['blocks'][str(self.sequential.location)]
         assert str(self.vertical.location) not in sequential_data['children']
+
+    @override_settings(OPEN_EDX_FILTERS_CONFIG={
+        'org.openedx.learning.course_blocks.transform.completed.v1': {
+            'pipeline': [f'{__name__}.TestLanguageBlockNamesStep'],
+            'fail_silently': False,
+        },
+    })
+    def test_navigation_cache_is_separate_per_language(self):
+        """
+        Test that block data that depends on the active language is cached separately for each language.
+        """
+        self.add_blocks_to_course()
+        CourseEnrollment.enroll(self.user, self.course.id, CourseMode.VERIFIED)
+        # Build the block structure up front so its version, part of the cache key, stays the same.
+        update_course_in_cache(self.course.id)
+        block_filter_languages.clear()
+
+        titles = {}
+        for language in ('en', 'es-419', 'en', 'es-419'):
+            response = self.client.get(self.url, HTTP_ACCEPT_LANGUAGE=language)
+            assert response.status_code == 200
+            title = response.data['blocks'][str(self.sequential.location)]['display_name']
+            assert titles.setdefault(language, title) == title
+
+        assert titles == {'en': 'Test [en]', 'es-419': 'Test [es-419]'}
+        # Repeated requests in the same language are served from the cache.
+        assert block_filter_languages == ['en', 'es-419']
+
+    def test_render_filter_replaces_response(self):
+        """
+        Test that the render filter can replace the response and receives the requesting user.
+        """
+        CourseEnrollment.enroll(self.user, self.course.id)
+        filter_users.clear()
+
+        with override_settings(OPEN_EDX_FILTERS_CONFIG={
+            'org.openedx.learning.course_navigation.api.render.started.v1': {
+                'pipeline': [f'{__name__}.TestReplaceOutlineDataStep'],
+                'fail_silently': False,
+            },
+        }):
+            response = self.client.get(self.url)
+
+        assert response.status_code == 200
+        assert response.data == {'course_key': str(self.course.id)}
+        assert filter_users == [self.user]
+
+    def test_render_filter_changes_are_not_cached(self):
+        """
+        Test that changes made by the render filter do not reach the cached blocks.
+        """
+        self.add_blocks_to_course()
+        CourseEnrollment.enroll(self.user, self.course.id, CourseMode.VERIFIED)
+        # Build the block structure up front so its version, part of the cache key, stays the same.
+        update_course_in_cache(self.course.id)
+        sequential_key = str(self.sequential.location)
+
+        with override_settings(OPEN_EDX_FILTERS_CONFIG={
+            'org.openedx.learning.course_navigation.api.render.started.v1': {
+                'pipeline': [f'{__name__}.TestRenameNavigationBlocksStep'],
+                'fail_silently': False,
+            },
+        }):
+            response = self.client.get(self.url)
+        assert response.data['blocks'][sequential_key]['display_name'] == 'Changed'
+
+        response = self.client.get(self.url)
+        assert response.data['blocks'][sequential_key]['display_name'] == 'Test'
diff --git a/lms/djangoapps/course_home_api/outline/views.py b/lms/djangoapps/course_home_api/outline/views.py
index c50b2eed5e4e..2fc0d71cc72d 100644
--- a/lms/djangoapps/course_home_api/outline/views.py
+++ b/lms/djangoapps/course_home_api/outline/views.py
@@ -11,6 +11,7 @@
 from django.core.cache import cache
 from django.shortcuts import get_object_or_404  # pylint: disable=wrong-import-order
 from django.urls import reverse  # pylint: disable=wrong-import-order
+from django.utils.translation import get_language  # pylint: disable=wrong-import-order
 from django.utils.translation import gettext as _  # pylint: disable=wrong-import-order
 from edx_django_utils import monitoring as monitoring_utils  # pylint: disable=wrong-import-order
 from edx_rest_framework_extensions.auth.jwt.authentication import (
@@ -20,6 +21,10 @@
     SessionAuthenticationAllowInactiveUser,  # pylint: disable=wrong-import-order
 )
 from opaque_keys.edx.keys import CourseKey  # pylint: disable=wrong-import-order
+from openedx_filters.learning.filters import (  # pylint: disable=wrong-import-order
+    CourseHomeOutlineAPIRenderStarted,
+    CourseNavigationAPIRenderStarted,
+)
 from rest_framework.decorators import (  # pylint: disable=wrong-import-order
     api_view,
     authentication_classes,
@@ -383,7 +388,8 @@ def get(self, request, *args, **kwargs):  # pylint: disable=too-many-statements
         if send_course_progress_analytics_for_student_is_enabled(course_key) and not user_is_masquerading:
             collect_progress_for_user_in_course.delay(course_key_string, request.user.id)
 
-        return Response(serializer.data)
+        data = CourseHomeOutlineAPIRenderStarted.run_filter(course_key=course_key, user=request.user, data=serializer.data)
+        return Response(data)
 
     def finalize_response(self, request, response, *args, **kwargs):
         """
@@ -436,7 +442,7 @@ class CourseNavigationBlocksView(RetrieveAPIView):
     serializer_class = CourseBlockSerializer
     COURSE_BLOCKS_CACHE_KEY_TEMPLATE = (
         'course_sidebar_blocks_{course_key_string}_{block_structure_version}_{user_id}_{user_cohort_id}'
-        '_{enrollment_mode}_{allow_public}_{allow_public_outline}_{is_masquerading}'
+        '_{enrollment_mode}_{allow_public}_{allow_public_outline}_{is_masquerading}_{language}'
     )
     COURSE_BLOCKS_CACHE_TIMEOUT = 60 * 60  # 1 hour
 
@@ -477,6 +483,9 @@ def get(self, request, *args, **kwargs):
             allow_public=allow_public,
             allow_public_outline=allow_public_outline,
             is_masquerading=user_is_masquerading,
+            # CourseBlocksTransformCompleted steps can vary block data by the request's
+            # active language, so the cached response must not be shared across languages.
+            language=get_language(),
         )
         if navigation_sidebar_caching_is_disabled := courseware_disable_navigation_sidebar_blocks_caching():
             course_blocks = None
@@ -503,8 +512,12 @@ def get(self, request, *args, **kwargs):
         })
 
         serializer = self.get_serializer_class()(course_blocks, context=context)
+        # Runs on the serialized response, after the cache, so the cached blocks stay as built.
+        data = CourseNavigationAPIRenderStarted.run_filter(
+            course_key=course_key, user=request.user, data=serializer.data
+        )
 
-        return Response(serializer.data)
+        return Response(data)
 
     def filter_inaccessible_blocks(self, course_blocks, course_key):
         """
diff --git a/lms/djangoapps/course_home_api/progress/serializers.py b/lms/djangoapps/course_home_api/progress/serializers.py
index 04ddccd22399..25fdc09f81e1 100644
--- a/lms/djangoapps/course_home_api/progress/serializers.py
+++ b/lms/djangoapps/course_home_api/progress/serializers.py
@@ -128,7 +128,8 @@ def get_assignment_policies(self, grading_policy):
         } for assignment_policy in grading_policy['GRADER']]
 
     def get_assignment_colors(self, obj):
-        return self.ASSIGNMENT_COLORS
+        # A copy, so a change to one response does not reach the shared class list.
+        return list(self.ASSIGNMENT_COLORS)
 
 
 class CertificateDataSerializer(ReadOnlySerializer):
diff --git a/lms/djangoapps/course_home_api/progress/tests/test_views.py b/lms/djangoapps/course_home_api/progress/tests/test_views.py
index 49ec170d97ef..056926c7adce 100644
--- a/lms/djangoapps/course_home_api/progress/tests/test_views.py
+++ b/lms/djangoapps/course_home_api/progress/tests/test_views.py
@@ -11,6 +11,7 @@
 from django.urls import reverse
 from django.utils.timezone import now
 from edx_toggles.toggles.testutils import override_waffle_flag
+from openedx_filters import PipelineStep
 from pytz import UTC
 
 from common.djangoapps.course_modes.models import CourseMode
@@ -36,6 +37,28 @@
 from openedx.features.course_duration_limits.models import CourseDurationLimitConfig
 from xmodule.modulestore.tests.factories import BlockFactory
 
+filter_users = []
+
+
+class TestReplaceProgressDataStep(PipelineStep):
+    """
+    Pipeline step that records the user and replaces the response data.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        filter_users.append(user)
+        return {'data': {'username': data['username'], 'course_key': str(course_key)}}
+
+
+class TestRecolorProgressStep(PipelineStep):
+    """
+    Pipeline step that changes the assignment colors in place.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        data['grading_policy']['assignment_colors'][0] = '#000000'
+        return {'data': data}
+
 
 @ddt.ddt
 @override_waffle_flag(COURSE_HOME_MICROFRONTEND_PROGRESS_TAB, active=True)
@@ -362,3 +385,45 @@ def test_course_grade_considers_subsection_grade_visibility(self, is_staff, expe
         assert response.status_code == 200
         assert response.data['course_grade']['percent'] == expected_percent
         assert response.data['course_grade']['is_passing'] == (expected_percent >= 0.5)
+
+    @override_settings(OPEN_EDX_FILTERS_CONFIG={
+        'org.openedx.learning.course_home_progress.api.render.started.v1': {
+            'pipeline': [f'{__name__}.TestReplaceProgressDataStep'],
+            'fail_silently': False,
+        },
+    })
+    def test_render_filter_receives_learner(self):
+        """
+        Test that the render filter can replace the response and receives the learner whose progress is shown.
+        """
+        learner = self.user
+        CourseEnrollment.enroll(learner, self.course.id)
+        filter_users.clear()
+
+        response = self.client.get(self.url)
+        assert response.status_code == 200
+        assert response.data == {'username': learner.username, 'course_key': str(self.course.id)}
+
+        self.switch_to_staff()
+        url = reverse('course-home:progress-tab-other-student', args=[self.course.id, learner.id])
+        response = self.client.get(url)
+        assert response.status_code == 200
+        assert response.data == {'username': learner.username, 'course_key': str(self.course.id)}
+        assert filter_users == [learner, learner]
+
+    def test_render_filter_changes_do_not_reach_later_responses(self):
+        """
+        Test that changing the response in place does not change the responses that follow.
+        """
+        CourseEnrollment.enroll(self.user, self.course.id)
+        with override_settings(OPEN_EDX_FILTERS_CONFIG={
+            'org.openedx.learning.course_home_progress.api.render.started.v1': {
+                'pipeline': [f'{__name__}.TestRecolorProgressStep'],
+                'fail_silently': False,
+            },
+        }):
+            filtered = self.client.get(self.url)
+        assert filtered.data['grading_policy']['assignment_colors'][0] == '#000000'
+
+        response = self.client.get(self.url)
+        assert response.data['grading_policy']['assignment_colors'][0] == '#D24242'
diff --git a/lms/djangoapps/course_home_api/progress/views.py b/lms/djangoapps/course_home_api/progress/views.py
index 577392dc9527..07a090e456de 100644
--- a/lms/djangoapps/course_home_api/progress/views.py
+++ b/lms/djangoapps/course_home_api/progress/views.py
@@ -9,6 +9,7 @@
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys.edx.keys import CourseKey
+from openedx_filters.learning.filters import CourseHomeProgressAPIRenderStarted
 from rest_framework.generics import RetrieveAPIView
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
@@ -307,5 +308,6 @@ def get(self, request, *args, **kwargs):    # pylint: disable=too-many-statement
         context['course_overview'] = course_overview
         context['enrollment'] = enrollment
         serializer = self.get_serializer_class()(data, context=context)
+        data = CourseHomeProgressAPIRenderStarted.run_filter(course_key=course_key, user=student, data=serializer.data)
 
-        return Response(serializer.data)
+        return Response(data)
diff --git a/lms/djangoapps/courseware/block_render.py b/lms/djangoapps/courseware/block_render.py
index 8ef637b82bf0..db81a6a0ce7a 100644
--- a/lms/djangoapps/courseware/block_render.py
+++ b/lms/djangoapps/courseware/block_render.py
@@ -33,6 +33,7 @@
 from eventtracking import tracker
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey, UsageKey
+from openedx_filters.learning.filters import LearnerXBlockBound
 from rest_framework.decorators import api_view
 from rest_framework.exceptions import APIException
 from web_fragments.fragment import Fragment
@@ -116,6 +117,9 @@
 
 log = logging.getLogger(__name__)
 
+# Marks a block that LearnerXBlockBound has not run on yet (anonymous users have no id).
+_NOT_FILTERED = object()
+
 # TODO: course_id and course_key are used interchangeably in this file, which is wrong.
 # Some brave person should make the variable names consistently someday, but the code's
 # coupled enough that it's kind of tricky--you've been warned!
@@ -487,8 +491,26 @@ def get_block_for_descriptor(
         )
         if access or caller_will_handle_access_error:
             block.has_access_error = bool(caller_will_handle_access_error)
-            return block
+            return _run_learner_xblock_bound(block, user, course_key)
         return None
+    return _run_learner_xblock_bound(block, user, course_key)
+
+
+def _run_learner_xblock_bound(block, user, course_key):
+    """
+    Run the LearnerXBlockBound filter once each time a block is bound to a known user.
+
+    Binding the same block instance to a different user runs it again.
+
+    Unknown users are block structure collection (whose result is cached and
+    shared by every learner) and no-auth callbacks, so their blocks are left as
+    authored. A block that was bound to this user while unknown is filtered the
+    first time it is returned to the known user.
+    """
+    if not getattr(user, 'known', True) or getattr(block, '_learner_xblock_bound_user_id', _NOT_FILTERED) == user.id:
+        return block
+    block = LearnerXBlockBound.run_filter(block=block, user=user, course_key=course_key)
+    block._learner_xblock_bound_user_id = user.id  # pylint: disable=protected-access
     return block
 
 
diff --git a/lms/djangoapps/courseware/tests/test_block_render.py b/lms/djangoapps/courseware/tests/test_block_render.py
index c58283a3d857..57de5b50aceb 100644
--- a/lms/djangoapps/courseware/tests/test_block_render.py
+++ b/lms/djangoapps/courseware/tests/test_block_render.py
@@ -1,6 +1,7 @@
 """
 Test for lms courseware app, block render unit
 """
+import contextlib
 import json
 import textwrap
 import warnings
@@ -87,6 +88,8 @@
 from lms.djangoapps.courseware.tests.tests import LoginEnrollmentTestCase
 from lms.djangoapps.lms_xblock.field_data import LmsFieldData
 from lms.djangoapps.verify_student.tests.factories import SoftwareSecurePhotoVerificationFactory
+from openedx.core.djangoapps.content.block_structure.api import get_block_structure_manager
+from openedx.core.djangoapps.content.block_structure.exceptions import BlockStructureNotFound
 from openedx.core.djangoapps.credit.api import set_credit_requirement_status, set_credit_requirements
 from openedx.core.djangoapps.credit.models import CreditCourse
 from openedx.core.djangoapps.oauth_dispatch.jwt import _create_jwt, create_jwt_for_user
@@ -476,6 +479,99 @@ def test_rebinding_same_user(self, block_type):
             course=self.toy_course
         )
 
+    @ddt.data(
+        ([True, True], 1),
+        ([False, False], 0),
+        ([False, True, True], 1),
+    )
+    @ddt.unpack
+    @patch(
+        'lms.djangoapps.courseware.block_render.LearnerXBlockBound.run_filter',
+        side_effect=lambda **kwargs: kwargs['block'],
+    )
+    def test_learner_xblock_bound_filter(self, known_per_call, expected_calls, run_filter):
+        """The bound filter runs once per binding to a known user, and not for unknown users."""
+        request = self.request_factory.get('')
+        request.user = self.mock_user
+        block = BlockFactory(category='html', parent=CourseFactory())
+        field_data_cache = FieldDataCache([self.toy_course, block], self.toy_course.id, self.mock_user)
+
+        for known in known_per_call:
+            self.mock_user.known = known
+            render.get_block_for_descriptor(
+                self.mock_user, request, block, field_data_cache, self.toy_course.id, course=self.toy_course
+            )
+
+        assert run_filter.call_count == expected_calls
+
+    @patch(
+        'lms.djangoapps.courseware.block_render.LearnerXBlockBound.run_filter',
+        side_effect=lambda **kwargs: kwargs['block'],
+    )
+    def test_learner_xblock_bound_changes_are_not_collected(self, run_filter):
+        """
+        Blocks changed by the bound filter, including children loaded later through the
+        learner's runtime, are not collected into the block structure shared by every user.
+        """
+        def change_display_name(block, **kwargs):  # pylint: disable=unused-argument
+            block.fields['display_name']._set_cached_value(block, 'Changed')  # pylint: disable=protected-access
+            return block
+
+        run_filter.side_effect = change_display_name
+        course = CourseFactory.create()
+        chapter = BlockFactory.create(parent=course, category='chapter', display_name='Authored')
+        BlockFactory.create(parent=chapter, category='sequential', display_name='Authored')
+        request = self.request_factory.get('')
+        request.user = self.mock_user
+        store = modulestore()
+        manager = get_block_structure_manager(course.id)
+        with contextlib.suppress(BlockStructureNotFound):
+            manager.clear()
+
+        with store.bulk_operations(course.id):
+            course_block = store.get_course(course.id, depth=None)
+            field_data_cache = FieldDataCache([course_block], course.id, self.mock_user)
+            bound = render.get_block_for_descriptor(
+                self.mock_user, request, course_block, field_data_cache, course.id, course=course_block
+            )
+            assert bound.display_name == 'Changed'
+            collected = manager.get_collected()
+
+        assert 'Changed' not in {collected.get_xblock_field(key, 'display_name') for key in collected}
+
+    @patch(
+        'lms.djangoapps.courseware.block_render.LearnerXBlockBound.run_filter',
+        side_effect=lambda **kwargs: kwargs['block'],
+    )
+    def test_learner_xblock_bound_filter_runs_again_for_another_user(self, run_filter):
+        """Binding the same block to a different user runs the bound filter again."""
+        block = BlockFactory(category='html', parent=CourseFactory())
+        other_user = UserFactory()
+
+        for user in (self.mock_user, other_user):
+            request = self.request_factory.get('')
+            request.user = user
+            field_data_cache = FieldDataCache([self.toy_course, block], self.toy_course.id, user)
+            render.get_block_for_descriptor(
+                user, request, block, field_data_cache, self.toy_course.id, course=self.toy_course
+            )
+
+        assert [call.kwargs['user'] for call in run_filter.call_args_list] == [self.mock_user, other_user]
+
+    @patch(
+        'lms.djangoapps.courseware.block_render.LearnerXBlockBound.run_filter',
+        side_effect=lambda **kwargs: kwargs['block'],
+    )
+    def test_learner_xblock_bound_filter_anonymous_user(self, run_filter):
+        """The bound filter runs once for an anonymous user, whose id is None."""
+        block = BlockFactory(category='html', parent=CourseFactory())
+        run_bound_filter = render._run_learner_xblock_bound  # pylint: disable=protected-access
+
+        for _ in range(2):
+            run_bound_filter(block, AnonymousUser(), self.toy_course.id)
+
+        assert run_filter.call_count == 1
+
     @override_settings(FIELD_OVERRIDE_PROVIDERS=(
         'lms.djangoapps.courseware.student_field_overrides.IndividualStudentOverrideProvider',
     ))
diff --git a/openedx/core/djangoapps/content/block_structure/manager.py b/openedx/core/djangoapps/content/block_structure/manager.py
index 5335182ba6e8..d260c057985e 100644
--- a/openedx/core/djangoapps/content/block_structure/manager.py
+++ b/openedx/core/djangoapps/content/block_structure/manager.py
@@ -112,8 +112,10 @@ def get_collected(self, user=None):
                 # we do not want to perform access checks. Access checks result in inconsistent blocks where
                 # inaccessible blocks are missing from the cache. Cached course blocks are then used for all the users.
                 user.known = False
-                block_structure = self._update_collected()
-                user.known = True
+                try:
+                    block_structure = self._update_collected()
+                finally:
+                    user.known = True
             else:
                 block_structure = self._update_collected()
 
@@ -133,18 +135,20 @@ def _update_collected(self):
         The store is updated with newly collected transformers data from
         the modulestore.
         """
+        course_key = self.root_block_usage_key.course_key
         with self._bulk_operations():
-            # Always uses published-only branch regardless of CMS or LMS context.
-            with self.modulestore.branch_setting(
-                ModuleStoreEnum.Branch.published_only,
-                self.root_block_usage_key.course_key
-            ):
-                block_structure = BlockStructureFactory.create_from_modulestore(
-                    self.root_block_usage_key,
-                    self.modulestore,
-                )
-
-            BlockStructureTransformers.collect(block_structure)
+            # The collected structure is cached and shared by every user, so it is
+            # never read from block instances this request has already loaded,
+            # which may have been changed in memory for one user.
+            with self.modulestore.fresh_block_reads(course_key):
+                # Always uses published-only branch regardless of CMS or LMS context.
+                with self.modulestore.branch_setting(ModuleStoreEnum.Branch.published_only, course_key):
+                    block_structure = BlockStructureFactory.create_from_modulestore(
+                        self.root_block_usage_key,
+                        self.modulestore,
+                    )
+
+                BlockStructureTransformers.collect(block_structure)
             self.store.add(block_structure)
             return block_structure
 
diff --git a/openedx/core/djangoapps/content/block_structure/tests/helpers.py b/openedx/core/djangoapps/content/block_structure/tests/helpers.py
index 30899a87f398..4ba2cf539c07 100644
--- a/openedx/core/djangoapps/content/block_structure/tests/helpers.py
+++ b/openedx/core/djangoapps/content/block_structure/tests/helpers.py
@@ -117,6 +117,13 @@ def branch_setting(self, branch_settings, course_id=None):  # pylint: disable=un
         """
         yield
 
+    @contextmanager
+    def fresh_block_reads(self, course_key):  # pylint: disable=unused-argument
+        """
+        A context manager for loading blocks as new instances.
+        """
+        yield
+
 
 class MockCache:
     """
diff --git a/openedx/core/djangoapps/content/block_structure/tests/test_manager.py b/openedx/core/djangoapps/content/block_structure/tests/test_manager.py
index 04a08accf6b3..f5e3b2a8a086 100644
--- a/openedx/core/djangoapps/content/block_structure/tests/test_manager.py
+++ b/openedx/core/djangoapps/content/block_structure/tests/test_manager.py
@@ -9,6 +9,9 @@
 from django.test import TestCase
 
 from xmodule.modulestore import ModuleStoreEnum
+from xmodule.modulestore.django import modulestore
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
 
 from ..block_structure import BlockStructureBlockData
 from ..exceptions import UsageKeyNotInBlockStructure
@@ -254,3 +257,27 @@ def mock_branch_setting(branch, course_key):
                 setattr(self.modulestore, attr_name, original_branch_setting)
             elif hasattr(self.modulestore, attr_name):
                 delattr(self.modulestore, attr_name)
+
+
+class TestBlockStructureManagerCollection(ModuleStoreTestCase):
+    """
+    Tests for collecting block structures from the modulestore.
+    """
+
+    def test_collect_ignores_blocks_changed_in_memory(self):
+        """
+        Block instances loaded and changed in memory earlier in a bulk operation are not
+        collected into the block structure, which is cached and shared by every user.
+        """
+        course = CourseFactory.create()
+        chapter = BlockFactory.create(parent=course, category='chapter', display_name='Authored')
+        store = modulestore()
+        manager = BlockStructureManager(course.location, store, MockCache())
+
+        with store.bulk_operations(course.id):
+            loaded = store.get_course(course.id, depth=None).get_children()[0]
+            loaded.display_name = 'Changed in memory'
+            collected = manager.get_collected()
+            assert store.get_item(chapter.location).display_name == 'Changed in memory'
+
+        assert collected.get_xblock_field(chapter.location, 'display_name') == 'Authored'
diff --git a/openedx/core/djangoapps/content/course_overviews/models.py b/openedx/core/djangoapps/content/course_overviews/models.py
index 61dd8898755c..dae0103a6d4c 100644
--- a/openedx/core/djangoapps/content/course_overviews/models.py
+++ b/openedx/core/djangoapps/content/course_overviews/models.py
@@ -312,7 +312,9 @@ def load_from_module_store(cls, course_id):
             course_id,
         )
         store = modulestore()
-        with store.bulk_operations(course_id):
+        # The overview is shared by every user, so it is never built from blocks this
+        # request has already loaded and may have changed in memory.
+        with store.bulk_operations(course_id), store.fresh_block_reads(course_id):
             course = store.get_course(course_id)
             if isinstance(course, CourseBlock):
                 try:
diff --git a/openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py b/openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py
index 95c9e2df965c..b4c5081335a5 100644
--- a/openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py
+++ b/openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py
@@ -1281,3 +1281,25 @@ def test_tabs_deletion_rollback_on_integrity_error(self):
             actual_tabs = {tab.tab_id for tab in course_overview.tab_set.all()}
             assert actual_tabs == expected_tabs
             assert course_overview.display_name != course.display_name
+
+
+class CourseOverviewFreshReadTestCase(ModuleStoreTestCase):
+    """
+    Tests that course overviews are built from the course as authored.
+    """
+
+    def test_load_ignores_course_changed_in_memory(self):
+        """
+        A course instance loaded and changed in memory earlier in a bulk operation is not
+        saved into the course overview, which is shared by every user.
+        """
+        course = CourseFactory.create(display_name='Authored')
+        store = modulestore()
+
+        with store.bulk_operations(course.id):
+            loaded = store.get_course(course.id)
+            loaded.fields['display_name']._set_cached_value(loaded, 'Changed')  # pylint: disable=protected-access
+            assert store.get_course(course.id).display_name == 'Changed'
+            overview = CourseOverview.load_from_module_store(course.id)
+
+        assert overview.display_name == 'Authored'
diff --git a/openedx/core/djangoapps/content/learning_sequences/tests/test_views.py b/openedx/core/djangoapps/content/learning_sequences/tests/test_views.py
index 49d9fafa068f..e9d7beefce1d 100644
--- a/openedx/core/djangoapps/content/learning_sequences/tests/test_views.py
+++ b/openedx/core/djangoapps/content/learning_sequences/tests/test_views.py
@@ -16,7 +16,9 @@
 from datetime import datetime, timezone
 
 import ddt
+from django.test import override_settings
 from opaque_keys.edx.keys import CourseKey
+from openedx_filters import PipelineStep
 from rest_framework.test import APIClient, APITestCase
 
 from common.djangoapps.student.models import CourseEnrollment
@@ -31,6 +33,21 @@
 from ..data import CourseOutlineData, CourseVisibility
 
 
+class TestCourseOutlineRenderStep(PipelineStep):
+    """
+    Utility pipeline step that records the user and changes sequence titles in place.
+    """
+
+    users = []
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        """Record the user and override every sequence title."""
+        self.users.append(user)
+        for sequence in data["outline"]["sequences"].values():
+            sequence["title"] = "Filtered title"
+        return {"data": data}
+
+
 @skip_unless_lms
 class CourseOutlineViewTest(CacheIsolationTestCase, APITestCase):
     """
@@ -119,6 +136,41 @@ def test_outline_as_staff(self):
         assert len(data['outline']['sections'][1]['sequence_ids']) == 2
         assert len(data['outline']['sequences']) == 4
 
+    def test_outline_render_filter(self):
+        """
+        The render filter can modify the response, receives the effective user, and does not change cached data.
+        """
+        filters_config = {
+            "org.openedx.learning.course_outline.api.render.started.v1": {
+                "pipeline": [
+                    "openedx.core.djangoapps.content.learning_sequences.tests.test_views.TestCourseOutlineRenderStep",
+                ],
+                "fail_silently": False,
+            },
+        }
+        CourseOverviewFactory(id=self.course_key)
+        CourseEnrollment.enroll(self.student, self.course_key)
+        TestCourseOutlineRenderStep.users = []
+        with override_settings(OPEN_EDX_FILTERS_CONFIG=filters_config):
+            self.client.login(username='student', password='student_pass')
+            student_result = self.client.get(self.course_url)
+            self.client.login(username='staff', password='staff_pass')
+            staff_as_student_result = self.client.get(self.course_url, {'user': 'student'})
+
+        for result in [student_result, staff_as_student_result]:
+            assert result.status_code == 200
+            titles = {seq['title'] for seq in result.data['outline']['sequences'].values()}
+            assert titles == {"Filtered title"}
+        assert TestCourseOutlineRenderStep.users == [self.student, self.student]
+
+        unfiltered_result = self.client.get(self.course_url)
+        authored_titles = {
+            str(seq.usage_key): seq.title for section in self.outline.sections for seq in section.sequences
+        }
+        assert {
+            key: seq['title'] for key, seq in unfiltered_result.data['outline']['sequences'].items()
+        } == authored_titles
+
 
 @ddt.ddt
 @skip_unless_lms
diff --git a/openedx/core/djangoapps/content/learning_sequences/views.py b/openedx/core/djangoapps/content/learning_sequences/views.py
index e1cee434df89..b86df22daa3b 100644
--- a/openedx/core/djangoapps/content/learning_sequences/views.py
+++ b/openedx/core/djangoapps/content/learning_sequences/views.py
@@ -11,6 +11,7 @@
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys.edx.keys import CourseKey
+from openedx_filters.learning.filters import CourseOutlineAPIRenderStarted
 from rest_framework import serializers
 from rest_framework.exceptions import NotAuthenticated, NotFound, PermissionDenied
 from rest_framework.response import Response
@@ -178,7 +179,8 @@ def get(self, request, course_key_str, format=None):  # pylint: disable=redefine
             raise NotFound() from does_not_exist_err
 
         serializer = self.UserCourseOutlineDataSerializer(user_course_outline_details)
-        return Response(serializer.data)
+        data = CourseOutlineAPIRenderStarted.run_filter(course_key=course_key, user=request.user, data=serializer.data)
+        return Response(data)
 
     def _determine_user(self, request, course_key: CourseKey) -> types.User:
         """
diff --git a/openedx/core/djangoapps/courseware_api/tests/test_views.py b/openedx/core/djangoapps/courseware_api/tests/test_views.py
index 3f4f5fae0ebc..22a5e87b1443 100644
--- a/openedx/core/djangoapps/courseware_api/tests/test_views.py
+++ b/openedx/core/djangoapps/courseware_api/tests/test_views.py
@@ -15,6 +15,7 @@
 from django.test.client import RequestFactory
 from edx_django_utils.cache import TieredCache
 from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
+from openedx_filters import PipelineStep
 
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.course_modes.tests.factories import CourseModeFactory
@@ -50,6 +51,18 @@
 _NEXT_WEEK = datetime.now() + timedelta(days=7)
 
 
+class TestReplaceCoursewareMetadata(PipelineStep):
+    """
+    Utility class used when getting steps for pipeline.
+    """
+
+    def run_filter(self, course_key, user, data):  # pylint: disable=arguments-differ
+        """
+        Pipeline step that replaces the response body with the course key and the user it received.
+        """
+        return {"data": {"course_key": str(course_key), "username": user.username}}
+
+
 @skip_unless_lms
 class BaseCoursewareTests(SharedModuleStoreTestCase):
     """
@@ -428,6 +441,31 @@ def test_can_access_proctored_exams_masquerading(self, masquerade_group_id, resu
         assert 'can_access_proctored_exams' in courseware_data
         assert courseware_data['can_access_proctored_exams'] == result
 
+    @ddt.data(False, True)
+    @override_settings(
+        OPEN_EDX_FILTERS_CONFIG={
+            "org.openedx.learning.courseware_metadata.api.render.started.v1": {
+                "pipeline": [
+                    "openedx.core.djangoapps.courseware_api.tests.test_views.TestReplaceCoursewareMetadata",
+                ],
+                "fail_silently": False,
+            },
+        },
+    )
+    def test_courseware_metadata_filter_replaces_response(self, masquerade):
+        """
+        The filter's returned data is sent to the client, and the filter receives the effective (masqueraded) user.
+        """
+        CourseEnrollment.enroll(self.user, self.course.id, 'audit')
+        if masquerade:
+            self.client.login(username=self.instructor.username, password='foo')
+            self.update_masquerade(username=self.user.username)
+
+        response = self.client.get(self.url)
+
+        assert response.status_code == 200
+        assert response.json() == {"course_key": str(self.course.id), "username": self.user.username}
+
     @ddt.data(
         True,
         False
diff --git a/openedx/core/djangoapps/courseware_api/views.py b/openedx/core/djangoapps/courseware_api/views.py
index 1b0bd0c50f5f..737926e3ce41 100644
--- a/openedx/core/djangoapps/courseware_api/views.py
+++ b/openedx/core/djangoapps/courseware_api/views.py
@@ -12,6 +12,7 @@
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey, UsageKey
+from openedx_filters.learning.filters import CoursewareMetadataAPIRenderStarted
 from rest_framework import status
 from rest_framework.exceptions import NotFound
 from rest_framework.generics import RetrieveAPIView
@@ -697,6 +698,16 @@ def set_last_seen_courseware_timezone(self, user):
                     defaults={'last_seen_courseware_timezone': browser_timezone},
                 )
 
+    def retrieve(self, request, *args, **kwargs):
+        """
+        Return the serialized course, after the CoursewareMetadataAPIRenderStarted filter.
+        """
+        response = super().retrieve(request, *args, **kwargs)
+        response.data = CoursewareMetadataAPIRenderStarted.run_filter(
+            course_key=CourseKey.from_string(self.kwargs['course_key_string']), user=request.user, data=response.data
+        )
+        return response
+
     def get_object(self):
         """
         Return the requested course object, if the user has appropriate
diff --git a/xmodule/modulestore/__init__.py b/xmodule/modulestore/__init__.py
index 79496eb55e16..bb1c6956b0ca 100644
--- a/xmodule/modulestore/__init__.py
+++ b/xmodule/modulestore/__init__.py
@@ -1039,6 +1039,18 @@ def bulk_operations(self, course_id, emit_signals=True, ignore_case=False):    #
         """
         yield
 
+    @contextmanager
+    def fresh_block_reads(self, course_key):  # pylint: disable=unused-argument
+        """
+        A context manager for loading the course's blocks as new instances, ignoring block
+        instances that this thread has already loaded (and may have changed in memory).
+
+        The default is a no-op, which is only correct for stores that build new block instances on
+        every read. A store that hands back already-loaded instances and serves learners must override
+        it.
+        """
+        yield
+
     def ensure_indexes(self):
         """
         Ensure that all appropriate indexes are created that are needed by this modulestore, or raise
diff --git a/xmodule/modulestore/mixed.py b/xmodule/modulestore/mixed.py
index 2a3a04699a79..9dfb09678421 100644
--- a/xmodule/modulestore/mixed.py
+++ b/xmodule/modulestore/mixed.py
@@ -1113,6 +1113,16 @@ def bulk_operations(self, course_id, emit_signals=True, ignore_case=False):
         with store.bulk_operations(course_id, emit_signals, ignore_case):
             yield
 
+    @contextmanager
+    def fresh_block_reads(self, course_key):
+        """
+        A context manager for loading the course's blocks as new instances, ignoring block
+        instances that this thread has already loaded (and may have changed in memory).
+        """
+        store = self._get_modulestore_for_courselike(course_key)
+        with store.fresh_block_reads(course_key):
+            yield
+
     def ensure_indexes(self):
         """
         Ensure that all appropriate indexes are created that are needed by this modulestore, or raise
diff --git a/xmodule/modulestore/split_mongo/split.py b/xmodule/modulestore/split_mongo/split.py
index 5a115dd16176..8687af36c0ed 100644
--- a/xmodule/modulestore/split_mongo/split.py
+++ b/xmodule/modulestore/split_mongo/split.py
@@ -59,6 +59,7 @@
 import datetime
 import logging
 from collections import defaultdict
+from contextlib import contextmanager
 from importlib import import_module
 from zoneinfo import ZoneInfo
 
@@ -222,15 +223,52 @@ def _clear_bulk_ops_record(self, course_key):
         """
         Clear the record for this course
         """
+        del self._active_bulk_ops.records[self._bulk_ops_record_key(course_key)]
+
+    def _bulk_ops_record_key(self, course_key):
+        """
+        Return the key of this course's record in the active bulk operations.
+        """
         if not isinstance(course_key, (CourseLocator, LibraryLocator)):
             raise TypeError(f'{course_key!r} is not a CourseLocator or LibraryLocator')
 
         if course_key.org and get_library_or_course_attribute(course_key) and course_key.run:
-            del self._active_bulk_ops.records[course_key.replace(branch=None, version_guid=None)]
-        else:
-            del self._active_bulk_ops.records[
-                course_key.replace(org=None, course=None, run=None, branch=None)
-            ]
+            return course_key.replace(branch=None, version_guid=None)
+        return course_key.replace(org=None, course=None, run=None, branch=None)
+
+    @contextmanager
+    def fresh_block_reads(self, course_key):
+        """
+        A context manager for loading the course's blocks as new instances, ignoring block
+        instances that this thread has already loaded (and may have changed in memory).
+
+        Set aside, and restored on exit:
+
+        * the blocks cached by the course's active bulk operation, which hand back
+          already-loaded instances;
+        * the request's runtimes (for every course), which may have been prepared for a
+          user and then bind each child block they load to that user.
+
+        Cached structures and definitions are still used. Meant for reads: blocks cached
+        by writes made inside the context are dropped on exit.
+        """
+        # Looked up without indexing, which would add a record to the defaultdict.
+        bulk_ops_record = self._active_bulk_ops.records.get(self._bulk_ops_record_key(course_key))
+        cache_data = self.request_cache.data if self.request_cache is not None else {}
+        saved_modules = bulk_ops_record.modules if bulk_ops_record is not None else None
+        saved_runtimes = cache_data.get('course_cache')
+        if bulk_ops_record is not None:
+            bulk_ops_record.modules = defaultdict(dict)
+        cache_data['course_cache'] = {}
+        try:
+            yield
+        finally:
+            if bulk_ops_record is not None:
+                bulk_ops_record.modules = saved_modules
+            if saved_runtimes is None:
+                cache_data.pop('course_cache', None)
+            else:
+                cache_data['course_cache'] = saved_runtimes
 
     def _start_outermost_bulk_operation(self, bulk_write_record, course_key, ignore_case=False):  # pylint: disable=arguments-differ
         """
diff --git a/xmodule/modulestore/xml.py b/xmodule/modulestore/xml.py
index eddc525970f5..f283cff91b32 100644
--- a/xmodule/modulestore/xml.py
+++ b/xmodule/modulestore/xml.py
@@ -910,6 +910,16 @@ def make_course_usage_key(self, course_key):
         """
         return BlockUsageLocator(course_key, 'course', course_key.run)
 
+    @contextmanager
+    def fresh_block_reads(self, course_key):  # pylint: disable=unused-argument
+        """
+        A no-op, although this store hands back the same block instances on every read.
+
+        This store only feeds course import and management commands; it is never configured to
+        serve learners, so its blocks are not bound to users or changed in memory per user.
+        """
+        yield
+
     def get_courses(self, **kwargs):
         """
         Returns a list of course blocks.  If there were errors on loading,