fix: threading approach in run_async_in_sync didn't wait for all tasks to complete (#487)

## Description
Fix the error when running tasks in threads

## PR Type

<!-- Delete the types that don't apply --!>

🐛 Bug Fix

## Relevant issues

<!-- e.g. "Fixes #123" -->

## Checklist
- [x] I have added unit tests that prove my fix/feature works
- [x] New and existing tests pass locally
- [x] Documentation was updated where necessary
- [x] I have read and followed the [contribution
guidelines](https://github.com/mozilla-ai/any-llm/blob/main/CONTRIBUTING.md)```

Author: njbrake

src/any_llm/utils/aio.py

@@ -42,7 +42,32 @@ def run_async_in_sync(coro: Coroutine[Any, Any, T], allow_running_loop: bool = T
         # If we get here, there's a loop running, so we can't use run_until_complete()
         # or asyncio.run() - must use threading approach
         def run_in_thread() -> T:
-            return asyncio.run(coro)
+            async def run_with_cleanup() -> T:
+                try:
+                    result = await coro
+
+                    # Wait for any pending background tasks to complete to prevent "Event loop is closed" errors
+                    pending_tasks = [
+                        task for task in asyncio.all_tasks() if not task.done() and task is not asyncio.current_task()
+                    ]
+
+                    if pending_tasks:
+                        await asyncio.gather(*pending_tasks, return_exceptions=True)
+                except Exception:
+                    pending_tasks = [
+                        task for task in asyncio.all_tasks() if not task.done() and task is not asyncio.current_task()
+                    ]
+
+                    for task in pending_tasks:
+                        task.cancel()
+
+                    if pending_tasks:
+                        await asyncio.gather(*pending_tasks, return_exceptions=True)
+
+                    raise
+                return result
+
+            return asyncio.run(run_with_cleanup())
 
         with concurrent.futures.ThreadPoolExecutor() as executor:
             return executor.submit(run_in_thread).result()

tests/unit/test_utils.py

@@ -0,0 +1,28 @@
+import asyncio
+
+from any_llm.utils.aio import run_async_in_sync
+
+
+def test_run_async_in_sync_fails_with_background_task_state() -> None:
+    task_completed = {"value": False}
+
+    async def operation_with_critical_background_task() -> str:
+        """Simulates an operation where a background task MUST complete for success."""
+
+        async def critical_background_work() -> None:
+            await asyncio.sleep(0.02)
+            task_completed["value"] = True
+
+        task = asyncio.create_task(critical_background_work())
+        assert task is not None
+        return "operation_started"
+
+    async def test_in_streamlit_context() -> None:
+        task_completed["value"] = False
+        # This triggers the threading in  run_async_in_sync
+        result = run_async_in_sync(operation_with_critical_background_task())
+        assert result == "operation_started"
+        await asyncio.sleep(0.05)
+        assert task_completed["value"] is True
+
+    asyncio.run(test_in_streamlit_context())