diff --git a/cms/djangoapps/contentstore/management/commands/migrate_course_legacy_library_blocks_to_item_bank.py b/cms/djangoapps/contentstore/management/commands/migrate_course_legacy_library_blocks_to_item_bank.py
new file mode 100644
index 000000000000..65ab416204ad
--- /dev/null
+++ b/cms/djangoapps/contentstore/management/commands/migrate_course_legacy_library_blocks_to_item_bank.py
@@ -0,0 +1,98 @@
+"""
+Management command to migrate legacy library content blocks to Item Bank blocks for course(s).
+
+This command can be run for a specific list of courses or for all courses.
+"""
+from __future__ import annotations
+
+import logging
+
+from django.contrib.auth.models import User  # pylint: disable=imported-auth-user
+from django.core.management.base import BaseCommand, CommandError
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
+
+from cms.djangoapps.contentstore.tasks import migrate_course_legacy_library_blocks_to_item_bank
+from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
+from xmodule.modulestore.django import modulestore  # pylint: disable=wrong-import-order
+
+log = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Migrate legacy library content blocks to Item Bank blocks for course(s).
+
+    Examples:
+        # Migrate specific courses.
+        $ ./manage.py cms migrate_course_legacy_library_blocks_to_item_bank \
+        --course-ids course-v1:edX+DemoX+2024,course-v1:edX+Demo2+2024 --user-id 3
+
+        # Migrate all courses.
+        $ ./manage.py cms migrate_course_legacy_library_blocks_to_item_bank --all-courses --user-id 3
+
+        # Migrate all courses, re-publishing blocks that were published before the migration.
+        $ ./manage.py cms migrate_course_legacy_library_blocks_to_item_bank --all-courses --user-id 3 \
+        --persist-publish-state
+    """
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--course-ids',
+            help='Comma-separated list of course keys to migrate.',
+        )
+        parser.add_argument(
+            '--all-courses',
+            action='store_true',
+            help='Migrate legacy library content blocks for all courses.',
+        )
+        parser.add_argument(
+            '--user-id',
+            type=int,
+            required=True,
+            help='ID of the user performing the migration.',
+        )
+        parser.add_argument(
+            '--persist-publish-state',
+            action='store_true',
+            help='Re-publish blocks that were published before the migration. Defaults to False.',
+        )
+
+    def handle(self, *args, **options):
+        course_ids = options['course_ids']
+        all_courses = options['all_courses']
+        user_id = options['user_id']
+        persist_publish_state = options['persist_publish_state']
+
+        if not course_ids and not all_courses:
+            raise CommandError('Either --course-ids or --all-courses argument should be provided.')
+        if course_ids and all_courses:
+            raise CommandError('Only one of --course-ids or --all-courses argument should be provided.')
+
+        try:
+            User.objects.get(id=user_id)
+        except User.DoesNotExist:
+            raise CommandError(f'No user found with id: {user_id}')  # pylint: disable=raise-missing-from  # noqa: B904
+
+        if all_courses:
+            raw_course_ids = CourseOverview.get_all_course_keys()
+        else:
+            raw_course_ids = [course_id.strip() for course_id in course_ids.split(',') if course_id.strip()]
+
+        course_keys = []
+        for raw_course_id in raw_course_ids:
+            try:
+                course_key = CourseKey.from_string(str(raw_course_id))
+            except InvalidKeyError:
+                log.error(f'Invalid course key: {raw_course_id}, skipping..')
+                continue
+            if not all_courses and not modulestore().get_course(course_key):
+                log.warning(f'Course not found: {course_key}, skipping..')
+                continue
+            course_keys.append(course_key)
+
+        for course_key in course_keys:
+            log.info(f'Dispatching legacy library migration for course: {course_key}')
+            migrate_course_legacy_library_blocks_to_item_bank.delay(user_id, str(course_key), persist_publish_state)
+
+        log.info(f'Dispatched migration for {len(course_keys)} course(s)')
diff --git a/cms/djangoapps/contentstore/management/commands/tests/test_migrate_course_legacy_library_blocks_to_item_bank.py b/cms/djangoapps/contentstore/management/commands/tests/test_migrate_course_legacy_library_blocks_to_item_bank.py
new file mode 100644
index 000000000000..7a5fb2051419
--- /dev/null
+++ b/cms/djangoapps/contentstore/management/commands/tests/test_migrate_course_legacy_library_blocks_to_item_bank.py
@@ -0,0 +1,108 @@
+"""
+Tests for `migrate_course_legacy_library_blocks_to_item_bank` Studio (cms) management command.
+"""
+from unittest import mock
+
+import ddt
+from django.core.management import CommandError, call_command
+
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase  # pylint: disable=wrong-import-order
+from xmodule.modulestore.tests.factories import CourseFactory  # pylint: disable=wrong-import-order
+
+
+@ddt.ddt
+class MigrateCourseLegacyLibraryBlocksToItemBankTests(ModuleStoreTestCase):
+    """ Tests for the `migrate_course_legacy_library_blocks_to_item_bank` management command. """
+    TASK_PATCH_LOCATION = (
+        'cms.djangoapps.contentstore.management.commands.migrate_course_legacy_library_blocks_to_item_bank'
+        '.migrate_course_legacy_library_blocks_to_item_bank'
+    )
+
+    def setUp(self):
+        """ Setup method - create a user and courses to migrate """
+        super().setUp()
+        self.user = UserFactory()
+        self.first_course = CourseFactory.create()
+        self.second_course = CourseFactory.create()
+
+    def _call_command(self, **options):
+        """ Invoke the command, defaulting `user_id` to a valid user. """
+        options.setdefault('user_id', self.user.id)
+        call_command('migrate_course_legacy_library_blocks_to_item_bank', **options)
+
+    @ddt.data(
+        ({}, 'Either --course-ids or --all-courses argument should be provided.'),
+        (
+            {'course_ids': 'course-v1:test+course+run', 'all_courses': True},
+            'Only one of --course-ids or --all-courses argument should be provided.',
+        ),
+    )
+    @ddt.unpack
+    def test_invalid_course_selector_raises_command_error(self, options, expected_message):
+        """ Test that specifying neither, or both, of --course-ids/--all-courses raises a CommandError. """
+        with self.assertRaisesRegex(CommandError, expected_message):  # noqa: PT027
+            self._call_command(**options)
+
+    def test_invalid_user_id_raises_command_error(self):
+        """ Test that an unknown --user-id raises a CommandError. """
+        invalid_user_id = self.user.id + 1000
+        with self.assertRaisesRegex(CommandError, f'No user found with id: {invalid_user_id}'):  # noqa: PT027
+            call_command(
+                'migrate_course_legacy_library_blocks_to_item_bank',
+                all_courses=True,
+                user_id=invalid_user_id,
+            )
+
+    @ddt.data(
+        'invalid_key',
+        'course-v1:test+nonexistent+run',
+    )
+    def test_unparsable_or_nonexistent_course_id_is_skipped(self, bad_course_id):
+        """
+        Test that an unparsable or nonexistent course key passed via --course-ids is skipped,
+        while other, valid, course keys are still dispatched.
+        """
+        course_ids = f'{bad_course_id},{self.first_course.id}'
+        with mock.patch(self.TASK_PATCH_LOCATION) as patched_task:
+            self._call_command(course_ids=course_ids)
+
+        patched_task.delay.assert_called_once_with(self.user.id, str(self.first_course.id), False)
+
+    def test_course_ids_dispatches_task_for_each_course(self):
+        """ Test that the task is dispatched once per course key passed via --course-ids. """
+        course_ids = f'{self.first_course.id},{self.second_course.id}'
+        with mock.patch(self.TASK_PATCH_LOCATION) as patched_task:
+            self._call_command(course_ids=course_ids)
+
+        expected_calls = [
+            mock.call(self.user.id, str(self.first_course.id), False),
+            mock.call(self.user.id, str(self.second_course.id), False),
+        ]
+        self.assertEqual(patched_task.delay.mock_calls, expected_calls)  # noqa: PT009
+
+    def test_all_courses_dispatches_task_for_every_course(self):
+        """ Test that --all-courses dispatches the task for every course known to CourseOverview. """
+        CourseOverviewFactory(id=self.first_course.id)
+        CourseOverviewFactory(id=self.second_course.id)
+
+        with mock.patch(self.TASK_PATCH_LOCATION) as patched_task:
+            self._call_command(all_courses=True)
+
+        expected_calls = [
+            mock.call(self.user.id, str(self.first_course.id), False),
+            mock.call(self.user.id, str(self.second_course.id), False),
+        ]
+        self.assertCountEqual(patched_task.delay.mock_calls, expected_calls)  # noqa: PT009
+
+    @ddt.data(True, False)
+    def test_persist_publish_state_passed_to_task(self, persist_publish_state):
+        """ Test that --persist-publish-state is forwarded to the task, defaulting to False. """
+        with mock.patch(self.TASK_PATCH_LOCATION) as patched_task:
+            self._call_command(
+                course_ids=str(self.first_course.id),
+                persist_publish_state=persist_publish_state,
+            )
+
+        patched_task.delay.assert_called_once_with(self.user.id, str(self.first_course.id), persist_publish_state)
diff --git a/cms/djangoapps/contentstore/tasks.py b/cms/djangoapps/contentstore/tasks.py
index 4d8e02dfd6be..db0481288a32 100644
--- a/cms/djangoapps/contentstore/tasks.py
+++ b/cms/djangoapps/contentstore/tasks.py
@@ -2340,12 +2340,21 @@ def _cancel_old_tasks(course_key: str, user: User, ignore_task_ids: list[str]):
 
 
 @shared_task(base=LegacyLibraryContentToItemBank, bind=True)
-def migrate_course_legacy_library_blocks_to_item_bank(self, user_id: int, course_key: str):
+def migrate_course_legacy_library_blocks_to_item_bank(
+    self, user_id: int, course_key: str, persist_publish_state: bool = False,
+):
     """
     Migrate legacy course library blocks to Item Bank.
 
     Depending on the number of blocks and its children blocks this operation can take a significant
     amount of time and this is why it is run as a celery task.
