fix(avatar): prevent takeover replies blocking across chats

digital-avatar-app/backend/services/takeover_service.py

@@ -86,24 +86,34 @@ class TakeoverService:
         self.reply_delay_seconds = reply_delay_seconds
         self.now = now
         self._sessions: dict[str, dict] = {}
-        self._run_lock = asyncio.Lock()
+        self._poll_lock = asyncio.Lock()
+        self._process_lock = asyncio.Lock()
 
     async def poll_and_process_messages(self):
-        """Run one complete cycle; polling always happens before reply dispatch."""
-        if self._run_lock.locked():
+        """Run one complete cycle for callers that do not use the split scheduler."""
+        await self.poll_messages()
+        await self.process_reply_tasks()
+
+    async def poll_messages(self):
+        """Fetch BOXIM events without blocking reply generation and dispatch."""
+        if self._poll_lock.locked():
             return
-        async with self._run_lock:
+        async with self._poll_lock:
             self._recover_stuck_tasks()
             avatar_ids = self._enabled_avatar_ids()
             self._cancel_disabled_tasks(set(avatar_ids))
             for avatar_id in avatar_ids:
                 await self._sync_avatar(avatar_id)
 
-            generated = await self._prepare_replies()
-            if generated:
-                # Catch a human reply sent while the model was preparing its answer.
-                for avatar_id in avatar_ids:
-                    await self._sync_avatar(avatar_id)
+    async def process_reply_tasks(self):
+        """Generate and send replies independently from BOXIM's long poll."""
+        if self._process_lock.locked():
+            return
+        async with self._process_lock:
+            self._recover_stuck_tasks()
+            avatar_ids = set(self._enabled_avatar_ids())
+            self._cancel_disabled_tasks(avatar_ids)
+            await self._prepare_replies()
             await self._dispatch_ready_replies()
 
     def _enabled_avatar_ids(self) -> list[str]:
@@ -454,11 +464,19 @@ class TakeoverService:
         finally:
             db.close()
 
-        generated = 0
-        for task_id in task_ids:
-            if await asyncio.to_thread(self._generate_reply, task_id):
-                generated += 1
-        return generated
+        if not task_ids:
+            return 0
+
+        # Each conversation owns its task, so unrelated contacts can generate in
+        # parallel instead of one slow model response delaying every other peer.
+        semaphore = asyncio.Semaphore(4)
+
+        async def generate(task_id: str) -> bool:
+            async with semaphore:
+                return await asyncio.to_thread(self._generate_reply, task_id)
+
+        results = await asyncio.gather(*(generate(task_id) for task_id in task_ids))
+        return sum(bool(result) for result in results)
 
     def _generate_reply(self, task_id: str) -> bool:
         db = self.session_factory()
@@ -548,8 +566,8 @@ class TakeoverService:
         finally:
             db.close()
 
-        for task_id in task_ids:
-            await self._send_task(task_id)
+        if task_ids:
+            await asyncio.gather(*(self._send_task(task_id) for task_id in task_ids))
 
     async def _send_task(self, task_id: str) -> bool:
         db = self.session_factory()