feat: P1 补齐（自动评审队列、自动化+分诊、并发 caps、运行时审批队列）

cloud/worker.py

@@ -147,6 +147,7 @@ from cloud.repos import RepoError, checkout_dir, resolve_repo
 from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
 from cloud.uploads import upload_root as _upload_root
 from cloud.sandbox import _default_runner as _sandbox_default_runner
+from cloud.caps import effective_capacity, resolve_capacity
 from cloud.workers import heartbeat_worker, register_worker
 
 # 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。
@@ -282,7 +283,7 @@ class Worker:
                  task_timeout=1800, max_retries=1,
                  retry_backoff_seconds=5.0, repos_dir="./repos",
                  repos_runner=None, heartbeat_file=None,
-                 worker_id=None, capacity=1, fallback_models=None,
+                 worker_id=None, capacity=None, fallback_models=None,
                  tenant_isolation=False, tenants_root=None,
                  tenant_template_home=None, tenant_template_repo=None,
                  egress_proxy_port=0, claim_executor_only=False,
@@ -412,10 +413,11 @@ class Worker:
         # 自然变 stale，不要用随机 uuid 让注册表无法对应同一台机器。
         self.worker_id = worker_id or "%s-%d" % (
             socket.gethostname(), os.getpid())
-        try:
-            self.capacity = int(capacity)
-        except (TypeError, ValueError):
-            self.capacity = 1
+        # 并发上限配置来源（见 cloud/caps.py）：显式构造参数 /
+        # CLI --worker-capacity > 环境变量 CLOUD_WORKER_CAPACITY >
+        # 默认 1；非法值归一为 1。self._env 在上面已定好，单测
+        # 注入的普通 dict 同样可读环境变量（无该键时默认 1）。
+        self.capacity = resolve_capacity(capacity, env=self._env)
         # 常驻启动时的代码签名：run_loop 每轮边界比对，变了就自退
         # （见 run_loop 的代码变化自退检查）。
         self._code_signature = _code_signature()
@@ -436,7 +438,8 @@ class Worker:
         try:
             register_worker(
                 self.db_path, self.worker_id,
-                executor=self._executor_name(), capacity=self.capacity)
+                executor=self._executor_name(), capacity=self.capacity,
+                running=self._running_count())
         except Exception as exc:
             log_event(self.logger, "worker_register_failed",
                       message="worker register failed: %s" % exc)
@@ -446,7 +449,8 @@ class Worker:
         try:
             heartbeat_worker(
                 self.db_path, self.worker_id,
-                executor=self._executor_name(), capacity=self.capacity)
+                executor=self._executor_name(), capacity=self.capacity,
+                running=self._running_count())
         except Exception as exc:
             log_event(self.logger, "worker_registry_heartbeat_failed",
                       message="worker registry heartbeat failed: %s" % exc)
@@ -531,7 +535,23 @@ class Worker:
         claim_executor_only 开启时（宿主级沙箱 worker 用）：WHERE
         只加 executor = ?，NULL 任务绝不认领（否则宿主 worker
         会抢走本该进 Docker 的任务）。
+
+        并发上限门（caps）：CAS 之前先在 _running_lock 下占位。
+        running 数达到 self.capacity 时直接返回 False，任务保持
+        queued，不碰数据库；占位成功才走 CAS。同一任务已在
+        _running 里时不重复计数（重复占位不增槽）。CAS 失败
+        （已被别的 worker 抢走）时，只清掉本次新建的占位，绝
+        不动已在跑任务的条目。占位条目在执行路径的 finally
+        里清（本地 _execute_via_process / Docker 同步路径），
+        认领后提前返回的三道门（缺 key / 仓库 / 配额）在
+        run_one 里显式清占位；CAS 语义本身一句未改。
         """
+        with self._running_lock:
+            already_running = task_id in self._running
+            if not already_running and \
+                    len(self._running) >= effective_capacity(self.capacity):
+                return False
+            self._running.setdefault(task_id, {})
         now = time.time()
         conn = _db_connect(self.db_path)
         try:
@@ -563,7 +583,10 @@ class Worker:
                     (now, now, self.host_name, task_id, executor),
                 )
             conn.commit()
-            return cur.rowcount == 1
+            claimed = cur.rowcount == 1
+            if not claimed and not already_running:
+                self._running_clear(task_id)
+            return claimed
         finally:
             conn.close()
 
@@ -908,6 +931,16 @@ class Worker:
             return {tid: dict(entry)
                     for tid, entry in self._running.items()}
 
+    def _running_count(self):
+        """当前占用槽位数（_running_lock 下计数）。
+
+        启动中的任务已在 _claim 里占位（空条目），所以这里数
+        到的既包括已起进程 / 容器的任务，也包括已认领未起
+        进程的任务；注册表心跳与认领门共用这个口径。
+        """
+        with self._running_lock:
+            return len(self._running)
+
     def _wait_for_running(self, timeout=None):
         """等 _running 变空；timeout 为 None 时无限等。
 