+
+    Arguments:
+        user_id: id of the user performing the migration.
+        course_key: the course whose legacy library content blocks should be migrated.
+        persist_publish_state: if True, blocks that were published before the migration
+            (and had no unpublished changes) are re-published afterward. Defaults to False,
+            leaving migrated blocks as drafts.
     """
     ensure_cms("Legacy library content references may only be executed in CMS")
     set_code_owner_attribute_from_module(__name__)
@@ -2363,7 +2372,9 @@ def migrate_course_legacy_library_blocks_to_item_bank(self, user_id: int, course
         with store.bulk_operations(key):
             for block in blocks:
                 self.status.set_state(f'Migrating block: {block.usage_key}')
-                block.v2_update_children_upstream_version(user_id)
+                block.v2_update_children_upstream_version(
+                    user_id, persist_publish_state=persist_publish_state
+                )
     except Exception as exc:  # pylint: disable=broad-except
         LOGGER.exception(f'Error while migrating blocks: {exc}')
         self.status.fail(str(exc))
diff --git a/xmodule/library_content_block.py b/xmodule/library_content_block.py
index 995eafad19d5..2c72c5ebe64a 100644
--- a/xmodule/library_content_block.py
+++ b/xmodule/library_content_block.py
@@ -329,14 +329,19 @@ def studio_post_paste(self, store, source_node) -> bool:
         self.sync_from_library(upgrade_to_latest=False)
         return True  # Children have been handled
 
-    def v2_update_children_upstream_version(self, user_id=None):
+    def v2_update_children_upstream_version(self, user_id=None, persist_publish_state=False):
         """
         Update the upstream and upstream version fields of all children to point to library v2 version of the legacy
         library blocks. This essentially converts this legacy block to new ItemBankBlock.
