fix: Use ThreadPoolExecutor to avoid event loop conflicts in HybridBackgroundTasks

- Wrap Docket async operations in ThreadPoolExecutor to run in separate thread
- This should resolve 'Event loop is closed' errors in CI
- Tests may need updating to reflect that Docket operations run in background

Author: abrookins

agent_memory_server/dependencies.py

@@ -1,4 +1,4 @@
-import asyncio
+import concurrent.futures
 from collections.abc import Callable
 from typing import Any
 
@@ -20,38 +20,40 @@ def add_task(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
 
         if settings.use_docket:
             logger.info("Scheduling task through Docket")
-            # Schedule task directly in Docket's Redis queue
-            self._schedule_docket_task(func, *args, **kwargs)
+
+            # Create a wrapper that will handle Docket scheduling in a thread
+            def docket_wrapper():
+                """Wrapper function that schedules the task through Docket"""
+
+                def run_in_thread():
+                    """Run the async Docket operations in a separate thread"""
+                    import asyncio
+
+                    from docket import Docket
+
+                    async def schedule_task():
+                        async with Docket(
+                            name=settings.docket_name,
+                            url=settings.redis_url,
+                        ) as docket:
+                            # Schedule task in Docket's queue
+                            await docket.add(func)(*args, **kwargs)
+
+                    # Run in a new event loop in this thread
+                    asyncio.run(schedule_task())
+
+                # Execute in a thread pool to avoid event loop conflicts
+                with concurrent.futures.ThreadPoolExecutor() as executor:
+                    future = executor.submit(run_in_thread)
+                    future.result()  # Wait for completion
+
+            # Add the wrapper to FastAPI background tasks
+            super().add_task(docket_wrapper)
         else:
             logger.info("Using FastAPI background tasks")
             # Use FastAPI's background tasks directly
             super().add_task(func, *args, **kwargs)
 
-    def _schedule_docket_task(
-        self, func: Callable[..., Any], *args: Any, **kwargs: Any
-    ) -> None:
-        """Schedule a task in Docket's Redis queue"""
-
-        async def schedule_task():
-            from docket import Docket
-
-            async with Docket(
-                name=settings.docket_name,
-                url=settings.redis_url,
-            ) as docket:
-                # Schedule task in Docket's queue
-                await docket.add(func)(*args, **kwargs)
-
-        # Run the async scheduling operation
-        try:
-            # Try to get the current event loop
-            loop = asyncio.get_running_loop()
-            # If we're in an async context, create a task
-            loop.create_task(schedule_task())
-        except RuntimeError:
-            # If no event loop is running, run it synchronously
-            asyncio.run(schedule_task())
-
 
 # Backwards compatibility alias
 DocketBackgroundTasks = HybridBackgroundTasks