diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index 7308a4ef..be35b4fb 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        CONDA_ENV: [py39, py310, py311, py312, py313]
+        CONDA_ENV: [py39, py310, py311, py312, py313, py314]
     # env:
       # STREAMZ_LAUNCH_KAFKA: true
 
diff --git a/ci/environment-py314.yml b/ci/environment-py314.yml
new file mode 100644
index 00000000..68b4ea71
--- /dev/null
+++ b/ci/environment-py314.yml
@@ -0,0 +1,27 @@
+name: test_env
+channels:
+  - conda-forge
+  - defaults
+dependencies:
+  - python=3.14
+  - pytest
+  - flake8
+  - black
+  - isort
+  - tornado
+  - toolz
+  - librdkafka
+  - dask
+  - distributed
+  - pandas
+  - python-confluent-kafka
+  - codecov
+  - coverage
+  - networkx
+  - graphviz
+  - pytest-asyncio
+  - python-graphviz
+  - bokeh
+  - ipywidgets
+  - flaky
+  - pytest-cov
diff --git a/streamz/orderedweakset.py b/streamz/orderedweakset.py
index ba4b3649..3d4e9a2b 100644
--- a/streamz/orderedweakset.py
+++ b/streamz/orderedweakset.py
@@ -26,6 +26,9 @@ def add(self, value):
     def discard(self, value):
         self._od.pop(value, None)
 
+    def copy(self):
+        return OrderedSet(self._od.copy())
+
 
 class OrderedWeakrefSet(weakref.WeakSet):
     def __init__(self, values=()):
diff --git a/streamz/tests/test_sources.py b/streamz/tests/test_sources.py
index b0ad87ba..8210a8d8 100644
--- a/streamz/tests/test_sources.py
+++ b/streamz/tests/test_sources.py
@@ -103,7 +103,7 @@ def test_http():
 def test_process():
     cmd = ["python", "-c", "for i in range(4): print(i, end='')"]
     s = Source.from_process(cmd, with_end=True)
-    if sys.platform != "win32":
+    if sys.platform != "win32" and sys.version_info < (3, 14):
         # don't know why - something with pytest and new processes
         policy = asyncio.get_event_loop_policy()
         watcher = asyncio.SafeChildWatcher()
@@ -119,7 +119,7 @@ def test_process():
 def test_process_str():
     cmd = 'python -c "for i in range(4): print(i)"'
     s = Source.from_process(cmd)
-    if sys.platform != "win32":
+    if sys.platform != "win32" and sys.version_info < (3, 14):
         # don't know why - something with pytest and new processes
         policy = asyncio.get_event_loop_policy()
         watcher = asyncio.SafeChildWatcher()