@@ -2082,6 +2115,9 @@ class Worker:
                       project_id=task["project_id"], status="failed")
             self._notify_task_finished(
                 task_id, task["project_id"], "failed", error=error)
+            # 认领占位在此释放：缺 key 门在执行器起进程之前返回，
+            # 不清占位会把槽位漏到 worker 退出（见 _claim 文档）。
+            self._running_clear(task_id)
             return {"dispatched": True, "task_id": task_id,
                     "host_name": self.executor._host_name(task)
                     if hasattr(self.executor, "_host_name") else task_id,
@@ -2106,6 +2142,8 @@ class Worker:
                       project_id=task["project_id"], status="failed")
             self._notify_task_finished(
                 task_id, task["project_id"], "failed", error=repo_error)
+            # 同缺 key 门：仓库解析失败在起进程之前返回，先清占位。
+            self._running_clear(task_id)
             return {"dispatched": True, "task_id": task_id,
                     "host_name": self.executor._host_name(task)
                     if hasattr(self.executor, "_host_name") else task_id,
@@ -2129,6 +2167,8 @@ class Worker:
             log_event(self.logger, "worker_quota_stop", task_id=task_id,
                       project_id=task["project_id"],
                       status="interrupted")
+            # 同缺 key 门：配额门在起进程之前返回，先清占位。
+            self._running_clear(task_id)
             return {"dispatched": True, "task_id": task_id,
                     "host_name": self.executor._host_name(task)
                     if hasattr(self.executor, "_host_name") else task_id,
@@ -2736,6 +2776,11 @@ class Worker:
                 if stop_event is not None and stop_event.is_set():
                     break
                 self._heartbeat_running_tasks()
+                # 注册表心跳也每 30 秒刷一次：长任务跑满一轮时
+                # run_all 久久不返回，只靠轮次结束的心跳会让
+                # /workers 的 running 数长期停在旧值；这里刷的
+                # running 含启动中占位，与认领门口径一致。
+                self._registry_heartbeat()
                 hb_stop.wait(30)
 
         hb_thread = threading.Thread(target=_hb_loop, daemon=True)
@@ -2795,6 +2840,14 @@ class Worker:
                     time.sleep(0.2)
         finally:
             hb_stop.set()
+            # 等心跳线程彻底退出再返回：否则它手头一次未写完的
+            # 注册表心跳可能在调用方清理 DB 目录时重建 control.db，
+            # 导致 tearDown rmtree 报 Directory not empty（全量套件
+            # 里 test_code_changed_exits_loop_before_claiming 偶发
+            # error 即此竞态）。daemon 线程 + set() 会唤醒 wait，
+            # 正常情况下 join 瞬间返回；5 秒超时兜底，避免写挂时
+            # 拖住退出。
+            hb_thread.join(timeout=5)
             # 优雅排空：先等正在跑的任务自然结束，超时后才杀。
             # 串行模式下 run_all 返回时任务已跑完，这里立刻通过；
             # 并行模式下停机信号退出时后台线程可能还在跑，先等
@@ -2900,10 +2953,12 @@ def main(argv=None):
     parser.add_argument("--worker-id", default=None,
                         help="多 worker 池身份（默认：主机名-进程号；"
                              "生产部署应配稳定 id，否则重启后旧行变 stale）")
-    parser.add_argument("--worker-capacity", type=int, default=1,
+    parser.add_argument("--worker-capacity", type=int, default=None,
                         help="worker 并发上限：同一 worker 内最多同时跑"
-                             "几个任务（默认 1，即旧串行行为；>1 时用"
-                             "线程池并行，认领仍是 CAS，成本写库持锁）")
+                             "几个任务（未给时读环境变量"
+                             " CLOUD_WORKER_CAPACITY，未设则默认 1，"
+                             "即旧串行行为；>1 时用线程池并行，认领"
+                             "仍是 CAS，成本写库持锁）")
     parser.add_argument("--host-name", default=None,
                         help="本 worker 所在主机名（HA 阶段 1）：认领任务"
                              "时写进 tasks.worktree_host，/pr 据此判断"