diff --git a/rx/core/operators/observeon.py b/rx/core/operators/observeon.py
index d725401b..d1ea4a56 100644
--- a/rx/core/operators/observeon.py
+++ b/rx/core/operators/observeon.py
@@ -22,8 +22,9 @@ def _observe_on(scheduler) -> Callable[[Observable], Observable]:
             Returns the source sequence whose observations happen on
             the specified scheduler.
         """
-        def subscribe(observer, _=None):
-            return source.subscribe(ObserveOnObserver(scheduler, observer))
+        def subscribe(observer, subscribe_scheduler=None):
+            return source.subscribe(ObserveOnObserver(scheduler, observer),
+                                    scheduler=subscribe_scheduler)
 
         return Observable(subscribe)
-    return observe_on
\ No newline at end of file
+    return observe_on

observe_on operator does not propagate subscription scheduler
The observe_on scheduler drops the scheduler parameter provided during subscription. As a consequence, the default scheduler provided in the pipe operator is lost when observe_on is used in the chain.
diff --git a/tests/test_observable/test_observeon.py b/tests/test_observable/test_observeon.py
index 450ed134..09d265b4 100644
--- a/tests/test_observable/test_observeon.py
+++ b/tests/test_observable/test_observeon.py
@@ -2,6 +2,7 @@ import unittest
 
 import rx
 from rx import operators as ops
+from rx.concurrency import ImmediateScheduler
 from rx.testing import TestScheduler, ReactiveTest
 
 on_next = ReactiveTest.on_next
@@ -75,3 +76,21 @@ class TestObserveOn(unittest.TestCase):
 
         assert results.messages == []
         assert xs.subscriptions == [subscribe(200, 1000)]
+
+    def test_observe_on_forward_subscribe_scheduler(self):
+        scheduler = ImmediateScheduler()
+        expected_subscribe_scheduler = ImmediateScheduler()
+
+        actual_subscribe_scheduler = None
+
+        def subscribe(observer, scheduler):
+            nonlocal actual_subscribe_scheduler
+            actual_subscribe_scheduler = scheduler
+            observer.on_completed()
+
+        xs = rx.create(subscribe)
+
+        xs.pipe(ops.observe_on(scheduler)).subscribe(
+            scheduler=expected_subscribe_scheduler)
+
+        assert expected_subscribe_scheduler == actual_subscribe_scheduler

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-asyncio pytest-cov
apt-get update apt-get install -y libgirepository1.0-dev gir1.2-gtk-3.0
3.9
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
coverage==7.8.0
exceptiongroup==1.2.2
iniconfig==2.1.0
packaging==24.2
pluggy==1.5.0
pytest==8.3.5
pytest-asyncio==0.26.0
pytest-cov==6.0.0
-e git+https://github.com/ReactiveX/RxPY.git@a964eac88787dabef61cbc96ba31801f5a9bbed9#egg=Rx
tomli==2.2.1
typing_extensions==4.13.0

name: RxPY
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - coverage==7.8.0
      - exceptiongroup==1.2.2
      - iniconfig==2.1.0
      - packaging==24.2
      - pluggy==1.5.0
      - pytest==8.3.5
      - pytest-asyncio==0.26.0
      - pytest-cov==6.0.0
      - tomli==2.2.1
      - typing-extensions==4.13.0
prefix: /opt/conda/envs/RxPY

tests/test_observable/test_observeon.py::TestObserveOn::test_observe_on_forward_subscribe_scheduler
tests/test_observable/test_observeon.py::TestObserveOn::test_observe_on_empty tests/test_observable/test_observeon.py::TestObserveOn::test_observe_on_error tests/test_observable/test_observeon.py::TestObserveOn::test_observe_on_never tests/test_observable/test_observeon.py::TestObserveOn::test_observe_on_normal
rx/core/operators/observeon.py
rx/core/operators/observeon.py