worker: always start capacity pool threads in run_all

run_all previously sized its pool from the ready snapshot at round start (min(capacity, ready_count)). If a round started with only 1 ready task (e.g. a just-claimed PM), only 1 thread ran; that thread entered a marker wait (up to 3600s) after its model turn, and children created/retried mid-round had no free thread to claim them. run_loop is blocked joining that round, so no new round starts; if the other worker had already scaled down, the queued child waited out the full wait timeout unclaimed. Fix: always start capacity threads. Idle threads exit immediately when no task is ready and nothing is active, and poll every 0.2s while something is active, so mid-round arrivals are claimed. Test: run_all with capacity=2, 1 initial task whose mocked run_one creates a child then polls for it; without the fix the PM poll times out in 5s (1 result), with the fix both tasks dispatch in the same round (0.2s).

cloud/worker.py

@@ -4602,15 +4602,18 @@ class Worker:
                     with results_lock:
                         results.append(result)
 
-        try:
-            ready_count = len(self._ready_tasks())
-        except Exception:
-            # 深度快照失败时退回开满线程，池线程自己的认领
-            # 失败容错会处理，不让一次读库失败炸掉整轮。
-            ready_count = self.capacity
+        # 池线程数恒为 capacity，绝不按本轮开始时的就绪快照缩减。
+        # 快照缩减是 PM 等待死锁的根因：本轮开始时若只有 1 个就绪
+        # 任务（如刚认领的 PM），只开 1 个线程；该线程跑 PM 模型回合
+        # 后进入标记等待（最长 3600 秒），期间 PM 新建/重试的子任务
+        # 入队，却再无空闲线程认领它——run_loop 又被本轮 run_all 的
+        # join 挡住、开不了新一轮；若此时另一 worker 已缩容，就只剩
+        # 一个被等待占住的 worker，子任务等满等待超时也无人跑。
+        # 恒开 capacity 个线程没有空转代价：空闲线程在 task 为 None
+        # 且 active 为 0 时立刻退出，有在跑任务时才 0.2 秒轮询认领，
+        # 正好接住轮中新建的任务。
         threads = [threading.Thread(target=_pool_thread, daemon=True)
-                   for _ in range(min(self.capacity,
-                                      max(1, ready_count)))]
+                   for _ in range(self.capacity)]
         for t in threads:
             t.start()
         for t in threads: