From de707fd464dd08bd10f2441d35fdc139a9a081d4 Mon Sep 17 00:00:00 2001
From: Bortlesboat <169967362+Bortlesboat@users.noreply.github.com>
Date: Tue, 8 Sep 2026 09:56:33 -0400
Subject: [PATCH] fix: propagate control-flow exceptions from async helper
 thread

---
 langfuse/_client/utils.py |  4 ++--
 tests/unit/test_utils.py  | 31 +++++++++++++++++++++++++++++++
 2 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/langfuse/_client/utils.py b/langfuse/_client/utils.py
index 9ef9a767d..2afa09651 100644
--- a/langfuse/_client/utils.py
+++ b/langfuse/_client/utils.py
@@ -96,13 +96,13 @@ def __init__(self, coro: Coroutine[Any, Any, Any]) -> None:
         self.coro = coro
         self.context = contextvars.copy_context()
         self.result: Any = None
-        self.exception: Exception | None = None
+        self.exception: BaseException | None = None
         super().__init__()
 
     def run(self) -> None:
         try:
             self.result = self.context.run(asyncio.run, self.coro)
-        except Exception as e:
+        except BaseException as e:
             self.exception = e
 
 
diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py
index 968bcb91b..96e0c0338 100644
--- a/tests/unit/test_utils.py
+++ b/tests/unit/test_utils.py
@@ -13,6 +13,37 @@
 class TestRunAsyncSafely:
     """Test suite for the run_async_safely function."""
 
+    @pytest.mark.parametrize(
+        "error_type", [asyncio.CancelledError, SystemExit, KeyboardInterrupt]
+    )
+    def test_control_flow_exception_in_sync_context(self, error_type):
+        """Control-flow exceptions retain their identity without an active loop."""
+        error = error_type("operation interrupted")
+
+        async def interrupted():
+            raise error
+
+        with pytest.raises(error_type) as caught:
+            run_async_safely(interrupted())
+
+        assert caught.value is error
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "error_type", [asyncio.CancelledError, SystemExit, KeyboardInterrupt]
+    )
+    async def test_control_flow_exception_from_thread(self, error_type):
+        """A worker's cancellation or exit is propagated to the calling thread."""
+        error = error_type("operation interrupted")
+
+        async def interrupted():
+            raise error
+
+        with pytest.raises(error_type) as caught:
+            run_async_safely(interrupted())
+
+        assert caught.value is error
+
     def test_run_sync_context_simple(self):
         """Test run_async_safely in sync context with simple coroutine."""