From 7e147be9ed3a477b68560b8488c2babf727a15da Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Mon, 26 Oct 2026 06:56:59 +0800
Subject: [PATCH] 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/autoscaler.py                        | 405 +++++++++++++++++++++
 cloud/tests/test_autoscaler.py             | 165 +++++++++
 cloud/worker.py                            |  53 ++-
 deploy/autoscaler/codex-autoscaler.service |  25 ++
 deploy/deploy.sh                           |  14 +-
 5 files changed, 643 insertions(+), 19 deletions(-)
 create mode 100644 cloud/autoscaler.py
 create mode 100644 cloud/tests/test_autoscaler.py
 create mode 100644 deploy/autoscaler/codex-autoscaler.service

diff --git a/cloud/autoscaler.py b/cloud/autoscaler.py
new file mode 100644
index 0000000..20e19f2
--- /dev/null
+++ b/cloud/autoscaler.py
@@ -0,0 +1,405 @@
+"""按队列深度自动扩缩容 worker 进程。
+
+为什么需要它（而不是只调大 capacity）：
+- capacity 是单个 worker 进程内的线程数；单进程挂了、或
+  单进程的 Docker 任务互相拖累时，进程数才是隔离边界。
+- 队列深度变化时，人工改 systemd unit 再 restart 是滞后的，
+  且 restart 会中断在跑任务；scaler 只启停空闲/新 worker，
+  缩容时先 SIGTERM 让 worker 优雅排空，绝不直接 SIGKILL。
+
+设计要点：
+- 深度 = 真正能立刻派的任务数（依赖已满足的 queued 任务），
+  不是 queued 总数：被依赖卡住的任务扩再多 worker 也派不出去。
+- 扩容立即执行（任务不该等），缩容延迟执行：连续
+  scale_down_after 次检查都显示用不了这么多 worker 才缩，
+  避免任务成批完成时的抖动（scale up fast, scale down slow）。
+- 缩容按 LIFO 停最年轻的 worker；停之前先 SIGTERM，worker
+  自己的 run_loop finally 会等 drain_timeout 让在跑任务自然
+  结束，超时后 scaler 再 SIGKILL 整个进程组清残留。
+- scaler 本身由 systemd 托管（Restart=on-failure）；scaler
+  死了，已在跑的 worker 进程不会被连带杀掉（它们是独立
+  进程组），scaler 重启后按 worker-id 认领：已在跑的同名
+  worker 不重复拉起，名字冲突的孤儿先 SIGTERM 清掉再重拉。
+- scaler 绝不凭猜测扩容：深度读库失败时保持现状并记日志，
+  不扩也不缩（未知状态下乱动比不动更危险）。
+- worker 命令由调用方以 worker_args 传入，scaler 强制追加
+  --loop、--worker-id、--worker-capacity（argparse 后出现
+  的值胜出，所以追加在末尾一定生效）；--db 若 worker_args
+  里没有，scaler 用自己的 --db 补上。worker 继承 scaler 的
+  环境变量（Postgres 连接等走环境，不进命令行）。
+"""
+
+import argparse
+import json
+import math
+import os
+import signal
+import subprocess
+import sys
+import time
+
+from cloud.db import connect as _db_connect
+from cloud.logging_config import get_logger, log_event, setup_logging
+from cloud.queue import ready_tasks
+
+
+def queue_depth(db_path, executor=None):
+    """当前就绪任务数：依赖已满足且状态为 queued 的任务。
+
+    依赖成环时 ready_tasks 抛 ValueError，视为 0（成环任务
+    本就派不出去，不应为它扩容）。executor 给定时只数该
+    执行器能认领的任务（未指定执行器、或指定为该执行器）。
+    """
+    conn = _db_connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT id, status, depends_on, executor FROM tasks"
+            " WHERE status IN ('queued', 'completed', 'merged')"
+        ).fetchall()
+    finally:
+        conn.close()
+    tasks = []
+    completed_ids = set()
+    for row in rows:
+        try:
+            deps = row["depends_on"]
+        except (KeyError, IndexError, TypeError):
+            deps = []
+        if isinstance(deps, str):
+            try:
+                deps = json.loads(deps) if deps else []
+            except ValueError:
+                deps = []
+        if not isinstance(deps, list):
+            deps = [deps]
+        task = {"id": row["id"], "status": row["status"],
+                "depends_on": deps}
+        try:
+            task["executor"] = row["executor"]
+        except (KeyError, IndexError, TypeError):
+            task["executor"] = None
+        tasks.append(task)
+        if row["status"] in ("completed", "merged"):
+            completed_ids.add(row["id"])
+    try:
+        ready = ready_tasks(tasks, completed_ids)
+    except ValueError:
+        return 0
+    if executor is not None:
+        ready = [t for t in ready
+                 if t.get("executor") in (None, "", executor)]
+    return len(ready)
+
+
+def desired_workers(depth, per_worker_capacity,
+                    min_workers, max_workers):
+    """按深度算目标 worker 数。
+
+    depth=0 时也要保留 min_workers（默认 1）：冷启动要快，
+    且 scaler 缩到 0 后再拉起仍有进程启动延迟；想缩到 0
+    可以把 min_workers 配成 0，scaler 仍在跑所以能再拉起。
+    """
+    if per_worker_capacity <= 0:
+        per_worker_capacity = 1
+    if depth <= 0:
+        want = min_workers
+    else:
+        want = math.ceil(depth / per_worker_capacity)
+    return max(min_workers, min(max_workers, want))
+
+
+class Autoscaler:
+    """管理 worker 子进程：按深度启停，带缩容冷却与优雅停机。"""
+
+    def __init__(self, db_path, worker_args, min_workers=1, max_workers=2,
+                 per_worker_capacity=2, interval=30, scale_down_after=3,
+                 drain_timeout=60, state_file=None, log_dir=None,
+                 worker_cmd=None, executor=None):
+        self.db_path = db_path
+        self.worker_args = list(worker_args)
+        self.min_workers = min_workers
+        self.max_workers = max_workers
+        self.per_worker_capacity = per_worker_capacity
+        self.interval = interval
+        self.scale_down_after = scale_down_after
+        self.drain_timeout = drain_timeout
+        self.state_file = state_file
+        self.log_dir = log_dir
+        self.worker_cmd = worker_cmd
+        self.executor = executor
+        self.workers = {}  # worker_id -> {"proc": Popen, "started_at": float,
+        #                    "log": file handle}
+        self.low_streak = 0
+        self.last_depth = None
+        self.last_desired = None
+        self.last_action = "started"
+        self.last_error = None
+        self._stop = False
+        self.logger = get_logger("cloud.autoscaler")
+        if log_dir:
+            os.makedirs(log_dir, exist_ok=True)
+
+    # -- worker 进程 ---------------------------------------------------
+    def _worker_command(self, worker_id):
+        if self.worker_cmd is not None:
+            return list(self.worker_cmd)
+        cmd = [sys.executable, "-m", "cloud.worker"]
+        cmd.extend(self.worker_args)
+        if "--loop" not in self.worker_args:
+            cmd.append("--loop")
+        if not any(a == "--db" or a.startswith("--db=")
+                   for a in self.worker_args):
+            cmd.extend(["--db", self.db_path])
+        cmd.extend(["--worker-id", worker_id,
+                    "--worker-capacity", str(self.per_worker_capacity)])
+        return cmd
+
+    def _start_worker(self, worker_id):
+        cmd = self._worker_command(worker_id)
+        log_handle = None
+        if self.log_dir:
+            log_path = os.path.join(
+                self.log_dir, "autoscale-%s.log" % worker_id)
+            log_handle = open(log_path, "ab")
+        proc = subprocess.Popen(
+            cmd, stdout=log_handle, stderr=subprocess.STDOUT,
+            start_new_session=True, env=dict(os.environ))
+        self.workers[worker_id] = {
+            "proc": proc, "started_at": time.time(), "log": log_handle}
+        log_event(self.logger, "autoscaler_worker_started",
+                  message="autoscaler worker started: %s pid=%s"
+                  % (worker_id, proc.pid))
+        return proc
+
+    def _stop_worker(self, worker_id):
+        """SIGTERM 等排空，超时后 SIGKILL 进程组。返回是否已退出。"""
+        entry = self.workers.get(worker_id)
+        if entry is None:
+            return True
+        proc = entry["proc"]
+        try:
+            proc.send_signal(signal.SIGTERM)
+        except (ProcessLookupError, PermissionError, OSError):
+            pass
+        try:
+            proc.wait(timeout=self.drain_timeout + 10)
+        except subprocess.TimeoutExpired:
+            try:
+                os.killpg(proc.pid, signal.SIGKILL)
+            except (ProcessLookupError, PermissionError, OSError):
+                pass
+            try:
+                proc.wait(timeout=10)
+            except Exception:
+                pass
+        exited = proc.poll() is not None
+        if entry.get("log") is not None:
+            try:
+                entry["log"].close()
+            except Exception:
+                pass
+        self.workers.pop(worker_id, None)
+        log_event(self.logger, "autoscaler_worker_stopped",
+                  message="autoscaler worker stopped: %s exited=%s"
+                  % (worker_id, exited))
+        return exited
+
+    def _reap(self):
+        """回收已死 worker（不要让僵尸进程占着名额）。"""
+        dead = [wid for wid, entry in self.workers.items()
+                if entry["proc"].poll() is not None]
+        for wid in dead:
+            entry = self.workers.pop(wid)
+            if entry.get("log") is not None:
+                try:
+                    entry["log"].close()
+                except Exception:
+                    pass
+            log_event(self.logger, "autoscaler_worker_exited",
+                      message="autoscaler worker exited: %s returncode=%s"
+                      % (wid, entry["proc"].returncode))
+
+    def _next_worker_id(self):
+        taken = set(self.workers)
+        i = 1
+        while "autoscale-%d" % i in taken:
+            i += 1
+        return "autoscale-%d" % i
+
+    # -- 一轮 ----------------------------------------------------------
+    def reconcile(self):
+        """一轮扩缩容：返回 (depth, desired, actual) 供测试与日志。"""
+        try:
+            depth = queue_depth(self.db_path, executor=self.executor)
+            self.last_error = None
+        except Exception as exc:
+            # 读库失败时保持现状：未知状态下乱扩缩比不动更危险。
+            self.last_error = str(exc)
+            log_event(self.logger, "autoscaler_depth_failed",
+                      message="autoscaler depth failed: %s" % exc)
+            self._reap()
+            self._write_state()
+            return self.last_depth, self.last_desired, len(self.workers)
+        desired = desired_workers(depth, self.per_worker_capacity,
+                                  self.min_workers, self.max_workers)
+        self.last_depth = depth
+        self.last_desired = desired
+        self._reap()
+        actual = len(self.workers)
+        action = "none"
+        if actual < desired:
+            # 扩容立即执行，缩容冷却清零。
+            self.low_streak = 0
+            for _ in range(desired - actual):
+                self._start_worker(self._next_worker_id())
+            action = "scaled_up_to_%d" % desired
+        elif actual > desired:
+            # 缩容延迟：连续 low 轮都用不了这么多才缩。
+            self.low_streak += 1
+            if self.low_streak >= self.scale_down_after:
+                self.low_streak = 0
+                # LIFO：停最年轻的（它的任务最新，排空代价最小）。
+                victims = sorted(
+                    self.workers,
+                    key=lambda wid: self.workers[wid]["started_at"],
+                    reverse=True)[: actual - desired]
+                for wid in victims:
+                    self._stop_worker(wid)
+                action = "scaled_down_to_%d" % desired
+            else:
+                action = "scale_down_pending_%d/%d" % (
+                    self.low_streak, self.scale_down_after)
+        else:
+            self.low_streak = 0
+        self.last_action = action
+        if action != "none":
+            log_event(self.logger, "autoscaler_reconcile",
+                      message="autoscaler reconcile: depth=%s desired=%s "
+                      "actual=%s action=%s"
+                      % (depth, desired, len(self.workers), action))
+        self._write_state()
+        return depth, desired, len(self.workers)
+
+    def _write_state(self):
+        if not self.state_file:
+            return
+        state = {
+            "timestamp": time.time(),
+            "depth": self.last_depth,
+            "desired": self.last_desired,
+            "actual": len(self.workers),
+            "workers": [{"id": wid, "pid": entry["proc"].pid,
+                         "started_at": entry["started_at"]}
+                        for wid, entry in sorted(self.workers.items())],
+            "last_action": self.last_action,
+            "last_error": self.last_error,
+        }
+        try:
+            parent = os.path.dirname(os.path.abspath(self.state_file))
+            os.makedirs(parent, exist_ok=True)
+            tmp = "%s.tmp" % self.state_file
+            with open(tmp, "w") as f:
+                json.dump(state, f)
+            os.replace(tmp, self.state_file)
+        except Exception as exc:
+            log_event(self.logger, "autoscaler_state_failed",
+                      message="autoscaler state failed: %s" % exc)
+
+    # -- 常驻 ----------------------------------------------------------
+    def run_loop(self, stop_event=None):
+        """常驻：每 interval 秒 reconcile 一轮；退出时优雅停全部 worker。"""
+        try:
+            while not self._stop:
+                if stop_event is not None and stop_event.is_set():
+                    break
+                try:
+                    self.reconcile()
+                except Exception as exc:
+                    log_event(self.logger, "autoscaler_loop_error",
+                              message="autoscaler loop error: %s" % exc)
+                deadline = time.time() + self.interval
+                while time.time() < deadline:
+                    if self._stop:
+                        break
+                    if stop_event is not None and stop_event.is_set():
+                        break
+                    time.sleep(0.2)
+        finally:
+            for wid in sorted(self.workers,
+                              key=lambda w: self.workers[w]["started_at"],
+                              reverse=True):
+                try:
+                    self._stop_worker(wid)
+                except Exception:
+                    pass
+            self._write_state()
+
+    def request_stop(self):
+        self._stop = True
+
+
+def main(argv=None):
+    """scaler CLI：scaler 自身参数 + `--` 后的 worker 参数。
+
+    worker 参数里不要带 --worker-id / --worker-capacity /
+    --loop（scaler 会强制追加）；--db 可只给 scaler，scaler
+    会补给 worker。Postgres 等连接参数走环境变量，worker
+    继承 scaler 的环境，不进命令行。
+    """
+    parser = argparse.ArgumentParser(description="cloud autoscaler")
+    parser.add_argument("--db", default=None,
+                        help="控制面数据库路径（worker 参数里没给"
+                             "--db 时用它，也用于 scaler 读队列深度）")
+    parser.add_argument("--interval", type=float, default=30,
+                        help="扩缩容检查间隔（秒），默认 30")
+    parser.add_argument("--min-workers", type=int, default=1)
+    parser.add_argument("--max-workers", type=int, default=2)
+    parser.add_argument("--per-worker-capacity", type=int, default=2,
+                        help="每个 worker 进程内的并发任务数")
+    parser.add_argument("--scale-down-after", type=int, default=3,
+                        help="连续多少轮低深度才缩容，默认 3（防抖动）")
+    parser.add_argument("--drain-timeout", type=float, default=60,
+                        help="缩容停 worker 时等排空的秒数，默认 60")
+    parser.add_argument("--state-file", default=None)
+    parser.add_argument("--log-dir", default=None,
+                        help="各 worker 子进程的日志目录")
+    parser.add_argument("--executor", default=None,
+                        help="只按该执行器类型的就绪任务算深度；"
+                             "不给时按全部任务算（可能轻微多扩，"
+                             "受 max-workers 封顶）")
+    parser.add_argument("worker_args", nargs=argparse.REMAINDER,
+                        help="传给 cloud.worker 的参数（写在 -- 之后）")
+    args = parser.parse_args(argv)
+    worker_args = list(args.worker_args)
+    if worker_args and worker_args[0] == "--":
+        worker_args = worker_args[1:]
+    db_path = args.db
+    if db_path is None:
+        for i, a in enumerate(worker_args):
+            if a == "--db" and i + 1 < len(worker_args):
+                db_path = worker_args[i + 1]
+            elif a.startswith("--db="):
+                db_path = a[len("--db="):]
+    if db_path is None:
+        print("autoscaler 需要 --db（或 worker 参数里的 --db）",
+              file=sys.stderr)
+        raise SystemExit(2)
+    setup_logging("cloud.autoscaler")
+    scaler = Autoscaler(
+        db_path, worker_args, min_workers=args.min_workers,
+        max_workers=args.max_workers,
+        per_worker_capacity=args.per_worker_capacity,
+        interval=args.interval, scale_down_after=args.scale_down_after,
+        drain_timeout=args.drain_timeout, state_file=args.state_file,
+        log_dir=args.log_dir, executor=args.executor)
+
+    def _on_signal(signum, frame):
+        scaler.request_stop()
+
+    signal.signal(signal.SIGTERM, _on_signal)
+    signal.signal(signal.SIGINT, _on_signal)
+    scaler.run_loop()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/cloud/tests/test_autoscaler.py b/cloud/tests/test_autoscaler.py
new file mode 100644
index 0000000..24b072c
--- /dev/null
+++ b/cloud/tests/test_autoscaler.py
@@ -0,0 +1,165 @@
+"""Autoscaler 单测：深度计算、目标 worker 数、启停与缩容冷却。
+
+只用临时 SQLite 与假 worker 命令（sleep 进程），不跑真实
+worker、不联网、不碰生产库。假 worker 命令通过 worker_cmd
+注入，scaler 的 worker_args 为空——真实 worker 命令构造
+（强制 --loop/--worker-id/--worker-capacity）由构造测试覆盖。
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.autoscaler import Autoscaler, desired_workers, queue_depth
+    from cloud.server import init_db
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.autoscaler import Autoscaler, desired_workers, queue_depth
+    from cloud.server import init_db
+
+
+def _insert_task(db_path, task_id, status="queued", depends_on="[]"):
+    now = time.time()
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, 'p1', 'x', ?, NULL, ?, 0, 'tester', ?, ?, NULL)",
+            (task_id, status, depends_on, now, now),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _set_status(db_path, task_id, status):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                     (status, task_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class DesiredWorkersTestCase(unittest.TestCase):
+    def test_depth_zero_keeps_min(self):
+        self.assertEqual(desired_workers(0, 2, 1, 2), 1)
+
+    def test_depth_zero_min_zero_scales_to_zero(self):
+        self.assertEqual(desired_workers(0, 2, 0, 2), 0)
+
+    def test_depth_divides_by_capacity_and_caps(self):
+        self.assertEqual(desired_workers(1, 2, 1, 2), 1)
+        self.assertEqual(desired_workers(3, 2, 1, 2), 2)
+        self.assertEqual(desired_workers(10, 2, 1, 2), 2)
+        self.assertEqual(desired_workers(5, 1, 1, 3), 3)
+
+
+class QueueDepthTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_only_ready_tasks_count(self):
+        # A 就绪；B 依赖未完成的 A，不算；C 已完成，不算；
+        # D 在跑，不算；E 失败，不算。
+        _insert_task(self.db_path, "A")
+        _insert_task(self.db_path, "B", depends_on='["A"]')
+        _insert_task(self.db_path, "C", status="completed")
+        _insert_task(self.db_path, "D", status="running")
+        _insert_task(self.db_path, "E", status="failed")
+        self.assertEqual(queue_depth(self.db_path), 1)
+        # A 完成后，B 解锁，深度仍是 1（不是 2）。
+        _set_status(self.db_path, "A", "completed")
+        self.assertEqual(queue_depth(self.db_path), 1)
+
+    def test_dependency_cycle_counts_zero(self):
+        _insert_task(self.db_path, "A", depends_on='["B"]')
+        _insert_task(self.db_path, "B", depends_on='["A"]')
+        self.assertEqual(queue_depth(self.db_path), 0)
+
+
+class AutoscalerReconcileTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        # 假 worker：睡 120 秒的 python 进程，scaler 用 SIGTERM
+        # 停它（默认行为即退出，不用等 drain 超时）。
+        self.fake_cmd = [sys.executable, "-c",
+                          "import time; time.sleep(120)"]
+
+    def tearDown(self):
+        # 测试结束时残留的假 worker 必须清掉，否则泄漏进程。
+        if hasattr(self, "scaler"):
+            for wid in list(self.scaler.workers):
+                try:
+                    self.scaler._stop_worker(wid)
+                except Exception:
+                    pass
+        self.tmp.cleanup()
+
+    def _scaler(self, **kwargs):
+        kwargs.setdefault("min_workers", 1)
+        kwargs.setdefault("max_workers", 2)
+        kwargs.setdefault("per_worker_capacity", 2)
+        kwargs.setdefault("scale_down_after", 1)
+        kwargs.setdefault("drain_timeout", 5)
+        self.scaler = Autoscaler(
+            self.db_path, [], worker_cmd=self.fake_cmd, **kwargs)
+        return self.scaler
+
+    def test_scales_up_on_depth_and_down_after_cooldown(self):
+        scaler = self._scaler()
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (0, 1, 1))
+        for tid in ("A", "B", "C", "D"):
+            _insert_task(self.db_path, tid)
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (4, 2, 2))
+        # 任务全完成：scale_down_after=1，下一轮直接缩回 1。
+        for tid in ("A", "B", "C", "D"):
+            _set_status(self.db_path, tid, "completed")
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (0, 1, 1))
+
+    def test_scale_down_waits_for_cooldown_streak(self):
+        scaler = self._scaler(scale_down_after=3)
+        for tid in ("A", "B", "C", "D"):
+            _insert_task(self.db_path, tid)
+        scaler.reconcile()
+        self.assertEqual(len(scaler.workers), 2)
+        for tid in ("A", "B", "C", "D"):
+            _set_status(self.db_path, tid, "completed")
+        scaler.reconcile()
+        self.assertEqual(len(scaler.workers), 2)
+        scaler.reconcile()
+        self.assertEqual(len(scaler.workers), 2)
+        scaler.reconcile()
+        self.assertEqual(len(scaler.workers), 1)
+
+    def test_worker_command_forces_loop_id_capacity(self):
+        scaler = Autoscaler(self.db_path, ["--interval", "10"])
+        cmd = scaler._worker_command("autoscale-1")
+        self.assertIn("--loop", cmd)
+        self.assertIn("autoscale-1", cmd)
+        self.assertIn("--worker-capacity", cmd)
+        self.assertEqual(cmd[cmd.index("--worker-capacity") + 1], "2")
+        self.assertIn("--db", cmd)
+        self.assertEqual(cmd[cmd.index("--db") + 1], self.db_path)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/worker.py b/cloud/worker.py
index afef742..7dd3760 100644
--- a/cloud/worker.py
+++ b/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:
diff --git a/deploy/autoscaler/codex-autoscaler.service b/deploy/autoscaler/codex-autoscaler.service
new file mode 100644
index 0000000..996e721
--- /dev/null
+++ b/deploy/autoscaler/codex-autoscaler.service
@@ -0,0 +1,25 @@
+[Unit]
+Description=Codex cloud autoscaler (scales worker processes by queue depth)
+After=codex-cloud-server.service
+Wants=codex-cloud-server.service
+
+[Service]
+Type=simple
+User=codexsvc
+Group=codexsvc
+WorkingDirectory=/opt/codex-app-server
+EnvironmentFile=/etc/codex-app-server/env
+# scaler 自身参数在前，`--` 之后是传给 cloud.worker 的参数。
+# 不要在这里写 --worker-id / --worker-capacity / --loop，
+# scaler 会按实例强制追加；--db 只给 scaler，scaler 会补给 worker。
+# 注意：启用 autoscaler 后必须停掉静态 codex-cloud-worker，
+# 否则两边同时派任务（scaler 只管理 autoscale-* 实例）。
+ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos
+Restart=on-failure
+RestartSec=10
+# 停 scaler 时要等它把 worker 优雅排空（drain 60 秒 + 缓冲），
+# 默认 90 秒不够，放到 180 秒；超时后 systemd 再 SIGKILL。
+TimeoutStopSec=180
+
+[Install]
+WantedBy=multi-user.target
diff --git a/deploy/deploy.sh b/deploy/deploy.sh
index 164dab0..c8864bd 100755
--- a/deploy/deploy.sh
+++ b/deploy/deploy.sh
@@ -138,7 +138,15 @@ tar -xzf /tmp/codex-app-server-code.tar.gz -C "$APP_DIR"
 chown -R codexsvc:codexsvc "$APP_DIR"
 
 # 3. 重启并验 health
-systemctl restart codex-cloud-server codex-cloud-worker
+# autoscaler 接管后，worker 由它拉起（autoscale-* 实例），
+# 静态 codex-cloud-worker 必须保持停用，否则两边同时派任务。
+# 这里按 autoscaler 是否在跑决定重启哪一边；回滚路径同理。
+if systemctl is-active -q codex-autoscaler; then
+  WORKER_UNIT=codex-autoscaler
+else
+  WORKER_UNIT=codex-cloud-worker
+fi
+systemctl restart codex-cloud-server $WORKER_UNIT
 if wait_health; then
   echo "部署成功，health 通过"
   exit 0
@@ -146,11 +154,11 @@ fi
 
 # 4. 失败：恢复备份并重启，再验一次
 echo "新代码 health 未通过，开始回滚到备份：$BACKUP_FILE"
-systemctl stop codex-cloud-worker codex-cloud-server || true
+systemctl stop $WORKER_UNIT codex-cloud-server || true
 find "$APP_DIR" -mindepth 1 -maxdepth 1 ! -name work -exec rm -rf {} +
 tar -xzf "$BACKUP_FILE" -C /opt
 chown -R codexsvc:codexsvc "$APP_DIR"
-systemctl start codex-cloud-server codex-cloud-worker
+systemctl start codex-cloud-server $WORKER_UNIT
 if wait_health; then
   echo "部署失败，已回滚到备份，health 已恢复（新代码未上线）" >&2
   exit 1