diff --git a/lms/djangoapps/course_api/blocks/transformers/block_completion.py b/lms/djangoapps/course_api/blocks/transformers/block_completion.py
index 472555c4c7f9..c22fae869946 100644
--- a/lms/djangoapps/course_api/blocks/transformers/block_completion.py
+++ b/lms/djangoapps/course_api/blocks/transformers/block_completion.py
@@ -43,18 +43,22 @@ def get_block_completion(cls, block_structure, block_key):
 
     @classmethod
     def collect(cls, block_structure):
-        block_structure.request_xblock_fields('completion_mode')
+        block_structure.request_xblock_fields('completion_mode', 'published_on',)
 
     @staticmethod
-    def _is_block_excluded(block_structure, block_key):
+    def _is_block_excluded_or_unpublished(block_structure, block_key):
         """
         Checks whether block's completion method is of `EXCLUDED` type.
+        Or if the block is unpublished.
         """
         completion_mode = block_structure.get_xblock_field(
             block_key, 'completion_mode'
         )
-
-        return completion_mode == CompletionMode.EXCLUDED
+        published_on = block_structure.get_xblock_field(
+            block_key, 'published_on'
+        )
+        # Exclude if the block is unpublished
+        return completion_mode == CompletionMode.EXCLUDED or published_on is None
 
     def mark_complete(self, complete_course_blocks, latest_complete_block_key, block_key, block_structure):
         """
@@ -75,7 +79,7 @@ def mark_complete(self, complete_course_blocks, latest_complete_block_key, block
             children = block_structure.get_children(block_key)
             all_children_complete = all(block_structure.get_xblock_field(child_key, self.COMPLETE)
                                         for child_key in children
-                                        if not self._is_block_excluded(block_structure, child_key))
+                                        if not self._is_block_excluded_or_unpublished(block_structure, child_key))
 
             if all_children_complete:
                 block_structure.override_xblock_field(block_key, self.COMPLETE, True)
diff --git a/lms/djangoapps/course_api/blocks/transformers/tests/test_block_completion.py b/lms/djangoapps/course_api/blocks/transformers/tests/test_block_completion.py
index 1bea04b0486c..2f1763d0124c 100644
--- a/lms/djangoapps/course_api/blocks/transformers/tests/test_block_completion.py
+++ b/lms/djangoapps/course_api/blocks/transformers/tests/test_block_completion.py
@@ -12,6 +12,10 @@
 from lms.djangoapps.course_api.blocks.transformers.block_completion import BlockCompletionTransformer
 from lms.djangoapps.course_blocks.api import get_course_blocks
 from lms.djangoapps.course_blocks.transformers.tests.helpers import ModuleStoreTestCase, TransformerRegistryTestMixin
+from lms.djangoapps.course_blocks.usage_info import CourseUsageInfo
+from openedx.core.djangoapps.content.block_structure.factory import BlockStructureFactory
+from openedx.core.djangoapps.content.block_structure.transformers import BlockStructureTransformers
+from xmodule.modulestore import ModuleStoreEnum
 from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory  # lint-amnesty, pylint: disable=wrong-import-order
 
 
@@ -127,6 +131,65 @@ def test_transform_gives_zero_for_ordinary_block(self):
 
         self._assert_block_has_proper_completion_values(block_structure, block.location, 0.0, False)
 
+    @XBlock.register_temp_plugin(StubAggregatorXBlock, identifier='aggregator')
+    @XBlock.register_temp_plugin(StubCompletableXBlock, identifier='comp')
+    def test_transform_with_unpublished_blocks_in_structure(self):
+        """
+        Test that when unpublished blocks are present in the block structure
+        (as might happen when CMS adds them to cache), the transformer correctly
+        excludes them from completion calculations by checking published_on field.
+
+        This is a test-only scenario to verify the fix works correctly.
+        """
+        from xmodule.modulestore.django import modulestore
+
+        course = CourseFactory.create()
+        aggregator = BlockFactory.create(category='aggregator', parent=course)
+        published_block = BlockFactory.create(category='comp', parent=aggregator)
+        BlockCompletion.objects.submit_completion(
+            user=self.user,
+            block_key=published_block.location,
+            completion=self.COMPLETION_TEST_VALUE,
+        )
+
+        # Create an unpublished block
+        unpublished_block = BlockFactory.create(category='comp', parent=aggregator, publish_item=False)
+        # Simulate a completion submission for the unpublished block (Should be ignored)
+        BlockCompletion.objects.submit_completion(
+            user=self.user,
+            block_key=unpublished_block.location,
+            completion=0.0,
+        )
+
+        # Now create a block structure that includes both published and unpublished blocks
+        # This simulates what happens when CMS modifies the cache
+        store = modulestore()
+        with store.branch_setting(ModuleStoreEnum.Branch.draft_preferred, course.id):
+            # Get the course with all blocks (published and unpublished)
+            draft_course = store.get_course(course.id, depth=None)
+            # Create block structure from the draft course (includes unpublished)
+            block_structure = BlockStructureFactory.create_from_modulestore(
+                root_block_usage_key=draft_course.location,
+                modulestore=store
+            )
+            usage_info = CourseUsageInfo(
+                user=self.user,
+                course_key=course.id,
+            )
+            transformers = BlockStructureTransformers([self.TRANSFORMER_CLASS_TO_TEST()], usage_info)
+            transformers.collect(block_structure)
+            transformers.transform(block_structure)
+
+        self._assert_block_has_proper_completion_values(
+            block_structure, published_block.location, self.COMPLETION_TEST_VALUE, True
+        )
+        self._assert_block_has_proper_completion_values(
+            block_structure, unpublished_block.location, 0.0, False
+        )
+        self._assert_block_has_proper_completion_values(
+            block_structure, aggregator.location, None, True
+        )
+
     def _assert_block_has_proper_completion_values(
             self, block_structure, block_key, expected_completion, expected_complete
     ):