diff --git a/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py b/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py
index 3d4c9eee26a..a67949e6717 100644
--- a/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py
+++ b/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py
@@ -33,7 +33,8 @@ def __init__(self):
         self.finished_current: Event = Event()
 
     def get_internal_marker(self) -> Optional[InternalMarker]:
-        return self.current_internal_marker
+        ret, self.current_internal_marker = self.current_internal_marker, None
+        return ret
 
     def get_input_tuple(self) -> Optional[Tuple]:
         ret, self.current_input_tuple = self.current_input_tuple, None
diff --git a/amber/src/main/python/core/runnables/main_loop.py b/amber/src/main/python/core/runnables/main_loop.py
index 7b22e086069..9356542a08c 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -342,7 +342,6 @@ def _process_ecm(self, ecm_element: ECMElement):
                     StartChannel: self._process_start_channel,
                     EndChannel: self._process_end_channel,
                 }[type(self.context.tuple_processing_manager.current_internal_marker)]()
-                self.context.tuple_processing_manager.current_internal_marker = None
 
     def _send_ecm_to_data_channels(
         self, method_name: str, alignment: EmbeddedControlMessageType
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala
index c13a5c13aa5..92dfba19de1 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala
@@ -256,11 +256,7 @@ class ReconfigurationSpec
     )
   }
 
-  // Disabled: the workflow hangs after the UDF processes the EndChannel ECM in this
-  // multi-worker (Python source -> Python UDF) propagation case, causing a 1-minute
-  // Await timeout. The single-op reconfigure path and the CSV-source variants pass.
-  // Re-enable once the source-propagation completion path is fixed.
-  "Engine" should "propagate reconfiguration through a source operator in workflow" ignore {
+  "Engine" should "propagate reconfiguration through a source operator in workflow" in {
     val sourceOpDesc = TestOperators.pythonSourceOpDesc(10000)
     val udfOpDesc = TestOperators.pythonOpDesc()
     val code = """