+
+        If `persist_publish_state` is True, and this block was published prior to the migration
+        (with no unpublished changes), it is re-published afterward so that the
+        upstream/upstream_version changes reach LMS.
         """
         from cms.djangoapps.modulestore_migrator import api as migrator_api
         store = modulestore()
         with store.bulk_operations(self.course_id):
+            was_published = persist_publish_state and not store.has_changes(self)
             children = self.get_children()
             # These are the v1 library item upstream UsageKeys
             child_old_upstream_keys = [
@@ -358,6 +363,8 @@ def v2_update_children_upstream_version(self, user_id=None):
             self.is_migrated_to_v2 = True
             self.save()
             store.update_item(self, user_id)
+            if was_published:
+                store.publish(self.location, user_id)
 
     def _validate_library_version(self, validation, lib_tools, version, library_key):
         """
diff --git a/xmodule/tests/test_library_content.py b/xmodule/tests/test_library_content.py
index d5a7fcbc1cc5..793a30dda591 100644
--- a/xmodule/tests/test_library_content.py
+++ b/xmodule/tests/test_library_content.py
@@ -826,3 +826,69 @@ def test_author_view(self):
         assert '<li>html 2</li>' in rendered.content
         assert '<li>html 3</li>' in rendered.content
         assert '<li>html 4</li>' in rendered.content
