From b48ae7157f96516614822e31b6a1d5f494b60ebd Mon Sep 17 00:00:00 2001
From: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
Date: Mon, 8 Sep 2025 05:06:11 -0400
Subject: [PATCH] Fixed bridge error

---
 src/cleveragents/core/application.py |  2 ++
 src/cleveragents/langgraph/bridge.py | 22 +++++++++++++++++++---
 2 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/src/cleveragents/core/application.py b/src/cleveragents/core/application.py
index 7793205b5d..7d5e70ff67 100644
--- a/src/cleveragents/core/application.py
+++ b/src/cleveragents/core/application.py
@@ -98,6 +98,8 @@ class ReactiveCleverAgentsApp:
 
         # Initialize LangGraph bridge
         self.langgraph_bridge = RxPyLangGraphBridge(self.stream_router)
+        # Connect bridge to stream router
+        self.stream_router._langgraph_bridge = self.langgraph_bridge
 
         # Initialize route bridge for dynamic type conversion
         self.route_bridge: Optional[RouteBridge] = None
diff --git a/src/cleveragents/langgraph/bridge.py b/src/cleveragents/langgraph/bridge.py
index 1d5f7c2f1b..7946ce014d 100644
--- a/src/cleveragents/langgraph/bridge.py
+++ b/src/cleveragents/langgraph/bridge.py
@@ -8,6 +8,7 @@ from typing import Any
 from typing import Dict
 from typing import List
 from typing import Optional
+import concurrent.futures
 
 import rx
 from rx import operators as ops
@@ -40,6 +41,21 @@ class RxPyLangGraphBridge:
         # Register custom operators
         self._register_langgraph_operators()
 
+    def _run_async_safely(self, coro):
+        """Run an async coroutine safely, handling existing event loops."""
+        try:
+            loop = asyncio.get_event_loop()
+            if loop.is_running():
+                # We're already in an event loop, run in thread pool
+                with concurrent.futures.ThreadPoolExecutor() as executor:
+                    future = executor.submit(asyncio.run, coro)
+                    return future.result()
+            else:
+                return asyncio.run(coro)
+        except RuntimeError:
+            # No event loop exists, create one
+            return asyncio.run(coro)
+
     def _register_langgraph_operators(self) -> None:
         """Register LangGraph-specific operators with the stream router."""
         # Register graph execution operator
@@ -166,7 +182,7 @@ class RxPyLangGraphBridge:
                 },
             )
 
-        return ops.map(lambda msg: asyncio.run(execute_graph(msg)))
+        return ops.map(lambda msg: self._run_async_safely(execute_graph(msg)))
 
     def _create_state_updater(self, params: Dict[str, Any]):
         """Create an operator that updates graph state."""
@@ -263,7 +279,7 @@ class RxPyLangGraphBridge:
                 },
             )
 
-        return ops.map(lambda msg: asyncio.run(execute_node(msg)))
+        return ops.map(lambda msg: self._run_async_safely(execute_node(msg)))
 
     def _create_conditional_router(self, params: Dict[str, Any]):
         """Create an operator that routes based on LangGraph conditions."""
@@ -321,7 +337,7 @@ class RxPyLangGraphBridge:
 
         # Create observer that executes graph
         def on_message(msg: StreamMessage):
-            asyncio.run(graph.execute(msg.content))
+            self._run_async_safely(graph.execute(msg.content))
 
         observer = Observer(on_next=on_message)