fix: raise RuntimeError when AsyncPipeline.run() is called from within an async context (#9712)

* Raise RuntimeError when AsyncPipeline.run() is called from an async context

* Add release note

Author: mpangrazzi

haystack/core/pipeline/async_pipeline.py

@@ -676,7 +676,20 @@ def run(
             Or if a Component fails or returns output in an unsupported type.
         :raises PipelineMaxComponentRuns:
             If a Component reaches the maximum number of times it can be run in this Pipeline.
+        :raises RuntimeError:
+            If called from within an async context. Use `run_async` instead.
         """
-        return asyncio.run(
-            self.run_async(data=data, include_outputs_from=include_outputs_from, concurrency_limit=concurrency_limit)
-        )
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            # No running loop: safe to use asyncio.run()
+            return asyncio.run(
+                self.run_async(
+                    data=data, include_outputs_from=include_outputs_from, concurrency_limit=concurrency_limit
+                )
+            )
+        else:
+            # Running loop present: do not create the coroutine and do not call asyncio.run()
+            raise RuntimeError(
+                "Cannot call run() from within an async context. Use 'await pipeline.run_async(...)' instead."
+            )

releasenotes/notes/fix-asyncpipeline-run-error-e82ede5c4f37f485.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Raise a `RuntimeError` when `AsyncPipeline.run` is called from within an async context, indicating that `run_async` should be used instead.

test/core/pipeline/test_async_pipeline.py

@@ -4,6 +4,8 @@
 
 import asyncio
 
+import pytest
+
 from haystack import AsyncPipeline
 
 
@@ -22,3 +24,22 @@ async def run_all():
     component_spans = [sp for sp in spying_tracer.spans if sp.operation_name == "haystack.component.run_async"]
     for span in component_spans:
         assert span.tags["haystack.component.visits"] == 1
+
+
+def test_run_in_sync_context(waiting_component):
+    pp = AsyncPipeline()
+    pp.add_component("wait", waiting_component())
+
+    result = pp.run({"wait_for": 0.001})
+
+    assert result == {"wait": {"waited_for": 0.001}}
+
+
+def test_run_in_async_context_raises_runtime_error():
+    pp = AsyncPipeline()
+
+    async def call_run():
+        pp.run({})
+
+    with pytest.raises(RuntimeError, match="Cannot call run\\(\\) from within an async context"):
+        asyncio.run(call_run())