+
+
+@ddt.ddt
+class TestLegacyLibraryContentBlockMigrationPublishing(LegacyLibraryContentTest):
+    """
+    Unit tests for the `persist_publish_state` flag of
+    LegacyLibraryContentBlock.v2_update_children_upstream_version.
+    """
+
+    def setUp(self):
+        from cms.djangoapps.modulestore_migrator import api
+        from cms.djangoapps.modulestore_migrator.data import CompositionLevel, RepeatHandlingStrategy
+        super().setUp()
+        user = UserFactory()
+        self._sync_lc_block_from_library()
+        self.organization = OrganizationFactory(short_name="myorg")
+        lib_api.create_library(
+            org=self.organization,
+            slug="mylib",
+            title="My Test V2 Library",
+        )
+        self.library_v2 = lib_api.ContentLibrary.objects.get(slug="mylib")
+        api.start_migration_to_library(
+            user=user,
+            source_key=self.library.location.library_key,
+            target_library_key=self.library_v2.library_key,
+            target_collection_slug=None,
+            composition_level=CompositionLevel.Component,
+            repeat_handling_strategy=RepeatHandlingStrategy.Skip,
+            preserve_url_slugs=True,
+            forward_source_to_target=True,
+        )
+
+    @ddt.data(
+        # Published before migration, flag True: re-published with the migration reflected.
+        (True, True),
+        # Published before migration, flag False (default): published branch is left untouched.
+        (True, False),
+        # Never published before migration, flag True: stays unpublished.
+        (False, True),
+    )
+    @ddt.unpack
+    def test_persist_publish_state(self, was_published_before, persist_publish_state):
+        """
+        Tests the `persist_publish_state` flag of `v2_update_children_upstream_version` under
+        the various combinations of prior publish state and flag value.
+        """
+        if was_published_before:
+            self.store.publish(self.course.location, self.user_id)
+
+        self.lc_block.v2_update_children_upstream_version(
+            self.user_id, persist_publish_state=persist_publish_state,
+        )
+
+        if was_published_before and persist_publish_state:
+            with self.store.branch_setting(ModuleStoreEnum.Branch.published_only):
+                published_block = self.store.get_item(self.lc_block.location)
+                assert published_block.is_migrated_to_v2 is True
+                assert published_block.get_children()[0].upstream == "lb:myorg:mylib:html:html_1"
+        elif was_published_before and not persist_publish_state:
+            with self.store.branch_setting(ModuleStoreEnum.Branch.published_only):
+                published_block = self.store.get_item(self.lc_block.location)
+                # The published version still reflects the pre-migration state.
+                assert published_block.is_migrated_to_v2 is False
+        else:
+            assert not self.store.has_published_version(self.store.get_item(self.lc_block.location))