fix: keep blocking docker and subprocess calls off hypervisor-agent and agent-spawner event loops

agents/hypervisor-agent/hypervisor_agent.py

@@ -435,6 +435,41 @@ def _check_cooldown(self, key: str, seconds: int) -> bool:
         return False
 
     # ── Auto-scaling (guarded) ───────────────────────────────────────
+    def _auto_scale_sync(self, tag: str) -> List[str]:
+        """Blocking sweep + stop/remove. Call via asyncio.to_thread."""
+        actions: List[str] = []
+        containers = self.docker_client.containers.list(all=True)
+
+        # 1. Remove zombie containers (excessive restarts, not currently running)
+        for c in containers:
+            restart_count = c.attrs.get('RestartCount', 0)
+            if restart_count > 10 and c.status != "running":
+                msg = f"{tag}remove zombie {c.name} (restart_count={restart_count})"
+                if ENFORCE_SCALING:
+                    try:
+                        c.remove(force=True)
+                    except Exception as e:
+                        logger.error(f"Failed to remove {c.name}: {e}")
+                        continue
+                actions.append(msg)
+                logger.info(f"🗑️ {msg}")
+
+        # 2. Stop non-critical monitoring containers to free RAM
+        for c in containers:
+            if c.status == "running" and any(nc in c.name.lower() for nc in NON_CRITICAL):
+                msg = f"{tag}stop non-critical {c.name} (memory pressure)"
+                if ENFORCE_SCALING:
+                    try:
+                        c.stop(timeout=15)
+                        self.stopped_by_agent.add(c.name)
+                    except Exception as e:
+                        logger.error(f"Failed to stop {c.name}: {e}")
+                        continue
+                actions.append(msg)
+                logger.info(f"⏸️ {msg}")
+
+        return actions
+
     async def auto_scale_containers(self, metrics: SystemMetrics) -> List[str]:
         """
         Shed load under sustained memory pressure. DRY-RUN unless ENFORCE_SCALING.
@@ -447,40 +482,29 @@ async def auto_scale_containers(self, metrics: SystemMetrics) -> List[str]:
         async with self.scale_lock:
             tag = "" if ENFORCE_SCALING else "[DRY-RUN] "
             try:
-                containers = self.docker_client.containers.list(all=True)
+                # docker-py is sync. list() + stop(timeout=15) per container is
+                # seconds of work; on the loop it stalls everything this process
+                # serves. Same reason _collect_container_metrics is threaded.
+                actions = await asyncio.to_thread(self._auto_scale_sync, tag)
             except Exception as e:
                 logger.error(f"Docker unreachable during auto-scale: {e}")
                 self._reset_docker()
-                return actions
-
-            # 1. Remove zombie containers (excessive restarts, not currently running)
-            for c in containers:
-                restart_count = c.attrs.get('RestartCount', 0)
-                if restart_count > 10 and c.status != "running":
-                    msg = f"{tag}remove zombie {c.name} (restart_count={restart_count})"
-                    if ENFORCE_SCALING:
-                        try:
-                            c.remove(force=True)
-                        except Exception as e:
-                            logger.error(f"Failed to remove {c.name}: {e}")
-                            continue
-                    actions.append(msg)
-                    logger.info(f"🗑️ {msg}")
-
-            # 2. Stop non-critical monitoring containers to free RAM
-            for c in containers:
-                if c.status == "running" and any(nc in c.name.lower() for nc in NON_CRITICAL):
-                    msg = f"{tag}stop non-critical {c.name} (memory pressure)"
-                    if ENFORCE_SCALING:
-                        try:
-                            c.stop(timeout=15)
-                            self.stopped_by_agent.add(c.name)
-                        except Exception as e:
-                            logger.error(f"Failed to stop {c.name}: {e}")
-                            continue
-                    actions.append(msg)
-                    logger.info(f"⏸️ {msg}")
+                return []
+
+        return actions
 
+    def _heal_back_sync(self, ram_percent: float) -> List[str]:
+        """Blocking get + start. Call via asyncio.to_thread."""
+        actions: List[str] = []
+        for name in list(self.stopped_by_agent):
+            try:
+                c = self.docker_client.containers.get(name)
+                c.start()
+                self.stopped_by_agent.discard(name)
+                actions.append(f"restarted {name} (pressure cleared)")
+                logger.info(f"♻️ Restarted {name} (RAM back to {ram_percent:.1f}%)")
+            except Exception as e:
+                logger.error(f"Failed to restart {name}: {e}")
         return actions
 
     async def heal_back(self, metrics: SystemMetrics) -> List[str]:
@@ -490,15 +514,7 @@ async def heal_back(self, metrics: SystemMetrics) -> List[str]:
             return actions
 
         async with self.scale_lock:
-            for name in list(self.stopped_by_agent):
-                try:
-                    c = self.docker_client.containers.get(name)
-                    c.start()
-                    self.stopped_by_agent.discard(name)
-                    actions.append(f"restarted {name} (pressure cleared)")
-                    logger.info(f"♻️ Restarted {name} (RAM back to {metrics.ram_percent:.1f}%)")
-                except Exception as e:
-                    logger.error(f"Failed to restart {name}: {e}")
+            actions = await asyncio.to_thread(self._heal_back_sync, metrics.ram_percent)
         return actions
 
     # ── Broadcast ────────────────────────────────────────────────────

services/agent-spawner/spawner.py

@@ -69,7 +69,11 @@ async def spawn_agent(agent_name: str) -> bool:
         logger.info(f"[{agent_name}] Executing: {' '.join(cmd)}")
 
         try:
-            result = subprocess.run(
+            # subprocess.run blocks for up to 30s. This process also holds a
+            # pub/sub connection whose read carries a 5s socket_timeout, so a
+            # blocked loop kills spawn_listener as well as stalling everything.
+            result = await asyncio.to_thread(
+                subprocess.run,
                 cmd,
                 cwd=COMPOSE_CWD,
                 capture_output=True,
@@ -114,9 +118,11 @@ async def shutdown_agent(agent_name: str) -> bool:
         # Try multiple naming conventions
         for name_variant in [agent_name, container_name]:
             try:
-                container = docker_client.containers.get(name_variant)
+                # docker-py is sync; stop(timeout=10) can hold the loop for 10s
+                # while spawn_listener's pub/sub read is waiting on a 5s timeout.
+                container = await asyncio.to_thread(docker_client.containers.get, name_variant)
                 if container.status == "running":
-                    container.stop(timeout=10)
+                    await asyncio.to_thread(container.stop, timeout=10)
                     logger.info(f"[{agent_name}] Shut down successfully")
                     agent_activity.pop(agent_name, None)
                     return True
@@ -184,7 +190,9 @@ async def spawn_listener(r: redis.Redis):
                     try:
                         for name_variant in [agent_name, f"{DOCKER_COMPOSE_PROJECT}_{agent_name}_1"]:
                             try:
-                                container = docker_client.containers.get(name_variant)
+                                container = await asyncio.to_thread(
+                                    docker_client.containers.get, name_variant
+                                )
                                 if container.status == "running":
                                     logger.info(f"[{agent_name}] Already running, updating activity")
                                     agent_activity[agent_name] = time.time()