Autoscale workers by queue depth (adaptive threads + process scaler)

Layer 1, in-worker: run_all pool thread count is now adaptive
per round (min(capacity, ready_count)) instead of always
capacity threads; ready_count counts only dependency-satisfied
queued tasks via a shared _ready_tasks() helper (pick_next
refactored onto it), so blocked tasks never inflate the pool.

Layer 2, process scaler (cloud/autoscaler.py): a systemd-managed
daemon that starts/stops cloud.worker subprocesses by ready
queue depth. desired = ceil(depth / per_worker_capacity)
clamped to [min_workers, max_workers]; scale-up is immediate,
scale-down waits scale_down_after consecutive low checks
(default 3) to avoid flapping on bursty completions. Scale-down
is LIFO (youngest worker first), SIGTERM for graceful drain
(worker run_loop drains up to drain_timeout), SIGKILL of the
process group only after drain timeout. Depth read failure
keeps the current worker count (never scale on unknown state).
Worker command forces --loop/--worker-id/--worker-capacity
(appended last so argparse last-wins); --db comes from scaler
or worker args; workers inherit scaler env (Postgres etc.).
State file (depth/desired/actual/workers) written atomically
for observability. deploy.sh now restarts codex-autoscaler
instead of the static worker unit when the scaler is active,
so the two never poll the same queue at once.
Tests: 8 autoscaler tests (desired math, ready-only depth,
cycle=0, scale up/down, cooldown streak, forced worker flags).

cloud/worker.py

@@ -603,6 +603,28 @@ class Worker:
         return budget_exceeded(self.usage_turns, self.max_turns,
                                self.usage_cost_usd, self.max_cost_usd)
 
+    def _ready_tasks(self):
+        """当前就绪任务列表（pick_next 的批量版）。
+
+        逻辑与 pick_next 完全一致（预算门、依赖满足、执行器
+        过滤），只是返回全部而不是第一个。自动扩缩容用它算
+        队列深度：只数真正能立刻派的任务，被依赖卡住的
+        queued 任务不算，否则 scaler 会为永远派不出去的任务
+        扩容。依赖成环时返回 []（与 pick_next 的 None 一致）。
+        """
+        if self._budget_exceeded():
+            return []
+        tasks = self.load_tasks()
+        completed_ids = {t["id"] for t in tasks
+                         if t["status"] in COMPLETED_STATUSES}
+        try:
+            ready = ready_tasks(tasks, completed_ids)
+        except ValueError:
+            return []
+        mine = self._executor_name()
+        return [t for t in ready
+                if t.get("executor") in (None, "", mine)]
+
     def pick_next(self):
         """下一个可派任务，或 None（无就绪任务 / 预算超限 / 依赖成环）。
 
@@ -610,21 +632,7 @@ class Worker:
         依赖成环时 ready_tasks 抛 ValueError，worker 视为无就绪任务
         （成环任务本就永远派不出去，run_all 不应因此崩溃）。
         """
-        if self._budget_exceeded():
-            return None
-        tasks = self.load_tasks()
-        completed_ids = {t["id"] for t in tasks
-                         if t["status"] in COMPLETED_STATUSES}
-        try:
-            ready = ready_tasks(tasks, completed_ids)
-        except ValueError:
-            return None
-        # 多 worker 池过滤：只取未指定执行器、或指定为本 worker
-        # 执行器类型的任务；_claim 里还有一道 WHERE 过滤兜底，
-        # pick_next 这里是第一道，避免 worker 老去抢别人的任务。
-        mine = self._executor_name()
-        ready = [t for t in ready
-                 if t.get("executor") in (None, "", mine)]
+        ready = self._ready_tasks()
         return ready[0] if ready else None
 
     # -- 并发运行状态 --------------------------------------------------
@@ -1341,6 +1349,12 @@ class Worker:
         # concurrent.futures 内部依赖标准库 queue.SimpleQueue，
         # 在该单测之后导入它会直接 AttributeError；手动用
         # threading.Condition + 共享计数实现线程池，避开这个坑。
+        # 池线程数按队列深度自适应：每轮开始先数就绪任务，
+        # 线程数 = min(capacity, 就绪数)——队列只有 1 个任务时
+        # 只开 1 个线程，不为空转线程池买单；队列深时开满
+        # capacity。就绪数是快照，任务完成解锁依赖时，池线程
+        # 靠“无就绪但仍有在跑任务则等待重试”继续接新任务，
+        # 不会因为初始快照小就漏派。
         # 池线程数 = capacity；每个池线程循环认领-执行，认领
         # 本身是 CAS，两个线程盯上同一任务时只有一个能拿到。
         # 依赖解锁：某线程暂无就绪任务时，若仍有线程在跑就等
@@ -1398,8 +1412,15 @@ class Worker:
                     with results_lock:
                         results.append(result)
 
+        try:
+            ready_count = len(self._ready_tasks())
+        except Exception:
+            # 深度快照失败时退回开满线程，池线程自己的认领
+            # 失败容错会处理，不让一次读库失败炸掉整轮。
+            ready_count = self.capacity
         threads = [threading.Thread(target=_pool_thread, daemon=True)
-                   for _ in range(self.capacity)]
+                   for _ in range(min(self.capacity,
+                                      max(1, ready_count)))]
         for t in threads:
             t.start()
         for t in threads: